0


【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 后:
在这里插入图片描述
收到了消息,证明我的延时消息已经成功

如您在阅读中发现不足,欢迎留言!!!


本文转载自: https://blog.csdn.net/qq_40065776/article/details/123856087
版权归原作者 Asurplus、 所有, 如有侵权,请联系我们删除。

“【SpringBoot】60、SpringBoot中整合RabbitMQ实现延时队列(死信队列篇)”的评论:

还没有评论