Redis Demo系列之(六)消息队列 pub/sub

小灰灰 2022-05-18 00:11 176阅读 0赞

### 前言 ###

redis设计的初衷并不是为了消息队列而设计的，但是有太多的人将Redis作为消息队列而使用。Redis消息队列时，当Redis宕机后，消息会丢失。如果收消息方未有重发和验证机制，Redis内的数据会出现丢失。所以，使用Redis的作为消息队列，通常是对于消息的准确性并非特别高的场景。当需要对数据非常敏感以及准确性较高的情况可以使用`Kafka`、`RabbitMQ`等专门等消息队列。但是，通常，这会增加系统的开发成本和维护成本。

本文相关代码，可在我的Github项目[https://github.com/SeanYanxml/bigdata/tree/master/redis][https_github.com_SeanYanxml_bigdata_tree_master_redis] 目录下可以找到。  
PS: (如果觉得项目不错，可以给我一个Star。)

--------------------

### Demo ###

*  RedisMQConst

public class RedisMQConst {
    
        public static final String MQ_KEY = "MQ_KEY";
    
    }

*  RedisMQDispatchMessageHandler 消息处理器

public class RedisMQDispatchMessageHandler {
    
        public void syncFile(){
            // do something about syncFile thing.
    // while(true){ 
                try{
                    String message = RedisMQService.lpop(RedisMQConst.MQ_KEY);
                    System.out.println("SYNC FILE: "+ System.currentTimeMillis() + " - " + message);
                    if(null != message){
                        // do something about syncFile.
                        // messageHandler.synfile(message);
                        System.out.println("SYNC FILE: "+ System.currentTimeMillis() + " - " + message);
                    }
                    Thread.currentThread().setName(UUID.randomUUID().toString());
                }catch(Exception ex){
                    // do something exception.
                }
    // }
        }
    
    
    
    }

* RedisMQService Redis基本服务工具类（订阅等）

public class RedisMQService { 
    
        public static JedisPoolManager manager = new JedisPoolManager();
    
        // 订阅 subscribe
        public static boolean subscribe(JedisPubSub jedisPubSub, String channels){
            Jedis jedis = manager.getJedis();
            jedis.subscribe(jedisPubSub, channels);
    // jedis.close();
            return true;
        }
    
        public static boolean publish(String channels,String message){
            Jedis jedis = manager.getJedis();
            jedis.publish(channels, message);
            jedis.close();
            return true;
        }
    
        public static String lpop(String key){
            Jedis jedis = manager.getJedis();
            String result = jedis.lpop(key);
            jedis.close();
            return result;
        }
    
        public static void lpush(String key,String value){
            Jedis jedis = manager.getJedis();
            jedis.lpush(key, value);
            jedis.close();
        }
    }
    
    /** * 多线程redis抛出异常B cannot be cast to java.lang.Long * https://blog.csdn.net/JavaMoo/article/details/77233976 * 原因：多个线程同时调用了同一个jedis对象，导致内存数据被多个线程竞争，产生数据混乱 * 解决方案：每个线程都new出一个自己的jedis对象 * Exception in thread "main" java.lang.ClassCastException: java.util.ArrayList cannot be cast to java.lang.Long at redis.clients.jedis.Connection.getIntegerReply(Connection.java:265) at redis.clients.jedis.Jedis.lpush(Jedis.java:882) at com.yanxml.redis.demos.mq.RedisMQService.lpush(RedisMQService.java:23) at com.yanxml.redis.demos.mq.RedisMQDemoMain.main(RedisMQDemoMain.java:15) Exception in thread "Thread-0" redis.clients.jedis.exceptions.JedisDataException: ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context at redis.clients.jedis.Protocol.processError(Protocol.java:127) at redis.clients.jedis.Protocol.process(Protocol.java:161) at redis.clients.jedis.Protocol.read(Protocol.java:215) at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:340) at redis.clients.jedis.Connection.getRawObjectMultiBulkReply(Connection.java:285) at redis.clients.jedis.JedisPubSub.process(JedisPubSub.java:121) at redis.clients.jedis.JedisPubSub.proceed(JedisPubSub.java:115) at redis.clients.jedis.Jedis.subscribe(Jedis.java:2680) at com.yanxml.redis.demos.mq.RedisMQService.subscribe(RedisMQService.java:14) at com.yanxml.redis.demos.mq.RedisMQSubscribeService$1.run(RedisMQSubscribeService.java:17) at java.lang.Thread.run(Thread.java:745) * */

*  RedisMQSubscribeService

public class RedisMQSubscribeService{
    
        private RedisMQSyncListener redisMQSyncListener;//订阅者
    
        public RedisMQSubscribeService(RedisMQSyncListener redisMQSyncListener,String channel){
            this.redisMQSyncListener = redisMQSyncListener;
            subscribe(this.redisMQSyncListener,channel);
        }
    
        public void subscribe(RedisMQSyncListener redisMQSyncListener, String channel) {
            new Thread(new Runnable() {
    
                @Override
                public void run() {
                    System.out.println("Subscribe start.");
    
                    RedisMQService.subscribe(redisMQSyncListener, channel);
    
                    System.out.println("Subscribe OK.");
    
                }
            }).start();
        }
    }

*  RedisMQSyncListener

public class RedisMQSyncListener extends JedisPubSub{ 
    
        private RedisMQDispatchMessageHandler dispatchMessageHandler;
    
        public RedisMQSyncListener(RedisMQDispatchMessageHandler dispatchMessageHandler){
            this.dispatchMessageHandler = dispatchMessageHandler; 
        }
    
        @Override
        public void onMessage(String channel, String message){
            System.out.println("On Message.");
            dispatchMessageHandler.syncFile();
        }
    
    
    }

*  RedisMQPublisherMain

/** * Redis MQ Publisher 启动类 * */
    public class RedisMQPublisherMain { 
        public static void main(String[] args) {
            int i=0;
            while(i<2){
                i++;
                RedisMQService.publish(RedisMQConst.MQ_KEY,String.valueOf(i));
            }
        }
    
    }

*  RedisMQSubscriberMain

/** * Redis MQ Subscriber 启动类 * */
    public class RedisMQSubscriberMain { 
    
        public static void main(String[] args) {
            RedisMQDispatchMessageHandler redisMQDispatchMessageHandler = new RedisMQDispatchMessageHandler();
    
            RedisMQSyncListener redisMQSyncListener = new RedisMQSyncListener(redisMQDispatchMessageHandler);
    
            RedisMQSubscribeService redisMQSubscribeService = new RedisMQSubscribeService(redisMQSyncListener,RedisMQConst.MQ_KEY);
    
    
        }
    
    }

--------------------

### Reference ###

\[1\]. [Jedis实现Publish/Subscribe功能][Jedis_Publish_Subscribe]  
\[2\]. [ 用redis实现消息队列（实时消费+ack机制） ][_redis_ack_]  
\[3\]. [【Redis系列】Redis频道发布与消息订阅][Redis_Redis]

[https_github.com_SeanYanxml_bigdata_tree_master_redis]: https://github.com/SeanYanxml/bigdata/tree/master/redis
[Jedis_Publish_Subscribe]: https://blog.csdn.net/lihao21/article/details/48370687
[_redis_ack_]: https://segmentfault.com/a/1190000012244418
[Redis_Redis]: https://blog.csdn.net/u010176014/article/details/51986758