Posted onSep 20, 2022 • Originally published atjsedano.dev onSep 19, 2022

Java Redis pub/sub example with Jedis

#java #redis

This is a Redis pub/sub example written in Java using Jedis.

You can find the complete code for this here:Java Redis pub/sub example.
For more neat stuff that Redis can do look here:Redis 2022.

Prerequisites

Download the code
Have an instance ofredis-stack running.
Apache Maven 3.8.6
Java 17

Libraries used

We are only usingjedis in order to keep the example simple.

Redis Commands used

this is used in order to subscribe to one or more channels:

subscribe channel1 channel2... channelN

unsubscribe

unsubscribe current client to one or more channels:

unsubscribe channel1 channel2... channelN

psubscribe

instead of subscribing to a channel, this allows you to subscribe to one of more patterns:

psubscribe pattern1 pattern2... patternN

for example if you subscribe tofoo* it means that you will get the messages meant for all the channels that start withfoo.

punsubscribe

unsubscribe from one or more patterns:

punsubscribe pattern1 pattern2... patternN

publish

publish sends a message to connected clients, returns the number of clients that got the message.

publish aChannel 'some message'

ping

sends a ping from a client to the server, optionally you can send a message and the ping will echo it.

ping

ping a message

Code walkthrough

We need to extendJedisPubSub in order to give our client functionality.

import redis.clients.jedis.JedisPubSub;public class LogPubSub extends JedisPubSub {

To keep this example simple we are only going to add very basic functionality, in order to know which client is the one that got the message we are adding a name field.

  private String name;  public LogPubSub(String name) {    this.name = name;  }

Forsubscribe,unsubscribe,psubscribe,punsubscribe andpong (which is the one that a successful ping triggers) we only print the information:

  public void onSubscribe(String channel, int subscribedChannels) {    System.out.printf(        "name: %s method: %s channel: %s subscribedChannels: %d\n",        name, "onSubscribe", channel, subscribedChannels);  }  public void onUnsubscribe(String channel, int subscribedChannels) {    System.out.printf(        "name: %s method: %s channel: %s subscribedChannels: %d\n",        name, "onUnsubscribe", channel, subscribedChannels);  }  public void onPUnsubscribe(String pattern, int subscribedChannels) {    System.out.printf(        "name: %s method: %s patten: %s subscribedChannels: %d\n",        name, "onPUnsubscribe", pattern, subscribedChannels);  }  public void onPSubscribe(String pattern, int subscribedChannels) {    System.out.printf(        "name: %s method: %s patten: %s subscribedChannels: %d\n",        name, "onPSubscribe", pattern, subscribedChannels);  }  public void onPong(String message) {    System.out.printf("name: %s method: %s message: %s\n", name, "onPong", message);  }}

When receiving a message in the client apart from printing the information if the message is a ping then we will do aping. If receiving the String “exit” we will unsubscribe from that channel or pattern (in the case of client subscribed using psubscribe).

  public void onMessage(String channel, String message) {    System.out.printf(        "name: %s method: %s channel: %s message: %s\n", name, "onMessage", channel, message);    switch (message) {      case "ping":        this.ping();        break;      case "exit":        this.unsubscribe(channel);        break;      default:        if (message.indexOf("ping") == 0 && message.indexOf(" ") == 4) {          this.ping(message.substring(5));        }    }  }  public void onPMessage(String pattern, String channel, String message) {    System.out.printf(        "name: %s method: %s pattern: %s channel: %s message: %s\n",        name, "onPMessage", pattern, channel, message);    switch (message) {      case "ping":        this.ping();        break;      case "exit":        this.punsubscribe(pattern);        break;      default:        if (message.indexOf("ping") == 0 && message.indexOf(" ") == 4) {          this.ping(message.substring(5));        }    }  }

Then we only need to make aJedisPooled connection and used it to subscribe to a channel or pattern:

JedisPooled jedis = new JedisPooled("localhost", 6379);ExecutorService executor = Executors.newFixedThreadPool(4);executor.execute(() -> jedis.subscribe(new LogPubSub("onlyOne"), "dev.one"));executor.execute(() -> jedis.subscribe(new LogPubSub("oneAndTwo"), "dev.one", "dev.two"));executor.execute(() -> jedis.psubscribe(new LogPubSub("pattern"), "dev.*"));

We need to run the subscribe and psubscribe on a thread since those are blocking operations.

In this example we are creating three clients, which we will identify as “onlyOne”, “oneAndTwo” and “pattern”, this doesn’t mean anything to redis, but it will be easier for us to keep track of whats happening.

“onlyOne” is subscribed to channel “dev.one”.
“oneAndTwo” is subscribed to channels “dev.one” and “dev.two”.
“pattern” is subscribed to the pattern “dev.*”.

With this we could send messages using publish on ourRedisInsight CLI which should be running onhttp://localhost:8001/redis-stack/browser if we installed theredis-stack.

But we will also send messages using jedis.

String message = "";try (BufferedReader br = new BufferedReader(new InputStreamReader(System.in))) {  do {    message = br.readLine();    int firstSpace = message.indexOf(' ');    if (firstSpace > 1) {      jedis.publish(message.substring(0, firstSpace), message.substring(firstSpace + 1));    }  } while (!"close".equals(message));  jedis.close();  System.exit(0);} catch (IOException e) {  throw new RuntimeException(e);}

For building the code runmvn package and a far jar will be generated on target and then you can run it like sojava -jar RedisPubSub-1.0.jar.

Example run:

... target % java -jar RedisPubSub-1.0.jarSLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".SLF4J: Defaulting to no-operation (NOP) logger implementationSLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.name: pattern method: onPSubscribe patten: dev.* subscribedChannels: 1name: oneAndTwo method: onSubscribe channel: dev.one subscribedChannels: 1name: oneAndTwo method: onSubscribe channel: dev.two subscribedChannels: 2dev.one helloname: pattern method: onPMessage pattern: dev.* channel: dev.one message: helloname: oneAndTwo method: onMessage channel: dev.one message: hellodev.five hello to youname: pattern method: onPMessage pattern: dev.* channel: dev.five message: hello to youdev.two also to youname: pattern method: onPMessage pattern: dev.* channel: dev.two message: also to youname: oneAndTwo method: onMessage channel: dev.two message: also to youdev.six exitname: pattern method: onPMessage pattern: dev.* channel: dev.six message: exitname: pattern method: onPUnsubscribe patten: dev.* subscribedChannels: 0dev.two exitname: oneAndTwo method: onMessage channel: dev.two message: exitname: oneAndTwo method: onUnsubscribe channel: dev.two subscribedChannels: 1dev.one ping say my namename: oneAndTwo method: onMessage channel: dev.one message: ping say my namename: oneAndTwo method: onPong message: say my namedev.two are you alive?dev.five no one is listening heredev.one this one is alivename: oneAndTwo method: onMessage channel: dev.one message: this one is alivedev.one exitname: oneAndTwo method: onMessage channel: dev.one message: exitname: oneAndTwo method: onUnsubscribe channel: dev.one subscribedChannels: 0dev.one not anymoreclose... target %

Download the complete code for this here:Java Redis pub/sub example.