【SpringBoot】60、SpringBoot中整合RabbitMQ实现延时队列（死信队列篇）

延时插件实现篇，参考文章：【SpringBoot】43、SpringBoot中整合RabbitMQ实现延时队列（延时插件篇）

一、实现原理

• 1、什么是死信队列

死信队列：DLX，dead-letter-exchange
利用DLX，当消息在一个队列中变成死信 (dead message) 之后，它能被重新publish到另一个Exchange，这个Exchange就是DLX

• 2、消息变成死信有以下几种情况

消息被拒绝(basic.reject / basic.nack)，并且requeue = false
消息TTL过期
队列达到最大长度

• 3、死信处理过程

DLX也是一个正常的Exchange，和一般的Exchange没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。
当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去，进而被路由到另一个队列。
可以监听这个队列中的消息做相应的处理。

• 4、死信队列实现延时消息原理

我们将消息发送到消息队列中，并设置一个过期时间，该队列没有消费者
消息的过期时间到了之后，由于没有消费者，就会进入死信队列
我们用一个消费者接收死信队列的消息，就能达到延迟消息的目的

二、实现过程

• 1、引入 maven 依赖

<!-- rabbitmq消息队列 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

• 2、配置 MQ 连接信息

spring:
  rabbitmq:
    host:127.0.0.1
    port:5672
    username: guest
    password: guest

• 3、创建队列，并绑定交换机

importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.DirectExchange;importorg.springframework.amqp.core.Queue;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importjava.util.HashMap;importjava.util.Map;@ConfigurationpublicclassRabbitMqConfig{/**
     * 死信队列
     */publicstaticfinalString DLX_QUEUE ="dlx_queue";publicstaticfinalString DLX_EXCHANGE ="dlx_exchange";publicstaticfinalString DLX_ROUTING_KEY ="dlx_routing_key";/**
     * 正常队列
     */publicstaticfinalString MSG_QUEUE ="msg_queue";publicstaticfinalString MSG_EXCHANGE ="msg_exchange";publicstaticfinalString MSG_ROUTING_KEY ="msg_routing_key";/**
     * 死信队列
     *
     * @return
     */@BeanQueuedlxQueue(){returnnewQueue(DLX_QUEUE,true,false,false);}/**
     * 死信交换机
     *
     * @return
     */@BeanDirectExchangedlxExchange(){returnnewDirectExchange(DLX_EXCHANGE,true,false);}/**
     * 绑定死信队列和死信交换机
     *
     * @return
     */@BeanBindingdlxBinding(){returnBindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(DLX_ROUTING_KEY);}/**
     * 普通消息队列
     *
     * @return
     */@BeanQueuemsgQueue(){Map<String,Object> args =newHashMap<>();//设置消息过期时间
        args.put("x-message-ttl",1000*10);//设置死信交换机
        args.put("x-dead-letter-exchange", DLX_EXCHANGE);//设置死信 routing_key
        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);returnnewQueue(MSG_QUEUE,true,false,false, args);}/**
     * 普通交换机
     *
     * @return
     */@BeanDirectExchangemsgExchange(){returnnewDirectExchange(MSG_EXCHANGE,true,false);}/**
     * 绑定普通队列和与之对应的交换机
     *
     * @return
     */@BeanBindingmsgBinding(){returnBindingBuilder.bind(msgQueue()).to(msgExchange()).with(MSG_ROUTING_KEY);}}

分别创建了私信队列和普通消息队列，普通消息队列设置了消息过期时间为 10s，设置了死信交换机、死信 routing_key，消息过期之后就能进入了死信队列中。

• 4、死信队列消费者

importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@Slf4j@ComponentpublicclassDlxConsumer{@RabbitListener(queues =RabbitMqConfig.DLX_QUEUE)publicvoidhandle(String message){
        log.info("------------------收到消息："+ message);}}

• 5、普通队列消息生产者

importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;@Slf4j@ComponentpublicclassMsgProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;publicvoidsendMsg(String message){try{
            rabbitTemplate.convertAndSend(RabbitMqConfig.MSG_EXCHANGE,RabbitMqConfig.MSG_ROUTING_KEY, message);
            log.info("-------------------消息发送成功");}catch(Exception e){
            log.error("-------------------消息发送失败");}}}

三、测试

• 1、测试用例

importcom.asurplus.common.rabbitmq.MsgProducer;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;@RestControllerpublicclassTestController{@AutowiredprivateMsgProducer msgProducer;@GetMapping("test")publicStringtest(){
        msgProducer.sendMsg("今天天气好好");return"success";}}

• 2、发送消息

访问接口：

localhost:8080/test

消息发送成功，10s 后：

收到了消息，证明我的延时消息已经成功

如您在阅读中发现不足，欢迎留言！！！