Topic

Java RTopic object implements Publish / Subscribe mechanism based on Redis Pub/Sub or Valkey Pub/Sub. It allows to subscribe on events published with multiple instances of RTopic object with the same name.

Listeners are re-subscribed automatically after reconnection or failover. All messages sent during absence of connection are lost. Use Reliable Topic for reliable delivery.

Code example:

  1. RTopic topic = redisson.getTopic("myTopic");
  2. int listenerId = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, SomeObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RTopic topic = redisson.getTopic("myTopic");
  11. long clientsReceivedMessage = topic.publish(new SomeObject());

Code example of Async interface usage:

  1. RTopicAsync topic = redisson.getTopic("myTopic");
  2. RFuture<Integer> listenerFuture = topic.addListenerAsync(SomeObject.class, new MessageListener<SomeObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, SomeObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RTopicAsync topic = redisson.getTopic("myTopic");
  11. RFuture<Long> publishFuture = topic.publishAsync(new SomeObject());

Code example of Reactive interface usage:

  1. RedissonReactiveClient redisson = redissonClient.reactive();
  2. RTopicReactive topic = redisson.getTopic("myTopic");
  3. Mono<Integer> listenerMono = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  4.     @Override
  5.     public void onMessage(CharSequence channel, SomeObject message) {
  6.         //...
  7.     }
  8. });
  10. // in other thread or JVM
  11. RTopicReactive topic = redisson.getTopic("myTopic");
  12. Mono<Long> publishMono = topic.publish(new SomeObject());

Code example of RxJava3 interface usage:

  1. RedissonRxClient redisson = redissonClient.rxJava();
  2. RTopicRx topic = redisson.getTopic("myTopic");
  3. Single<Integer> listenerMono = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  4.     @Override
  5.     public void onMessage(CharSequence channel, SomeObject message) {
  6.         //...
  7.     }
  8. });
  10. // in other thread or JVM
  11. RTopicRx topic = redisson.getTopic("myTopic");
  12. Single<Long> publishMono = topic.publish(new SomeObject());

Partitioning

This feature is available only in Redisson PRO edition.

Although each Topic instance is cluster-compatible, it can be connected only to a single Redis or Valkey node which owns the topic name. That may cause the following issues:

  • CPU overload on a single node.
  • Overload of network or data traffic to a single node.
  • Full interruption of the message flow during failover.

Topic partitioning addresses these challenges by enabling connections to all nodes in cluster and distributing messages effectively. It brings the following benefits:

  • Increases throughput of the topic.
  • Minimizes interruptions during failover.
  • Lowers CPU and network load on Valkey or Redis nodes.
  • Scales the message flow to multiple Valkey or Redis nodes.

Partitions amount is defined through the global topicSlots setting or per instance through ClusteredTopicOptions.slots() setting, which overrides the global setting.

Slots definition per instance:

  1. RClusteredTopic topic = redisson.getClusteredTopic(ClusteredTopicOptions.name("myTopic").slots(15));

