【转】使用Redis的Pub/Sub来实现类似于JMS的消息持久化-白红宇

【转】使用Redis的Pub/Sub来实现类似于JMS的消息持久化

阅读量：6083 次

发布时间：2019-06-20

本文共 5531 字，大约阅读时间需要 18 分钟。

http://blog.csdn.net/canot/article/details/52040415

关于个人对Redis提供的Pub/Sub机制的认识在上一篇博客中涉及到了，也提到了关于如何避免Redis的Pub/Sub的一个最大的缺陷的思路—消息的持久化()。这篇文章主要是关于其思路(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); } } }

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); } }

通道的消息发布的客户端:

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
      
        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);//删除 } }

测试引导类:

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); } } }

转载于:https://www.cnblogs.com/mimime/p/6203615.html

你可能感兴趣的文章