SpringBoot和redis发布订阅

小咪咪 2022-04-15 06:21 229阅读 0赞

### 前言： ###

> 发布订阅是消息的一种常用模式。例如，在企业分配任务之后 ，可以通过邮件、 短信或者微信  
> 通知到相关的责任人，这就是一种典型的发布订阅模式。首先是 Redis 提供一个渠道，让消息能够发  
> 送到这个渠道上 ，而多个系统可以监听这个渠道， 如短信、微信和邮件系统都可以监昕这个渠道，  
> 当一条消息发送到渠道，渠道就会通知它的监昕者，这样短信、微信和邮件系统就能够得到这个渠  
> 道给它们的消息了，这些监听者会根据自己的需要去处理这个消息，于是我们就可以得到各种各样  
> 的通知了 。 其原理如下图：

### ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70][] ###

### 本次测试为了方便测试，发布和订阅都在一个工程中，代码如下： ###

### **1.导入依赖** ###

<parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.0.4.RELEASE</version>
            <relativePath/> 
        </parent>
    
       <dependencies>
    
           
           <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
                <version>2.0.0.M7</version>
            </dependency>
    
           
           <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-redis</artifactId>
                <version>1.5.7.RELEASE</version>
           </dependency>
    
       </dependencies>

### **2.application.properties配置如下** ###

server.port=9999
    #redis
    spring.redis.port=6379
    spring.redis.jedis.pool.min-idle=5
    spring.redis.jedis.pool.max-active=10
    spring.redis.jedis.pool.max-idle=10
    spring.redis.jedis.pool.max-wait=2000ms
    spring.redis.host=localhost
    spring.redis.timeout=1000ms

### **3.redis订阅方** ###

> 为了接收 Redis 渠道发送过来的消息，我们先定义一个消息监听器（ MessageListener ），代码如下：

/**
     * 为了接收 Redis 渠道发送过来的消息，我们先定义一个消息监听器（ MessageListener ），代码
     */
    @Component
    public class RedisSubscribe implements MessageListener {
    
        /**
         * 这里的 onMessage 方法是得到消息后的处理方法， 其中 message 参数代表 Redis 发送过来的消息，
         * pattern是渠道名称，onMessage方法里打印 了它们的内容。这里因为标注了 ＠Component 注解，所以
         * 在 Spring Boot 扫描后，会把它自动装配到 IoC 容器中 ,监听着对象RedisMessageListener会自动
         * 将消息进行转换。
         * @param message
         * @param pattern
         */
        @Override
        public void onMessage(Message message, byte[] pattern) {
            //消息体
            String body = new String(message.getBody());
            //渠道名称
            String topic = new String(pattern);
            System.out.println("消息体："+body);
            System.out.println("渠道名称："+topic);
        }
    }

> 接着我们在 Spring Boot 的启动文件中配置其他信息，让系统能够监控 Redis 的消息，代码如下：

/**
     * @Author: Lee
     * @Date: 2018/11/7 13:25
     * @Description:
     */
    @Configuration
    public class RedisSubConfig {
    
        /**
         * 这里 RedisTemplate 和 RedisConnectionFactory 对象都是 Spring Boot 自动创建的，所以这里只是
         * 把它们注入进来，只需要使用＠Autowired 注解即可 。然后定义了一个任务池 ，并设置了任务池大小
         * 为 2 0 ，这样它将可 以运行线程 ，井进行阻塞，等待 Redis 消息的传入。接着再定义了一个 Redis消息
         * 监听的容器 RedisMessageListenerContainer，并且往容器设置了 Redis 连接工厂和指定运行消息的线
         * 程池，定义了接收“ topicl ”渠道的消息，这样系统就可以监听 Redis 关于“ topicl ＂渠道的消息了 。
         */
    
    
    
        /** redis 连接工厂 */
        @Resource
        private RedisConnectionFactory redisConnectionFactory;
    
        /**
         *     Redis消息监听器
         */
        @Resource
        private MessageListener redisMsgListener;
    
        //任务池
        private ThreadPoolTaskScheduler taskScheduler;
    
        /**
         * 创建任务池，运行线程等待处理redis的消息
         */
        @Bean
        public ThreadPoolTaskScheduler initTaskScheduler() {
            if (null != taskScheduler) {
                return taskScheduler;
            }
            taskScheduler = new ThreadPoolTaskScheduler();
            taskScheduler.setPoolSize(20);
            return taskScheduler;
        }
    
        /**
         * 定义redis的监听器
         * @return 监听容器
         */
        @Bean
        public RedisMessageListenerContainer initRedisContainer() {
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            //Redis连接工厂
            container.setConnectionFactory(redisConnectionFactory);
            //设置运行任务的线程池
            container.setTaskExecutor(initTaskScheduler());
            //定义监听渠道，名称为topic1
            Topic topic = new ChannelTopic("topic1");
            //使用监听器监听Redis的消息
            container.addMessageListener(redisMsgListener, topic);
            return container;
        }
    }

### **4.redis发布方** ###

> 创建一个控制器和处理器，代码如下：

@RestController
    public class TestController {
    
      
        @Resource
        private RedisTemplate redisTemplate;
    
        @GetMapping(value = "/testPush", produces = "application/json;charset=utf-8")
        public void testPush(String body) {
            /**
             * 使用redisTemplate的convertAndSend()函数，
             * String channel, Object message
             * channel代表管道，
             * message代表发送的信息
             */
            redisTemplate.convertAndSend("topic1", body);
        }
    }

>  这样就实现了异步的发布消息，订阅消息了。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70]: /images/20220415/462004cd8ac449f6bb22279cc1a3feef.png