RabbitMQ - 死信、TTL原理、延迟队列安装和配置

清疚 2024-03-02 09:17 34阅读 0赞

**目录**

一、死信交换机

1.1、什么是死信交换机

1.2、TTL

1.2.1、什么是 TTL

1.2.2、通过 TTL 模拟触发死信

二、延迟队列

2.1、什么是延迟队列

2.2、配置延迟队列插件

2.2.1、延迟队列配置

a）下载镜像

b）运行容器

c）刚刚设定的RabbitMQ的数据卷名称为\`mq-plugins\`，所以我们使用下面命令查看数据卷：

d）在此目录下，进入 MQ 容器内部.

e）开启插件

2.3、SpringAMQP 使用延迟队列插件

--------------------

## 一、死信交换机 ##

--------------------

### 1.1、什么是死信交换机 ###

想要知道什么是死信交换机，先来看看什么是死信（dead letter）~

当生产者发送了一个消息，经过交换机到达队列时，满足下列情况之一时，就可以成为死信：

*  消费者使用 basic.reject 或 basic.nack 声明消费失败，并且消息的 requeue 参数设置为 false（消息不重新加入到队列中）.
 *  消息设置了过期时间，到了时间没有被消费掉.
 *  要投递的队列消息堆积满了（队列设置了最大消息数目），最早的消息可能会成为死信（LRU 算法淘汰的消息）.

那么如果这个时候，一个队列配置了 dead-letter-exchange 属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为 死信交换机.

![badc19ba354a47a8ac15b84f6759a870.png][]

### 1.2、TTL ###

#### 1.2.1、什么是 TTL ####

TTL 就是过期时间. 如果一个队列中的消息到了过期时间还没有被消费, 就会变成死信.

这里的消息到了过期时间实际上有两种情况：

*  消息所在的队列设置了消息过期时间（x\_message\_ttl）.
 *  消息本身设置了存活时间.

#### 1.2.2、通过 TTL 模拟触发死信 ####

a）声明一个直接交换机和一个配置了过期时间（x-message-ttl 属性）以及配 deadLetterExchange、deadLetterRoutingKey 属性的普通队列，用来生成死信

@Configuration
    public class TTLMessageConfig {
    
        @Bean
        public DirectExchange ttlDirectExchange() {
            return new DirectExchange("ttl.direct");
        }
    
        @Bean
        public Queue ttlQueue() {
            return QueueBuilder
                    .durable("ttl.queue")
                    .ttl(5000)  //延时 5 s
                    .deadLetterExchange("dl.direct") //消息如果超时没被消费就给这个死信交换机
                    .deadLetterRoutingKey("dl")
                    .build();
        }
    
        @Bean
        public Binding ttlBinding() {
            return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with("ttl");
        }
    
    }

b）这里我们基于注解的方式，声明一组死信交换机和队列

@RabbitListener(bindings = @QueueBinding(
                value = @Queue(name = "dl.queue", durable = "true"),
                exchange = @Exchange(name = "dl.exchange"),
                key = "dl"
        ))
        public void listenDlQueue(String msg) {
            log.info("消费者收到死信消息！msg=" + msg);
        }

c）生产者发送一个过期时间为 5s 的消息

@Test
        public void testTTLMessage() {
            //1.构造一个消息
            Message message = MessageBuilder.withBody("hello ttl message".getBytes())
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                    .setExpiration("5000")
                    .build();
            //2.发送消息
            rabbitTemplate.convertAndSend("ttl.direct", "ttl", message);
            //3.记录日志
            log.info("消息已经成功发送!");
        }

d）执行结果如下

![67b7a4148fb34bdc9f12dcb1d1bf1fa4.png][]

![4a7736b0d1734ab082c7e76850162a0f.png][]

> Ps：通过执行结果，也可以看出，如果消息和队列都设置了过期时间，那么以时间短的为主.

## 二、延迟队列 ##

--------------------

### 2.1、什么是延迟队列 ###

刚刚我们利用 TTL 结合死信交换机，实现了当消息发出后，消费者延迟收到消息的效果。这种消息模式就成为 延迟队列（Delay Queue） 模式。

延迟队列经常用于以下场景：

1.  延迟发送短信.
2.  用户下单，如果再 5 分钟内没有支付，就自动取消.
3.  预约工作会议，10 分钟后自动通知所有参会人员.

### 2.2、配置延迟队列插件 ###

由于 利用 TTL 结合死信交换机的方式实现起来比较麻烦，并且延迟队列的需求又非常多，因此 RabbitMQ 官方推出了一个插件，可以通过更简单的方式，达到延迟队列的效果.

#### 2.2.1、延迟队列配置 ####

我们在Centos7虚拟机中使用Docker来安装。

##### a）下载镜像 #####

docker pull rabbitmq:3.8-management

##### b）运行容器 #####

docker run \
     -e RABBITMQ_DEFAULT_USER=itcast \
     -e RABBITMQ_DEFAULT_PASS=123321 \
     -v mq-plugins:/plugins \
     --name mq \
     --hostname mq1 \
     -p 15672:15672 \
     -p 5672:5672 \
     -d \
     rabbitmq:3.8-management

> Ps：此命令还额外配置了插件目录对应的数据卷.

##### c）刚刚设定的RabbitMQ的数据卷名称为\`mq-plugins\`，所以我们使用下面命令查看数据卷： #####

docker volume inspect mq-plugins

结果如下

![c4f4b40b60cf4270bd0f3ae5ac34eea6.png][]

使用 cd 命令切换到 Mountpoint 指定的目录下.

##### d）在此目录下，进入 MQ 容器内部. #####

我的容器名为\`mq\`，所以执行下面命令：

docker exec -it mq bash

##### e）开启插件 #####

进入容器内部后，执行以下命令开启插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

![32a054c78496437a873dd1972626de1a.png][]

### 2.3、SpringAMQP 使用延迟队列插件 ###

a）声明一个延迟队列. 这里实际上和声明普通交换机只多出了一个 delayed 属性，设置为 true 就表示为延迟队列.

以下是基于 注解的方式声明交换机、队列、绑定.

> Ps：如果是通过 java 代码的方式声明交换机，只需要 ExchangeBuilder().directExhange.delay() 即可.

![bc0d7dd6c1a14e399e661f8f122ec70f.png][]

@Component
    @Slf4j
    public class SpringRabbitListener {
    
        @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue", durable = "true"),
                exchange = @Exchange(name = "delay.direct", delayed = "true"),
                key = "delay"
        ))
        public void listenDelayExchange(String msg) {
            log.info("消费者接收到到了延迟消息！msg=" + msg);
        }
    
    }

b）生产者只需要在生产消息的时候添加一个 header："x-delay"，对应的值就是延迟时间，单位是毫秒：

@Test
        public void testDelayMessage() {
            //1.准备消息
            Message message = MessageBuilder.withBody("hello ttl message".getBytes())
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                    .setHeader("x-delay", 5000) // 消息延迟时间
                    .build();
            //2.消息 ID 需要封装到 CorrelationData 中
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            //3.发送消息
            rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
            log.info("消息已经成功发送!");
        }

c）结果如下

![0e65c4c7b8754193b13a5cfae6dc1c85.png][]

![8b6d41e10f6b498eaf131f1f3e6e4307.png][]

![1d2fff5d47f345dca9e49f2f953cab8c.gif][]

[badc19ba354a47a8ac15b84f6759a870.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/da1516b3f42a4e94bc4b088f039406d2.png
[67b7a4148fb34bdc9f12dcb1d1bf1fa4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/11260761cd834a0a972878986803abd7.png
[4a7736b0d1734ab082c7e76850162a0f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/29c6d38fc7b24c9ea7247d4161bea133.png
[c4f4b40b60cf4270bd0f3ae5ac34eea6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/568c191aebc1417a827fc85a3ee43c07.png
[32a054c78496437a873dd1972626de1a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/c4f8f997c5fe421f9d0b8dd23e234c0b.png
[bc0d7dd6c1a14e399e661f8f122ec70f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/03fbf55348ad4753bccf83d7e4dbefd6.png
[0e65c4c7b8754193b13a5cfae6dc1c85.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/8d4cf9efc73047d99e88ea8949df0a6f.png
[8b6d41e10f6b498eaf131f1f3e6e4307.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/ac2390bd1d944ce990deacf645b64dc4.png
[1d2fff5d47f345dca9e49f2f953cab8c.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/86853a4472834548b66744d9891886bb.png