How CoSky Uses Redis Pub/Sub for Real-Time Configuration and Service Updates

CoSky leverages Redis Pub/Sub channels to broadcast configuration changes and service topology updates across distributed nodes, enabling millisecond-level propagation without polling.

CoSky is a lightweight service registry and configuration center built on Redis. This article examines how the ahoo-wang/cosky repository implements real-time event propagation using Redis Pub/Sub for both configuration management and service discovery, analyzing the concrete Kotlin implementation in the source files.

Architecture Overview

CoSky uses Redis as the backbone for cluster communication. The architecture separates concerns into two distinct Pub/Sub channels: one for configuration changes and one for service registry updates.

Channel Naming Conventions

CoSky generates channel names using dedicated key generators:

• Configuration changes: cosky:config:{namespace}:{configId} — generated by ConfigKeyGenerator
• Service topology changes: cosky:service:{namespace}:idx — generated by DiscoveryKeyGenerator

Event Flow

The flow follows a standard publish-subscribe pattern:

1. Publishers (RedisConfigService, RedisServiceRegistry) call redisTemplate.convertAndSend(channel, payload) after persisting state changes
2. Listeners (RedisConfigEventListenerContainer, RedisServiceEventListenerContainer) wrap Spring's ReactiveRedisMessageListenerContainer and expose reactive Flux<T> streams
3. Subscribers (such as CoSkyConfigRefresher and CoSkyDiscoveryClient) consume these streams to update in-memory caches instantly

Real-Time Configuration Updates

Configuration changes propagate through dedicated per-config channels, allowing granular subscription to specific configuration keys.

Publishing Configuration Changes

In cosky-config/src/main/kotlin/me/ahoo/cosky/config/redis/RedisConfigService.kt, the setConfig method persists data and immediately broadcasts the change:

override fun setConfig(namespace: String, configId: String, config: String): Mono<Void> {
    val key = ConfigKeyGenerator.getConfigKey(namespace, configId)
    // Store the raw config data
    return redisTemplate.opsForValue().set(key, config)
        // After persisting, broadcast the change
        .then(Mono.fromFuture {
            redisTemplate.convertAndSend(key, ConfigChangedEvent(namespace, configId, config))
        })
}

The ConfigKeyGenerator ensures the channel name matches the storage key, creating a direct correlation between the data location and the notification channel.

Subscribing to Config Events

The RedisConfigEventListenerContainer in cosky-config/src/main/kotlin/me/ahoo/cosky/config/redis/RedisConfigEventListenerContainer.kt adapts Spring's reactive Redis listener to CoSky's domain events:

override fun receiveEvent(topic: NamespacedConfigId): Flux<ConfigChangedEvent> {
    val topicStr = ConfigKeyGenerator.getConfigKey(topic.namespace, topic.configId)
    return delegate.receive(ChannelTopic.of(topicStr))
        .map { asEvent(it) }  // Convert raw Redis message to domain event
}

The delegate is a ReactiveRedisMessageListenerContainer. The method returns a Flux<ConfigChangedEvent> that downstream components subscribe to for real-time updates.

Service Discovery and Registry Events

Service topology changes use a single channel per namespace rather than per-service channels, optimizing for the discovery pattern where clients typically need to know when any service in a namespace changes.

Publishing Service Changes

In cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/redis/RedisServiceRegistry.kt, registration and deregistration operations publish to the service index channel:

override fun register(serviceInstance: ServiceInstance): Mono<Void> {
    val idxKey = DiscoveryKeyGenerator.getServiceIdxKey(serviceInstance.namespace)
    // Persist instance data (hash operations omitted for brevity)
    return Mono.fromFuture {
        // Notify all listeners that the topology of this namespace changed
        redisTemplate.convertAndSend(idxKey, serviceInstance.namespace)
    }
}

The payload is simply the namespace string. Listeners use this signal to fetch the updated service list from Redis.

Listening for Topology Updates

The RedisServiceEventListenerContainer in cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/redis/RedisServiceEventListenerContainer.kt handles these notifications:

