SpringBoot整合RabbitMQ系列--用死信队列实现延迟消息

淩亂°似流年 2023-09-29 15:58 15阅读 0赞

原文网址：[SpringBoot整合RabbitMQ系列--用死信队列实现延迟消息\_IT利刃出鞘的博客-CSDN博客][SpringBoot_RabbitMQ_--_IT_-CSDN]

## 简介 ##

**说明**

本文用示例介绍SpringBoot整合RabbitMQ时如何处理死信队列/延迟队列。

**RabbitMQ消息简介**

RabbitMQ的消息默认不会超时。

**什么是死信队列？什么是延迟队列？**

死信队列：

DLX，全称为Dead-Letter-Exchange,可以称之为死信交换器，也有人称之为死信邮箱。当消息在一个队列中变成死信（dead message)之后，它能被重新被发送到另一个交换器中，这个交换器就是DLX，绑定DLX的队列就称之为死信队列。

以下几种情况会导致消息变成死信：

1.  消息被拒绝（Basic.Reject/Basic.Nack)，并且设置requeue参数为false;
2.  消息过期；
3.  队列达到最大长度。

延迟队列：

延迟队列用来存放延迟消息。延迟消息：指当消息被发送以后，不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。

**相关网址**

[RabbitMQ--死信队列/延迟队列--使用/原理\_IT利刃出鞘的博客-CSDN博客\_死信队列原理][RabbitMQ--_--_IT_-CSDN]

## 实例代码 ##

### 路由配置 ###

package com.example.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitRouterConfig {
        public static final String EXCHANGE_TOPIC_WELCOME   = "Exchange@topic.welcome";
        public static final String EXCHANGE_FANOUT_UNROUTE  = "Exchange@fanout.unroute";
        public static final String EXCHANGE_TOPIC_DELAY     = "Exchange@topic.delay";
    
        public static final String ROUTINGKEY_HELLOS        = "hello.#";
        public static final String ROUTINGKEY_DELAY         = "delay.#";
    
        public static final String QUEUE_HELLO              = "Queue@hello";
        public static final String QUEUE_HI                 = "Queue@hi";
        public static final String QUEUE_UNROUTE            = "Queue@unroute";
        public static final String QUEUE_DELAY              = "Queue@delay";
    
        public static final Integer TTL_QUEUE_MESSAGE       = 5000;
    
        @Autowired
        AmqpAdmin amqpAdmin;
    
        @Bean
        Object initBindingTest() {
            amqpAdmin.declareExchange(ExchangeBuilder.fanoutExchange(EXCHANGE_FANOUT_UNROUTE).durable(true).autoDelete().build());
            amqpAdmin.declareExchange(ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_DELAY).durable(true).autoDelete().build());
            amqpAdmin.declareExchange(ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_WELCOME)
                    .durable(true)
                    .autoDelete()
                    .withArgument("alternate-exchange", EXCHANGE_FANOUT_UNROUTE)
    
                    .build());
    
            amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_HI).build());
            amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_HELLO)
                    .withArgument("x-dead-letter-exchange", EXCHANGE_TOPIC_DELAY)
                    .withArgument("x-dead-letter-routing-key", ROUTINGKEY_DELAY)
                    .withArgument("x-message-ttl", TTL_QUEUE_MESSAGE)
                    .build());
            amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_UNROUTE).build());
            amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_DELAY).build());
    
            amqpAdmin.declareBinding(new Binding(QUEUE_HELLO, Binding.DestinationType.QUEUE,
                    EXCHANGE_TOPIC_WELCOME, ROUTINGKEY_HELLOS, null));
            amqpAdmin.declareBinding(new Binding(QUEUE_UNROUTE, Binding.DestinationType.QUEUE,
                    EXCHANGE_FANOUT_UNROUTE, "", null));
            amqpAdmin.declareBinding(new Binding(QUEUE_DELAY, Binding.DestinationType.QUEUE,
                    EXCHANGE_TOPIC_DELAY, ROUTINGKEY_DELAY, null));
    
            return new Object();
        }
    }

### 控制器 ###

package com.example.controller;
    
    import com.example.config.RabbitRouterConfig;
    import com.example.mq.Sender;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.time.LocalDateTime;
    
    @RestController
    public class HelloController {
        @Autowired
        private Sender sender;
    
        @PostMapping("/hi")
        public void hi() {
            sender.send(RabbitRouterConfig.QUEUE_HI, "hi1 message:" + LocalDateTime.now());
        }
    
        @PostMapping("/hello1")
        public void hello1() {
            sender.send("hello.a", "hello1 message:" + LocalDateTime.now());
        }
    
        @PostMapping("/hello2")
        public void hello2() {
            sender.send(RabbitRouterConfig.EXCHANGE_TOPIC_WELCOME, "hello.b", "hello2 message:" + LocalDateTime.now());
        }
    
        @PostMapping("/ae")
        public void aeTest() {
            sender.send(RabbitRouterConfig.EXCHANGE_TOPIC_WELCOME, "nonono", "ae message:" + LocalDateTime.now());
        }
    }

### 发送器 ###

package com.example.mq;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    
    @Component
    public class Sender {
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send(String routingKey, String message) {
            this.rabbitTemplate.convertAndSend(routingKey, message);
        }
    
        public void send(String exchange, String routingKey, String message) {
            this.rabbitTemplate.convertAndSend(exchange, routingKey, message);
        }
    }

### 接收器 ###

package com.example.mq;
    
    import com.example.config.RabbitRouterConfig;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Receiver {
        @RabbitListener(queues = RabbitRouterConfig.QUEUE_HI)
        public void hi(String payload) {
            System.out.println ("Receiver(hi) : "  + payload);
        }
    
        // @RabbitListener(queues = RabbitRouterConfig.QUEUE_HELLO)
        // public void hello(String hello) throws InterruptedException {
        //     System.out.println ("Receiver(hello) : "  + hello);
        //     Thread.sleep(5 * 1000);
        //     System.out.println("(hello):sleep over");
        // }
        //
        // @RabbitListener(queues = RabbitRouterConfig.QUEUE_UNROUTE)
        // public void unroute(String hello) throws InterruptedException {
        //     System.out.println ("Receiver(unroute) : "  + hello);
        //     Thread.sleep(5 * 1000);
        //     System.out.println("(unroute):sleep over");
        // }
    
        @RabbitListener(queues = RabbitRouterConfig.QUEUE_DELAY)
        public void delay(String hello) throws InterruptedException {
            System.out.println ("Receiver(delay) : "  + hello);
            Thread.sleep(5 * 1000);
            System.out.println("(delay):sleep over");
        }
    }

### application.yml ###

server:
    #  port: 9100
      port: 9101
    spring:
      application:
    #    name: demo-rabbitmq-sender
        name: demo-rabbitmq-receiver
      rabbitmq:
        host: localhost
        port: 5672
        username: admin
        password: 123456
    #    virtualHost: /
        publisher-confirms: true
        publisher-returns: true
    #    listener:
    #      simple:
    #        acknowledge-mode: manual
    #      direct:
    #        acknowledge-mode: manual

### 实例测试 ###

分别启动发送者和接收者。

访问：[Node Exporter][]

五秒钟后输出：

Receiver(delay) : hello2 message:2020-11-27T09:30:51.548
    (delay):sleep over

## 其他网址 ##

《RabbitMQ实战指南》=> 第4章 RabbitMQ进阶

[SpringBoot_RabbitMQ_--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/124955698
[RabbitMQ--_--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/121320951
[Node Exporter]: http://localhost:9100/hello2