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

雨点打透心脏的1/2处 2022-09-24 06:17 14阅读 0赞

关于个人对Redis提供的Pub/Sub机制的认识在上一篇博客中涉及到了，也提到了关于如何避免Redis的Pub/Sub的一个最大的缺陷的思路—消息的持久化([http://blog.csdn.net/canot/article/details/51975566][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);
            }
        }
    }

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

测试引导类:

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

[http_blog.csdn.net_canot_article_details_51975566]: http://blog.csdn.net/canot/article/details/51975566