override fun receiveEvent(topic: String): Flux<String> {
    val serviceIdxKey = DiscoveryKeyGenerator.getServiceIdxKey(topic)
    return delegate.receive(ChannelTopic.of(serviceIdxKey))
        .map { getNamespaceOfKey(it.channel) }  // Extract namespace from channel name
}

RedisServiceDiscovery subscribes to this Flux<String> and updates its in-memory ServiceTopology cache whenever a namespace event arrives.

Spring Cloud Integration

CoSky provides Spring Cloud starters that auto-configure these reactive components. In cosky-spring-cloud-starter-config/src/main/kotlin/me/ahoo/cosky/config/spring/cloud/CoSkyConfigAutoConfiguration.kt, the configuration wires the publisher and listener:

@Bean
fun redisConfigService(
    redisTemplate: ReactiveStringRedisTemplate,
    eventListener: RedisConfigEventListenerContainer
): ConfigService = RedisConfigService(redisTemplate, eventListener)

@Bean
fun configRefresher(
    configService: ConfigService,
    applicationContext: ConfigurableApplicationContext
): CoSkyConfigRefresher = CoSkyConfigRefresher(configService, applicationContext)

The CoSkyConfigRefresher subscribes to the Flux<ConfigChangedEvent> and triggers EnvironmentChangeEvent in the Spring context, causing @RefreshScope beans to reload instantly without application restart.

Similarly, CoSkyDiscoveryAutoConfiguration in the discovery starter sets up RedisServiceRegistry, RedisServiceDiscovery, and the event listener container for service discovery clients.

Summary

• Redis Pub/Sub serves as the real-time messaging backbone for CoSky's distributed configuration and service registry.
• Configuration changes use dedicated per-config channels (cosky:config:{namespace}:{configId}) published by RedisConfigService and consumed by RedisConfigEventListenerContainer.
• Service topology changes use namespace-scoped channels (cosky:service:{namespace}:idx) published by RedisServiceRegistry and consumed by RedisServiceEventListenerContainer.
• Reactive streams (Flux<T>) connect publishers to subscribers, enabling non-blocking, millisecond-level propagation of changes across the cluster.
• Spring Cloud integration provides auto-configuration that wires these components into @RefreshScope refreshers and discovery clients.

Frequently Asked Questions

How does CoSky ensure real-time updates without polling?

CoSky uses Redis Pub/Sub channels as a push-based messaging layer. When configuration or service data changes, the publisher (RedisConfigService or RedisServiceRegistry) immediately calls redisTemplate.convertAndSend() to broadcast the event. Subscribers (RedisConfigEventListenerContainer and RedisServiceEventListenerContainer) listen via ReactiveRedisMessageListenerContainer and emit reactive Flux streams, delivering updates to application components within milliseconds of the change occurring.

What is the difference between configuration and service discovery channels in CoSky?

Configuration channels are granular and specific: cosky:config:{namespace}:{configId} creates a unique Pub/Sub channel for every individual configuration key, allowing clients to subscribe only to changes they care about. Service discovery channels are namespace-scoped: cosky:service:{namespace}:idx uses a single channel per namespace to signal that any service topology in that namespace has changed, optimizing for the typical discovery pattern where clients need to refresh the entire service list for a namespace.

How does CoSky integrate with Spring Cloud to refresh configuration beans?

The CoSkyConfigAutoConfiguration class creates a CoSkyConfigRefresher bean that subscribes to the Flux<ConfigChangedEvent> from RedisConfigEventListenerContainer. When a configuration change event arrives, the refresher publishes a Spring EnvironmentChangeEvent to the application context. Beans annotated with @RefreshScope automatically reload their properties in response to this event, achieving real-time configuration updates without requiring application restarts.

What are the key source files for understanding CoSky's Redis Pub/Sub implementation?

The core implementation resides in four main files: RedisConfigService.kt handles configuration publishing, RedisConfigEventListenerContainer.kt handles configuration subscription, RedisServiceRegistry.kt publishes service changes, and RedisServiceEventListenerContainer.kt subscribes to service topology updates. Auto-configuration wiring is found in CoSkyConfigAutoConfiguration.kt and CoSkyDiscoveryAutoConfiguration.kt within the Spring Cloud starter modules.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}