SpringBoot之redis实现消息队列

﹏ヽ暗。殇╰゛Y 2022-03-24 17:52 1149阅读 0赞

### 一.前言 ###

消息队列作为一种常用的异步通信解决方案，而redis是一款高性能的nosql产品，本文介绍如何使用redis实现消息队列，并整合到springboot。

### 二.两个消息模型 ###

##### 1.生产者消费者模式 #####

生产者生产消息放到队列里，多个消费者同时监听队列，谁先抢到消息谁就会从队列中取走消息；即对于每个消息只能被最多一个消费者拥有。

Redis通常都被用做一个处理各种后台工作或消息任务的消息服务器。 一个简单的队列模式就是：生产者把消息放入一个列表中，等待消息的消费者用 RPOP 命令（用轮询方式）， 或者用 BRPOP 命令（如果客户端使用阻塞操作会更好）来得到这个消息。  
![在这里插入图片描述][webp]

##### 2.发布者订阅者模式 #####

发布者生产消息放到队列里，多个监听队列的消费者都会收到同一份消息；即正常情况下每个消费者收到的消息应该都是一样的。

在发布者和订阅者之间存在时间依赖性。发布者需要建立一个topic，以便客户能够购订阅。订阅者必须保持持续的活动状态以接收消息，除非订阅者建立了持久的订阅。在那种情况下，在订阅者未连接时发布的消息将在订阅者重新连接时重新发布。

![在这里插入图片描述][webp 1]

### 三.发布订阅模式实例 ###

##### 1.pom包依赖 #####

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

##### 2.创建一个消息接收者 #####

创建一个Receive,用来接收消息

public class MsgReceiver { 
        private static final Logger logger = LoggerFactory.getLogger(MsgReceiver.class);
    
        private CountDownLatch latch;
    
        @Autowired
        public MsgReceiver(CountDownLatch latch) { 
            this.latch = latch;
        }
    
        public void receiveMessage(String message) { 
            logger.info("消息接收:" + message);
            latch.countDown();
        }
    }

##### 3.配置类 #####

创建连接工厂,绑定消息监听者和接收器,注册订阅者

@Configuration
    public class SubscriberConfig { 
        /** * 创建连接工厂 * @param connectionFactory * @param listenerAdapter * @return */
        @Bean
        public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter){ 
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.addMessageListener(listenerAdapter,new PatternTopic("topic"));
            return container;
        }
    
        /** * 绑定消息监听者和接收监听的方法 * @param receiver * @return */
        @Bean
        public MessageListenerAdapter listenerAdapter(MsgReceiver receiver){ 
            return new MessageListenerAdapter(receiver,"receiveMessage");
        }
    
        /** * 注册订阅者 * @param latch * @return */
        @Bean
        public MsgReceiver receiver(CountDownLatch latch){ 
            return new MsgReceiver(latch);
        }
    
        /** * 计数器，用来控制线程 * @return */
        @Bean
        public CountDownLatch latch(){ 
            return new CountDownLatch(1);//指定了计数的次数 1
        }
    }

##### 4.controller测试方法 #####

@RestController
    public class MsgController { 
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
    
        @GetMapping("/sendMsg")
        public String sendMsg(){ 
            stringRedisTemplate.convertAndSend("topic", "Hello world!");
            return "ok";
        }
    }

访问 [http://localhost:8011/sendMsg][http_localhost_8011_sendMsg] 结果:

> INFO 3176 — \[ container-2\] com.example.redis.receive.MsgReceiver : 消息接收:Hello world!

[webp]: /images/20220323/c5604b9fa38a4a5187055723d6b036bf.png
[webp 1]: /images/20220323/cf62ee3809ef437bbd188a1fbc0da174.png
[http_localhost_8011_sendMsg]: http://localhost:8011/sendMsg