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 of redis-stack running.
- Apache Maven 3.8.6
- Java 17
Libraries used
We are only using jedis in order to keep the example simple.
Redis Commands used
subscribe
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 to foo*
it means that you will get the messages meant for all the channels that start with foo
.
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
or
ping a message
Code walkthrough
We need to extend JedisPubSub
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;
}
For subscribe
, unsubscribe
, psubscribe
, punsubscribe
and pong
(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 a ping
. 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 a JedisPooled
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 our RedisInsight CLI which should be running on http://localhost:8001/redis-stack/browser if we installed the redis-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 run mvn package
and a far jar will be generated on target and then you can run it like so java -jar RedisPubSub-1.0.jar
.
Example run:
... target % java -jar RedisPubSub-1.0.jar
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
name: pattern method: onPSubscribe patten: dev.* subscribedChannels: 1
name: oneAndTwo method: onSubscribe channel: dev.one subscribedChannels: 1
name: oneAndTwo method: onSubscribe channel: dev.two subscribedChannels: 2
dev.one hello
name: pattern method: onPMessage pattern: dev.* channel: dev.one message: hello
name: oneAndTwo method: onMessage channel: dev.one message: hello
dev.five hello to you
name: pattern method: onPMessage pattern: dev.* channel: dev.five message: hello to you
dev.two also to you
name: pattern method: onPMessage pattern: dev.* channel: dev.two message: also to you
name: oneAndTwo method: onMessage channel: dev.two message: also to you
dev.six exit
name: pattern method: onPMessage pattern: dev.* channel: dev.six message: exit
name: pattern method: onPUnsubscribe patten: dev.* subscribedChannels: 0
dev.two exit
name: oneAndTwo method: onMessage channel: dev.two message: exit
name: oneAndTwo method: onUnsubscribe channel: dev.two subscribedChannels: 1
dev.one ping say my name
name: oneAndTwo method: onMessage channel: dev.one message: ping say my name
name: oneAndTwo method: onPong message: say my name
dev.two are you alive?
dev.five no one is listening here
dev.one this one is alive
name: oneAndTwo method: onMessage channel: dev.one message: this one is alive
dev.one exit
name: oneAndTwo method: onMessage channel: dev.one message: exit
name: oneAndTwo method: onUnsubscribe channel: dev.one subscribedChannels: 0
dev.one not anymore
close
... target %
Download the complete code for this here: Java Redis pub/sub example.
Top comments (0)