redis 实现发布订阅功能具体实现

港控/mmm° 2024-03-17 18:49 42阅读 0赞

## 一： 说一下当时使用的场景 ##

因为需要用到服务端调用 客户端的场景，这里就无法用到http请求了，因为客户端的pc无固定IP地址，就像手机一样，如果需要推送消息，就必须做任务调度。但是做任务调度的话不停的循环推送消息存在的问题是，

1：需要不停的循环，如果一天只需要推送几次，但是他调用的次数可能成千上万次。

2：循环肯定就有时间间隔，无法实时触发

出于以上两个问题，先后了解到了netty 的长链接机制，mqtt框架的发送机制，都因为太过复杂，并且集成到项目比较困难，所以选择用redis 的发布订阅机制。

发布订阅特点：

1：有消息时 才推送，不需要不同的循环检索，

2：实时触发消息推送，不需要等待，时间延迟

3：可以用渠道实现点对点推送，也就是可以固定到具体某一个客户端，或者某一类客户端

这样就解决了 服务端触发客户端事件的问题，说通俗点就是，服务端可以根据需要去触发客户端端的方法实现接口。当然这里不能把客户端的方法理解为API，比较不能直接调用

如图，就是 服务端（发布者 也可以理解时redis服务的客户端）跟客户端（订阅者）之间没有直接关系。

## 二:直接用redis控制台实现发布订阅 ##

1：在安装的redis路径下找到redis-cli.exe

2:双击打开一个客户端链接：在上面输入关键之：subscribe

订阅者 **subscribe test1**

SUBSCRIBE channel [channel ...]

订阅给定的一个或多个频道的信息。

test1 是渠道，也就是发布者必须发布到该渠道上，这个订阅者才可以收到消息，不一样的话是收不到信息。

发布者：**publish test1 message**

当在发布者上面发布一条消息后，订阅者马上可以收到一条相关内容的消息。

PUBLISH channel message

将信息 message 发送到指定的频道 channel 。

## 三：java -spring boot 框架下集成redis 发布订阅功能 ##

具体的spring boot集成redis 的流程请参考：

[什么鬼，面试官竟然让我用Redis实现一个消息队列！！？ - 简书][Redis_ -]

这里是在该文档基础上实现 发布订阅功能

### 1：在原来的 RedisConfig 配置基础上继续添加方法 ，全部拷贝如下： ###

这里需要注意：

container.addMessageListener(listenerAdapter, new PatternTopic("test1"));这里是监听渠道，

test1是渠道名称，就是可以监听到这个渠道的所有消息。

这里 addMessageListener 方法可以监听多个渠道，可以是一个lists

订阅者：

@Configuration
        public class RedisConfig {
         
            @Bean
            public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory factory){
                RedisTemplate<String,Object> redisTemplate = new RedisTemplate<>();
                redisTemplate.setConnectionFactory(factory);
                GenericJackson2JsonRedisSerializer serializer = new GenericJackson2JsonRedisSerializer();
                // 值采用json序列化
                redisTemplate.setValueSerializer(serializer);
                //使用StringRedisSerializer来序列化和反序列化redis的key值
                redisTemplate.setKeySerializer(new StringRedisSerializer());
                // 设置hash key 和value序列化模式
                redisTemplate.setHashKeySerializer(new StringRedisSerializer());
                redisTemplate.setHashValueSerializer(serializer);
                redisTemplate.afterPropertiesSet();
                return redisTemplate;
            }
         
            @Bean
            public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                           MessageListenerAdapter listenerAdapter) {
                RedisMessageListenerContainer container = new RedisMessageListenerContainer();
                container.setConnectionFactory(connectionFactory);
         
                Collection<Topic> lists = new ArrayList();
                Topic list1= new PatternTopic("test1");
                Topic list2= new PatternTopic("test2");
                Topic list3= new PatternTopic("test3");
         
                lists.add(list1);
                lists.add(list2);
                lists.add(list3);
                container.addMessageListener(listenerAdapter,lists);
                //container.addMessageListener(listenerAdapter, new PatternTopic("test1"));
                return container;
            }
         
            /**
             * 绑定消息监听者和接收监听的方法,必须要注入这个监听器，不然会报错
             */
            @Bean
            public MessageListenerAdapter listenerAdapter() {
                return new MessageListenerAdapter(new Receiver(), "receiveMessage");
            }
         
        }

### 2：上面的配置中有一个类是自定义类：Receiver ###

在controller里面添加一个类：Receiver，

package com.unit.mapping.controller;
         
        public class Receiver {
            public void receiveMessage(String message) {
                System.out.println(message);
            }
        }

### 3：发布者：这个可以在其他任何项目中实现： ###

用redisTemplate.converAndSend("渠道"，"发送消息") ; 这样发送出去，上面那个项目就可以接收到。

例如：我的服务被某前端调用后，我需要把这个消息分发出去，但是发送给谁呢？这个就由渠道决定。我只需要添加下面的这段代码。把需要分发出去的消息添加到模板中就可以。

package com.unit.mapping.controller;
         
        import org.junit.Test;
        import org.junit.runner.RunWith;
        import org.springframework.boot.test.context.SpringBootTest;
        import org.springframework.data.redis.core.RedisTemplate;
        import org.springframework.test.context.junit4.SpringRunner;
         
        import javax.annotation.Resource;
         
        @RunWith(SpringRunner.class)
        @SpringBootTest
        public class TestPack {
            @Resource
            private RedisTemplate redisTemplate;
            @Test
            public void test(){
                for (int i = 0; i < 10; i++) {
                    System.out.println(i);
                    redisTemplate.convertAndSend("test1","这是我发送的第"+i+"个消息");
                }
            }
        }

[Redis_ -]: https://www.jianshu.com/p/f64567b7b66f