【Redis】jedis客户端实现redis消息的发布订阅（实时消息中间件）

WindySky 2017-10-17

展开全文

发布

[java] view plain copy

print?

package com.chiwei.redis;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
public class RedisPublisher {
private static final Logger log = LoggerFactory.getLogger(RedisPublisher.class);
private final Jedis pubJedis;
private final String[] channel;
public RedisPublisher(Jedis pubJedis, String[] channel) {
this.pubJedis = pubJedis;
this.channel = channel;
}
public void start() {
log.debug("Type your message (type quit to exit)");
int channelLen = channel.length;
try {
BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
while (true) {
String line = br.readLine();
if (!"quit".equals(line)) {
for (int i = 0; i < channelLen; i++) {
if (channel[i].matches("^chiwei.*")) {
log.debug("Match...");
pubJedis.publish(channel[i], line + "haha");
} else {
pubJedis.publish(channel[i], line);
}
log.debug("Publish to {}", channel[i]);
}
} else {
break;
}
}
} catch (Exception e) {
log.error("IO fail while reading input", e);
}
}
}

以上发布类，发布的频道是一个数组，即同时将内容发布到多个频道中，你可以根据内容去判断，不同的内容发布到不同的频道中。

package com.chiwei.redis;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.JedisPubSub;
public class RedisSubscriber extends JedisPubSub{
private static final Logger log = LoggerFactory.getLogger(RedisSubscriber.class);
//取得订阅的消息后的处理
public void onMessage(String s, String s1) {
// TODO Auto-generated method stub
log.debug("Message received,Channel:{},Msg:{}",s,s1);
}
//取得按表达式的方式订阅的消息后的处理
public void onPMessage(String s, String s1, String s2) {
// TODO Auto-generated method stub
log.debug("Pattern:{}",s);
log.debug("Pattern Message received,Channel:{},Msg:{}",s1,s2);
}
//初始化按表达式的方式订阅时候的处理
public void onPSubscribe(String s, int i) {
// TODO Auto-generated method stub
log.debug("Pattern Subscribe,Pattern:{},ChannelNum:{}",s,i);
}
//取消按表达式的方式订阅时候的处理
public void onPUnsubscribe(String s, int i) {
// TODO Auto-generated method stub
log.debug("Pattern Unsubscribe,Pattern:{},ChannelNum:{}",s,i);
}
//初始化订阅时候的处理
public void onSubscribe(String s, int i) {
// TODO Auto-generated method stub
log.debug("Subscribe,Channel:{},ChannelNum:{}",s,i);
}
//取消订阅时候的处理
public void onUnsubscribe(String s, int i) {
// TODO Auto-generated method stub
log.debug("Unsubscribe,Channel:{},ChannelNum:{}",s,i);
}
}

该类就是订阅的实现类，对于订阅的各项操作实现具体的处理方法。

启动主类

[java] view plain copy

print?

package com.chiwei.redis;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
public class RedisPubSubMain {
public static final String[] CHANNEL_NAME = new String[] { "chiwei.momo", "chiwei.nono","taotao"};
private static final Logger log = LoggerFactory.getLogger(RedisPubSubMain.class);
public static void main(String[] args) {
// TODO Auto-generated method stub
log.debug("=========================");
JedisPoolConfig config = new JedisPoolConfig();
config = new JedisPoolConfig();
config.setMaxTotal(100);
config.setMaxIdle(10);
config.setMaxWaitMillis(1000L);
config.setTestOnBorrow(true);
config.setTestOnReturn(true);
JedisPool jedisPool = new JedisPool(config, "192.168.11.176", 7379);
final Jedis subJedis = jedisPool.getResource();
final RedisSubscriber sub = new RedisSubscriber();
new Thread(new Runnable() {
public void run() {
try {
//subJedis.subscribe(sub, CHANNEL_NAME);
subJedis.psubscribe(sub, "^chiwei.*");
log.debug("Subscribe ended");
} catch (Exception e) {
log.error("Subscribe failed", e);
}
}
}).start();
Jedis pubJedis = jedisPool.getResource();
new RedisPublisher(pubJedis, CHANNEL_NAME).start();
sub.unsubscribe();
jedisPool.returnResourceObject(subJedis);
jedisPool.returnResourceObject(pubJedis);
jedisPool.close();
}
}

subJedis.psubscribe(sub, "^chiwei.*");按照正则匹配订阅的频道

由于订阅类会阻塞当前线程的执行，所以在main线程中另起一个线程来启动订阅，然后启动发布线程去发布内容。

2015-04-17 10:35:43,751 - com.chiwei.redis.RedisPubSubMain[18] -0 [main] DEBUG - =========================
2015-04-17 10:35:43,843 - com.chiwei.redis.RedisSubscriber[29] -92 [Thread-3] DEBUG - Pattern Subscribe,Pattern:^chiwei.*,ChannelNum:1
2015-04-17 10:35:43,848 - com.chiwei.redis.RedisPublisher[25] -97 [main] DEBUG - Type your message (type quit to exit)
3
2015-04-17 10:35:53,132 - com.chiwei.redis.RedisPublisher[34] -9381 [main] DEBUG - Match...
2015-04-17 10:35:53,138 - com.chiwei.redis.RedisPublisher[39] -9387 [main] DEBUG - Publish to chiwei.momo
2015-04-17 10:35:53,140 - com.chiwei.redis.RedisPublisher[34] -9389 [main] DEBUG - Match...
2015-04-17 10:35:53,146 - com.chiwei.redis.RedisPublisher[39] -9395 [main] DEBUG - Publish to chiwei.nono
2015-04-17 10:35:53,153 - com.chiwei.redis.RedisPublisher[39] -9402 [main] DEBUG - Publish to taotao