Usage example:

  1. RClusteredTopic topic = redisson.getClusteredTopic("myTopic");
  2. int listenerId = topic.addListener(MyObject.class, new MessageListener<MyObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, MyObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RClusteredTopic topic = redisson.getClusteredTopic("myTopic");
  11. long clientsReceivedMessage = topic.publish(new MyObject());

Topic pattern

Java implementation of Redis or Valkey based RPatternTopic object. It allows to subscribe to multiple topics by specified glob-style pattern.

Listeners are re-subscribed automatically after reconnection to a server or failover.

Pattern examples:

  • topic? subscribes to topic1, topicA …
  • topic?_my subscribes to topic_my, topic123_my, topicTEST_my …
  • topic[ae] subscribes to topica and topice only

Code example:

  1. // subscribe to all topics by `topic*` pattern
  2. RPatternTopic patternTopic = redisson.getPatternTopic("topic*");
  3. int listenerId = patternTopic.addListener(Message.class, new PatternMessageListener<Message>() {
  4.     @Override
  5.     public void onMessage(CharSequence pattern, CharSequence channel, Message msg) {
  6.         //...
  7.     }
  8. });

Code example of Async interface usage:

  1. RPatternTopicAsync patternTopic = redisson.getPatternTopic("topic*");
  2. RFuture<Integer> listenerFuture = patternTopic.addListenerAsync(Message.class, new PatternMessageListener<Message>() {
  3.     @Override
  4.     public void onMessage(CharSequence pattern, CharSequence channel, Message msg) {
  5.         //...
  6.     }
  7. });

Code example of Reactive interface usage:

  1. RedissonReactiveClient redisson = redissonClient.reactive();
  2. RTopicReactive patternTopic = redisson.getPatternTopic("topic*");
  3. Mono<Integer> listenerMono = patternTopic.addListener(Message.class, new PatternMessageListener<Message>() {
  4.     @Override
  5.     public void onMessage(CharSequence pattern, CharSequence channel, Message msg) {
  6.         //...
  7.     }
  8. });

Code example of RxJava3 interface usage:

  1. RedissonRxClient redisson = redissonClient.rxJava();
  2. RTopicRx patternTopic = redisson.getPatternTopic("topic*");
  3. Single<Integer> listenerSingle = patternTopic.addListener(Message.class, new PatternMessageListener<Message>() {
  4.     @Override
  5.     public void onMessage(CharSequence pattern, CharSequence channel, Message msg) {
  6.         //...
  7.     }
  8. });

Sharded topic

Java implementation of Redis or Valkey based RShardedTopic object implements Sharded Publish / Subscribe mechanism. It allows to subscribe on events published with multiple instances of RShardedTopic object with the same name. Subscribe/publish operations are executed only on Redis or Valkey node in Cluster which is bounded to specific topic name. Published messages via RShardedTopic aren’t broadcasted across all nodes as for RTopic object. Which reduces network bandwidth usage between Redis and Valkey nodes and their CPU load, as well as their CPU load.

Listeners are re-subscribed automatically after reconnection to a server or failover. All messages sent during absence of connection are lost. Use Reliable Topic for reliable delivery.

Code example:

  1. RShardedTopic topic = redisson.getShardedTopic("myTopic");
  2. int listenerId = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, SomeObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RShardedTopic topic = redisson.getShardedTopic("myTopic");
  11. long clientsReceivedMessage = topic.publish(new SomeObject());

Code example of Async interface usage:

  1. RShardedTopicAsync topic = redisson.getShardedTopic("myTopic");
  2. RFuture<Integer> listenerFuture = topic.addListenerAsync(SomeObject.class, new MessageListener<SomeObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, SomeObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RShardedTopicAsync topic = redisson.getShardedTopic("myTopic");
  11. RFuture<Long> publishFuture = topic.publishAsync(new SomeObject());

Code example of Reactive interface usage:

  1. RedissonReactiveClient redisson = redissonClient.reactive();
  2. RShardedTopicReactive topic = redisson.getShardedTopic("myTopic");
  3. Mono<Integer> listenerMono = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  4.     @Override
  5.     public void onMessage(CharSequence channel, SomeObject message) {
  6.         //...
  7.     }
  8. });
  10. // in other thread or JVM
  11. RShardedTopicReactive topic = redisson.getShardedTopic("myTopic");
  12. Mono<Long> publishMono = topic.publish(new SomeObject());

Code example of RxJava3 interface usage:

  1. RedissonRxClient redisson = redissonClient.rxJava();
  2. RShardedTopicRx topic = redisson.getShardedTopic("myTopic");
  3. Single<Integer> listenerMono = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  4.     @Override
  5.     public void onMessage(CharSequence channel, SomeObject message) {
  6.         //...
  7.     }
  8. });
  10. // in other thread or JVM
  11. RShardedTopicRx topic = redisson.getShardedTopic("myTopic");
  12. Single<Long> publishMono = topic.publish(new SomeObject());

Partitioning

This feature is available only in Redisson PRO edition.

Although each ShardedTopic instance is cluster-compatible, it can be connected only to a single Redis or Valkey node which owns the topic name. That may cause the following issues:

  • CPU overload on a single node.
  • Overload of network or data traffic to a single node.
  • Full interruption of the message flow during failover.

ShardedTopic partitioning addresses these challenges by enabling connections to all nodes in cluster and distributing messages effectively. It brings the following benefits:

  • Increases throughput of the topic.
  • Minimizes interruptions during failover.
  • Lowers CPU and network load on Valkey or Redis nodes.
  • Scales the message flow to multiple Valkey or Redis nodes.

Partitions amount is defined through the global topicSlots setting or per instance through ClusteredTopicOptions.slots() setting, which overrides the global setting.

Slots definition per instance:

  1. RClusteredTopic topic = redisson.getClusteredTopic(ClusteredTopicOptions.name("myTopic").slots(15));

