在很多场景下,我们都有延迟队列的需求,然而默认情况下 RabbitMQ 并未提供延迟队列相关的功能,不过,在前面的文章中,松哥和大家分享了 RabbitMQ 中的死信队列,结合死信队列以及消息过期时间的设定,我们也可以实现延迟队列的功能,今天我们就一起来看下。
RabbitMQ 其实有提供专门的延迟队列插件,这个咱们有空了再聊。
1. 什么时候需要延迟队列?
先来说说什么时候我们需要延迟队列。
举几个简单的例子。
在电商项目中,当我们下单之后,一般需要 20 分钟之内或者 30 分钟之内付款,否则订单就会进入异常处理逻辑中,被取消,那么进入到异常处理逻辑中,就可以当成是一个延迟队列。
我买了一个智能砂锅,可以用来煮粥,上班前把素材都放到锅里,然后设置几点几分开始煮粥,这样下班后就可以喝到香喷喷的粥了,那么这个煮粥的指令也可以看成是一个延迟任务,放到一个延迟队列中,时间到了再执行。
公司的会议预定系统,在会议预定成功后,会在会议开始前半小时通知所有预定该会议的用户。
安全工单超过 24 小时未处理,则自动拉企业微信群提醒相关责任人。
用户下单外卖以后,距离超时时间还有 10 分钟时提醒外卖小哥即将超时。
...
很多场景下我们都需要延迟队列。
可能有小伙伴说至于嘛,直接搞个定时任务不更方便?如果项目中只有一个这样的延迟队列的场景,那么搞个定时任务似乎也可以,但是如果项目中有很多这样的场景,那么定时任务很明显就不是最佳方案了,我们可以通过延迟队列来实现一个通用的解决方案。
2. 延迟队列实现思路
延迟队列实现的思路也很简单,就是上篇文章我们所说的 DLX(死信交换机)+TTL(消息超时时间)。
我们可以把死信队列就当成延迟队列。
具体来说是这样:
假如一条消息需要延迟 30 分钟执行,我们就设置这条消息的有效期为 30 分钟,同时为这条消息配置死信交换机和死信 routing_key,并且不为这个消息队列设置消费者,那么 30 分钟后,这条消息由于没有被消费者消费而进入死信队列,此时我们有一个消费者就在“蹲点”这个死信队列,消息一进入死信队列,就立马被消费了。
这就是延迟队列的实现思路,是不是很简单?
3. 案例
接下来松哥通过一个简单的案例,来和大家演示一下延迟队列的具体实现。
首先准备好一个启动的 RabbitMQ。
然后我们创建一个 Spring Boot 项目,引入 RabbitMQ 依赖:
然后在 application.properties 中配置一下 RabbitMQ 的基本连接信息:
- spring.rabbitmq.host=localhost
- spring.rabbitmq.username=guest
- spring.rabbitmq.password=guest
- spring.rabbitmq.port=5672
接下来我们来配置两个消息队列:一个普通队列,一个死信队列:
- @Configuration
- publicclassQueueConfig{
- publicstaticfinalStringJAVABOY_QUEUE_NAME="javaboy_queue_name";
- publicstaticfinalStringJAVABOY_EXCHANGE_NAME="javaboy_exchange_name";
- publicstaticfinalStringJAVABOY_ROUTING_KEY="javaboy_routing_key";
- publicstaticfinalStringDLX_QUEUE_NAME="dlx_queue_name";
- publicstaticfinalStringDLX_EXCHANGE_NAME="dlx_exchange_name";
- publicstaticfinalStringDLX_ROUTING_KEY="dlx_routing_key";
- /**
- *死信队列
- *@return
- */
- @Bean
- QueuedlxQueue(){
- returnnewQueue(DLX_QUEUE_NAME,true,false,false);
- }
- /**
- *死信交换机
- *@return
- */
- @Bean
- DirectExchangedlxExchange(){
- returnnewDirectExchange(DLX_EXCHANGE_NAME,true,false);
- }
- /**
- *绑定死信队列和死信交换机
- *@return
- */
- @Bean
- BindingdlxBinding(){
- returnBindingBuilder.bind(dlxQueue()).to(dlxExchange())
- .with(DLX_ROUTING_KEY);
- }
- /**
- *普通消息队列
- *@return
- */
- @Bean
- QueuejavaboyQueue(){
-
Map
args=newHashMap<>(); - //设置消息过期时间
- args.put("x-message-ttl",1000*10);
- //设置死信交换机
- args.put("x-dead-letter-exchange",DLX_EXCHANGE_NAME);
- //设置死信routing_key
- args.put("x-dead-letter-routing-key",DLX_ROUTING_KEY);
- returnnewQueue(JAVABOY_QUEUE_NAME,true,false,false,args);
- }
- /**
- *普通交换机
- *@return
- */
- @Bean
- DirectExchangejavaboyExchange(){
- returnnewDirectExchange(JAVABOY_EXCHANGE_NAME,true,false);
- }
- /**
- *绑定普通队列和与之对应的交换机
- *@return
- */
- @Bean
- BindingjavaboyBinding(){
- returnBindingBuilder.bind(javaboyQueue())
- .to(javaboyExchange())
- .with(JAVABOY_ROUTING_KEY);
- }
- }
这段配置代码虽然略长,不过原理其实简单。
配置可以分为两组,第一组配置死信队列,第二组配置普通队列。每一组都由消息队列、消息交换机以及 Binding 三者组成。
配置消息队列时,为消息队列指定死信队列,不熟悉的小伙伴可以翻一下上篇文章,传送门:RabbitMQ 中的消息会过期吗?。
配置队列中的消息过期时间时,默认的时间单位时毫秒。
接下来我们为死信队列配置一个消费者,如下:
- @Component
- publicclassDlxConsumer{
- privatestaticfinalLoggerlogger=LoggerFactory.getLogger(DlxConsumer.class);
- @RabbitListener(queues=QueueConfig.DLX_QUEUE_NAME)
- publicvoidhandle(Stringmsg){
- logger.info(msg);
- }
- }
收到消息后就将之打印出来。
这就完事了。
启动项目。
最后我们在单元测试中发送一条消息:
- @SpringBootTest
- classDelayQueueApplicationTests{
- @Autowired
- RabbitTemplaterabbitTemplate;
- @Test
- voidcontextLoads(){
- System.out.println(newDate());
- rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_EXCHANGE_NAME,QueueConfig.JAVABOY_ROUTING_KEY,"hellojavaboy!");
- }
- }
这个就没啥好说的了,就是普通的消息发送,10 秒之后这条消息会在死信队列的消费者中被打印出来。
4. 小结
好啦,这就是我们用 RabbitMQ 做延迟队列的思路~
原文链接:https://mp.weixin.qq.com/s/x53MSd7VNfA6OwMpFA-Bsg