JAVA 实现 Redis 发布订阅

一时失言乱红尘 2024-03-16 17:25 81阅读 0赞

## Redis 发布订阅 ##

发布订阅：**消息发布者发布消息** 和 **消息订阅者接收消息**，两者之间通过某种媒介联系起来

> 例如订杂志，当自己订阅了爱格杂志，每个月会发刊一本。到发布的时候派送员将杂志送到自己手上就能看到杂志内容。只有我们订阅了该杂志才会派送给我们

Redis 发布订阅（pub/sub）是一种 消息通信模式 ：发送者（pub）发送消息，订阅者（sub）接收消息。

Redis 客户端可以订阅任意数量的频道。

订阅 / 发布消息图：

![9c6a9091252ce655cfba3af72f443efd.png][]

> 图中可以看出，所需：
> 
> 1.  消息发送者 、 2. 频道 、 3. 消息订阅者

#### 发布订阅机制 ####

1.  当一个客户端通过 `PUBLISH` 命令向订阅者发布消息的时候，称这个客户端为发布者`publisher`
2.  当一个客户端通过`subscribe` 或者 `PSUBSCRIBE` 接收消息时，称这个客户端为 订阅者 `subscriber`
3.  为了解耦发布者和订阅者之间的关系，Redis 使用了频道`channel(频道)`作为两者之间的中介，发布者直接把消息发送给 channel，而 channel 负责把消息发送给订阅者，发布者和订阅者之间没有直接的联系，都不知道对方的存在

> 订阅者 1，2，3 订阅了频道 channel，当有消息发布给频道时，这个消息就会被发送到三个订阅者客户端

![98f324adce9ac375b5790275c3edcd09.png][]

#### demo 实现 ####

[学习参考链接][Link 1]

引入依赖

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

添加配置文件

spring:
      redis:
        host: 127.0.0.1
        database: 5
        password:
        port: 6379

1.  ## Listener模式 ##

创建一个监听容器

@Configuration
    public class CatListenerConfig extends CachingConfigurerSupport {
        
    
        /**
         * 消息监听容器
         *
         * @param factory
         * @return
         */
        @Bean
        RedisMessageListenerContainer container(RedisConnectionFactory factory){
        
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            container.setConnectionFactory(factory);
            //订阅一个通道 该处的通道名是发布消息时的名称
           container.setConnectionFactory(connectionFactory);
            //订阅了一个叫cat 的通道
            container.addMessageListener(catAdapter, new PatternTopic("cat"));
            container.addMessageListener(fishAdapter, new PatternTopic("fish"));
            return container;
        }
    
        
    
    }

@Component
    public class CatListener implements MessageListener {
        
        @Override
        public void onMessage(Message message, byte[] bytes) {
        
            System.out.println("我是监听者,我监听到的消息是 " + message.toString());
        }
    }

package com.maoxs.listener;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.connection.Message;
    import org.springframework.data.redis.connection.MessageListener;
    import org.springframework.data.redis.core.RedisTemplate;
    
    /**
     * 监听发送的消息
     */
    public class FishListener  implements MessageListener  {
        
    
        @Autowired
        RedisTemplate redisTemplate;
    
        @Override
        public void onMessage(Message message, byte[] bytes) {
        
            System.out.println("我是Fish监听" + message.toString());
        }
    }

建测试类，测试发布监听

@RestController
    public class TestController {
        
    
        @Resource
        StringRedisTemplate stringRedisTemplate;
    
        @PostMapping("/cat")
        public void test2(){
        
            stringRedisTemplate.convertAndSend("cat","测试：消息发布者发布消息");
        }
      
       @PostMapping("/fish")
        public void fish(){
        
            stringRedisTemplate.convertAndSend("fish","测试：消息发布者发布消息");
        }
    }

测试结果

![c5bebffae9178e6389221a17638748f2.png][]

1.  ## Adapter模式 ##
    
    2.1 配置
    
        /**
             * MessageListenerAdapter 模式
             * 该处topic的 key为bean的name
             * @param connectionFactory
             * @param adapterMap
             * @return
             */
            @Bean
            public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,Map<String, MessageListenerAdapter> adapterMap) {
              
                RedisMessageListenerContainer container = new RedisMessageListenerContainer();
                container.setConnectionFactory(connectionFactory);
                adapterMap.keySet().forEach(topic-> container.addMessageListener(adapterMap.get(topic),new PatternTopic(topic)));
                return container;
            }
    
    ### 监听1 ###
    
        package com.sst.loan.risk.listener;
        
        import com.sst.loan.risk.manager.RuleLoadManager;
        import lombok.extern.slf4j.Slf4j;
        import org.springframework.data.redis.connection.Message;
        import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
        import org.springframework.stereotype.Component;
        
        import javax.annotation.Resource;
        
        /**
         * 监听适配器
         *
         * @author 蔡定努
         * @date 2023/06/13 10:26
         */
        @Slf4j
        @Component("ruleRefreshAdapter")
        public class RuleRefreshAdapter  extends MessageListenerAdapter {
              
          
        
        
            @Override
            public void onMessage(Message message, byte[] bytes) {
              
                log.info(">>>>>>> 消息适配器收到刷新规则的请求 <<<<<<<<<<<<<<");
        
            }
        }
    
    ### 监听2 ###
    
        package com.sst.loan.risk.listener;
        
        import com.sst.loan.risk.manager.RuleLoadManager;
        import lombok.extern.slf4j.Slf4j;
        import org.springframework.data.redis.connection.Message;
        import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
        import org.springframework.stereotype.Component;
        
        import javax.annotation.Resource;
        
        /**
         * 监听适配器
         *
         * @author 蔡定努
         * @date 2023/06/13 10:26
         */
        @Slf4j
        @Component("ruleRefreshAdapter2")
        public class RuleRefreshAdapter2  extends MessageListenerAdapter {
              
          
        
        
            @Override
            public void onMessage(Message message, byte[] bytes) {
              
                log.info(">>>>>>> 消息适配器收到刷新规则的请求 <<<<<<<<<<<<<<");
                
            }
        }
    
    ## 测试 ##
    
        @Resource
            private StringRedisTemplate stringRedisTemplate;
        
        
            /**
             * 
             * @author 蔡定努
             */
            @GetMapping("refresh")
            public Object refresh() {
              
                stringRedisTemplate.convertAndSend("ruleRefreshAdapter","refresh");
                return Result.success();
            }
            
            
            /**
             * 
             * @author 蔡定努
             */
            @GetMapping("refresh")
            public Object refresh() {
              
                stringRedisTemplate.convertAndSend("ruleRefreshAdapter2","refresh");
                return Result.success();
            }

[9c6a9091252ce655cfba3af72f443efd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/31bd0a7105aa4864a9f334b48e2a40a9.png
[98f324adce9ac375b5790275c3edcd09.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2789bd0710e347a19ec65ca2747d99a8.png
[Link 1]: https://blog.csdn.net/qq_32867467/article/details/82944209
[c5bebffae9178e6389221a17638748f2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/89ac3ca0d7644a548777a8f5e406e5dc.png