Usage example:

  1. RClusteredTopic topic = redisson.getClusteredTopic("myTopic");
  2. int listenerId = topic.addListener(MyObject.class, new MessageListener<MyObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, MyObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RClusteredTopic topic = redisson.getClusteredTopic("myTopic");
  11. long clientsReceivedMessage = topic.publish(new MyObject());

Reliable Topic

Java implementation of Redis or Valkey based RReliableTopic object implements Publish / Subscribe mechanism with reliable delivery of messages. In case of Redis or Valkey connection interruption all missed messages are delivered after reconnection to Redis. Message considered as delivered when it was received by Redisson and submited for processing by topic listeners.

Each RReliableTopic object instance (subscriber) has own watchdog which is started when the first listener was registered. Subscriber expires after org.redisson.config.Config#reliableTopicWatchdogTimeout timeout if watchdog didn’t extend it to the next timeout time interval. This prevents against infinity grow of stored messages in topic due to Redisson client crash or any other reason when subscriber unable to consume messages.

Topic listeners are resubscribed automatically after reconnection to a server or failover.

Code example:

  1. RReliableTopic topic = redisson.getReliableTopic("anyTopic");
  2. topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, SomeObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RReliableTopic topic = redisson.getReliableTopic("anyTopic");
  11. long subscribersReceivedMessage = topic.publish(new SomeObject());

Code example of Async interface usage:

  1. RReliableTopicAsync topic = redisson.getReliableTopic("anyTopic");
  2. RFuture<String> listenerFuture = topic.addListenerAsync(SomeObject.class, new MessageListener<SomeObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, SomeObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RReliableTopicAsync topic = redisson.getReliableTopic("anyTopic");
  11. RFuture<Long> future = topic.publishAsync(new SomeObject());

Code example of Reactive interface usage:

  1. RedissonReactiveClient redisson = redissonClient.reactive();
  3. RReliableTopicReactive topic = redisson.getReliableTopic("anyTopic");
  4. Mono<String> listenerMono = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  5.     @Override
  6.     public void onMessage(CharSequence channel, SomeObject message) {
  7.         //...
  8.     }
  9. });
  11. // in other thread or JVM
  12. RReliableTopicReactive topic = redisson.getReliableTopic("anyTopic");
  13. Mono<Long> publishMono = topic.publish(new SomeObject());

Code example of RxJava3 interface usage:

  1. RedissonRxClient redisson = redissonClient.rxJava();
  3. RReliableTopicRx topic = redisson.getReliableTopic("anyTopic");
  4. Single<String> listenerRx = topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
  5.     @Override
  6.     public void onMessage(CharSequence channel, SomeObject message) {
  7.         //...
  8.     }
  9. });
  11. // in other thread or JVM
  12. RReliableTopicRx topic = redisson.getReliableTopic("anyTopic");
  13. Single<Long> publisRx = topic.publish(new SomeObject());

Partitioning

This feature is available only in Redisson PRO edition.

Although each ReliableTopic instance is cluster-compatible, it can be connected only to a single Redis or Valkey node which owns the topic name. That may cause the following issues:

  • CPU overload on a single node.
  • Overload of network or data traffic to a single node.
  • Full interruption of the message flow during failover.

ReliableTopic partitioning addresses these challenges by enabling connections to all nodes in cluster and distributing messages effectively. It brings the following benefits:

  • Increases throughput of the topic.
  • Minimizes interruptions during failover.
  • Lowers CPU and network load on Valkey or Redis nodes.
  • Scales the message flow to multiple Valkey or Redis nodes.

Partitions amount is defined through the global topicSlots setting or per instance through ClusteredTopicOptions.slots() setting, which overrides the global setting.

Slots definition per instance:

  1. RClusteredReliableTopic topic 
  2.     = redisson.getClusteredReliableTopic(ClusteredTopicOptions.name("myTopic").slots(15));

Usage example:

  1. RClusteredReliableTopic topic = redisson.getClusteredReliableTopic("myTopic");
  2. int listenerId = topic.addListener(MyObject.class, new MessageListener<MyObject>() {
  3.     @Override
  4.     public void onMessage(CharSequence channel, MyObject message) {
  5.         //...
  6.     }
  7. });
  9. // in other thread or JVM
  10. RClusteredReliableTopic topic = redisson.getClusteredReliableTopic("myTopic");
  11. long clientsReceivedMessage = topic.publish(new MyObject());