Skip to content

Publish / Subscribe

The included RedisPubSubService uses Redis' SUBSCRIBE/PUBLISH commands to provide a messaging service, which allows threads to communicate across JVM instances.

It multiplexes many (thousands of) threads/clients and their (virtual) channels onto a single channel on Redis. This is more scalable than letting each thread/client start its own dedicated subscription-connection to Redis.

It can for example be used to distribute events among JVMs/nodes and route those to their respective WebSocket sessions.

Things to be aware of:

  • Delivery is at-most-once: messages published while a node is disconnected are lost.
  • Each subscription buffers its incoming messages and handles them in order on its own virtual thread. Message order is preserved per publisher, not across publishers.
  • If the message handler is too slow and the buffer (messageBufferLimit, default 10) is full, new messages are dropped and a warning is logged.
  • By default, a subscription also receives the messages it publishes itself (see echoSelf below).
  • Messages are held in memory (ByteArray) and shared between subscribers, so do not modify them in a handler.

Set up a shared RedisPubSubService

// share this as a bean/singleton. It handles re-connect upon failures internally.
val redisPubSubService = RedisPubSubService(
    connectionPool = connectionPool,
).apply { start() } // waits until subscribed on Redis (up to connectTimeoutMs of the pool by default)

// on shutdown
redisPubSubService.stop()

Optional parameters:

Parameter Default Description
redisChannelName the class name The channel used on Redis. Services which shall communicate must use the same name.
delayBetweenReconnectMs 1000 (milliseconds) Delay before reconnecting after a connection failure.
heartbeatIntervalMs 15000 (milliseconds) If nothing is received for this long, a PING is sent. If no reply arrives within another interval, the connection is considered dead and re-established.

Start listening for messages/events

Start listening on the channels "foo" and "bar": 

val subscription = redisPubSubService.subscribe(listOf("foo", "bar")) { channel: String, msg: ByteArray ->
    println("$channel: ${msg.toString(redisStringCharset)}")
}

Send messages

Send a message to the channel "foo": 

subscription.publish(listOf("foo"), "test".toByteArray(redisStringCharset))

The message is delivered to all subscriptions of these channels, on all JVMs. Pass echoSelf = false to not deliver it to the publishing subscription itself. A subscription created with an empty channel list can be used for publishing only.

Stop listening

subscription.close()