http://blog.csdn.net/canot/article/details/52040415
关于个人对Redis提供的Pub/Sub机制的认识在上一篇博客中涉及到了，也提到了关于如何避免Redis的Pub/Sub的一个最大的缺陷的思路—消息的持久化(http://blog.csdn.net/canot/article/details/51975566)。这篇文章主要是关于其思路(Redis的Pub/Sub的消息持久化)的代码实现:
Pub/Sub机制中最核心的Listener的实现:
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;
import redis.clients.util.RedisInputStream;
public class PubSubListener extends JedisPubSub {
private String clientId;
private HandlerRedis handlerRedis;
// 生成PubSubListener的时候必须制定一个id
public PubSubListener(String cliendId, Jedis jedis) {
this.clientId = cliendId;
jedis.auth("xxxx");
handlerRedis = new HandlerRedis(jedis);
}
@Override
public void onMessage(String channel, String message) {
if ("quit".equals(message)) {
this.unsubscribe(channel);
}
handlerRedis.handler(channel, message);
}
// 真正处理接受的地方
private void message(String channel, String message) {
System.out.println("message receive:" + message + ",channel:" + channel + "...");
}
@Override
public void onPMessage(String pattern, String channel, String message) {
// TODO Auto-generated method stub
}
@Override
public void onSubscribe(String channel, int subscribedChannels) {
// 将订阅者保存在一个"订阅活跃者集合中"
handlerRedis.subscribe(channel);
System.out.println("subscribe:" + channel);
}
@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
handlerRedis.ubsubscribe(channel);
System.out.println("unsubscribe:" + channel);
}
@Override
public void onPUnsubscribe(String pattern, int subscribedChannels) {
// TODO Auto-generated method stub
}
@Override
public void onPSubscribe(String pattern, int subscribedChannels) {
// TODO Auto-generated method stub
}
class HandlerRedis {
private Jedis jedis;
public HandlerRedis(Jedis jedis) {
this.jedis = jedis;
}
public void handler(String channel, String message) {
int index = message.indexOf("/");
if (index < 0) {
// 消息不合法,丢弃
return;
}
Long txid = Long.valueOf(message.substring(0, index));
String key = clientId + "/" + channel;
while (true) {
String lm = jedis.lindex(key, 0);// 获取第一个消息
if (lm == null) {
break;
}
int li = lm.indexOf("/");
// 如果消息不合法，删除并处理
if (li < 0) {
String result = jedis.lpop(key);// 删除当前message
// 为空
if (result == null) {
break;
}
message(channel, lm);
continue;
}
Long lxid = Long.valueOf(lm.substring(0, li));// 获取消息的txid
// 直接消费txid之前的残留消息
if (txid >= lxid) {
jedis.lpop(key);// 删除当前message
message(channel, lm);
continue;
} else {
break;
}
}
}
// 持久化订阅操作
public void subscribe(String channel) {
// 保证在订阅者集合中的格式为 唯一标识符/订阅的通道
String key = clientId + "/" + channel;
// 判断该客户端是否在集合中存在
boolean isExist = jedis.sismember("PERSIS_SUB", key);
if (!isExist) {
// 不存在则添加
jedis.sadd("PERSIS_SUB", key);
}
}
public void ubsubscribe(String channel) {
String key = clientId + "/" + channel;
// 从“活跃订阅者”集合中
jedis.srem("PERSIS_SUB", key);
// 删除“订阅者消息队列”
jedis.del(channel);
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
Listener中定义了一个内部类HandlerRedis。Listener类将onMessage以及onSubscribe两个方法交付于HandlerRedis。Handler处理这个方法的时候也即进行着队列的维护。Listener类中定义了一个message()方法，该方法是handler的回调方法，即真正的处理消息的地方。
通道的订阅客户端类:
import redis.clients.jedis.Jedis;
public class SubClient {
private Jedis jedis;
private PubSubListener listener;
public SubClient(String host,PubSubListener pubSubListener){
jedis = new Jedis(host);
jedis.auth("XXXXX");
this.listener = pubSubListener;
}
public void sub(String channel){
jedis.subscribe(listener, channel);
}
public void unsubscribe(String channel){
listener.unsubscribe(channel);
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
通道的消息发布的客户端:
import java.util.Set;
import redis.clients.jedis.Jedis;
public class PubClient {
private Jedis jedis;
public PubClient(String host) {
jedis = new Jedis(host);
jedis.auth("wx950709");
}
/**
* 发布的每条消息，都需要在“订阅者消息队列”中持久
*
* @param message
*/
public void put(String message) {
//获取所有活跃的消息接收者客户端 clientID/channel
Set<String> subClients = jedis.smembers("PERSIS_SUB");
for (String subs : subClients) {
// 保存每个客户端的消息
jedis.rpush(subs, message);
}
}
public void publish(String channel, String message) {
// 每个消息，都有具有一个全局唯一的id
// txid为了防止订阅端在数据处理时“乱序”，这就要求订阅者需要解析message
Long txid = jedis.incr("MESSAGE_TXID");
String content = txid + "/" + message;
this.put(content);
jedis.publish(channel, content);//为每个消息设定id，最终消息格式1000/messageContent
}
public void close(String channel){
jedis.publish(channel, "quit");
jedis.del(channel);//删除
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
测试引导类:
import redis.clients.jedis.Jedis;
public class Main {
public static void main(String[] args) throws Exception{
PubClient pubClient = new PubClient("127.0.0.1");
final String channel = "pubsub-channel222";
PubSubListener listener = new PubSubListener("client_one", new Jedis("127.0.0.1"));
SubClient subClient = new SubClient("127.0.0.1", listener);
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
//在API级别，此处为轮询操作，直到unsubscribe调用，才会返回
subClient.sub(channel);
}
});
t1.setDaemon(true);
t1.start();
int i = 0;
while(i < 2){
pubClient.publish(channel, "message"+i);
i++;
Thread.sleep(1000);
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
