RabbitMQ延迟消息：死信交换机

╰+攻爆jí腚メ 2024-02-20 09:51 15阅读 0赞

**目录**

一、死信交换机

二、延迟消息插件

声明延迟交换机

发送延迟消息

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

**使用场景**

在电商的支付业务中，对于一些库存有限的商品，为了更好的用户体验，通常都会在用户下单时立刻扣减商品库存。例如电影院购票、高铁购票，下单后就会锁定座位资源，其他人无法重复购买。

但是这样就存在一个问题，假如用户下单后一直不付款，就会一直占有库存资源，导致其他客户无法正常交易，最终导致商户利益受损！

因此，电商中通常的做法就是：**对于超过一定时间未支付的订单，应该立刻取消订单并释放占用的库存**。

例如，订单支付超时时间为30分钟，则我们应该在用户下单后的第30分钟检查订单支付状态，如果发现未支付，应该立刻取消订单，释放库存。

**延迟消息**：生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。

**延迟任务**：设置在一定时间之后才执行的任务

![9fcd7a4b906542de9ff7c0c2ff2686db.png][]

#### 一、死信交换机 ####

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

*  消费者使用`basic.reject`或 `basic.nack`声明消费失败，并且消息的`requeue`参数设置为false
 *  消息是一个过期消息，超时无人消费
 *  要投递的队列消息满了，无法投递

如果一个队列中的消息已经成为死信，并且这个队列通过**`dead-letter-exchange`**属性指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为**死信交换机**（Dead Letter Exchange）。

**死信交换机的作用**

1.  收集那些因处理失败而被拒绝的消息
2.  收集那些因队列满了而被拒绝的消息
3.  收集因TTL（有效期）到期的消息

**首先创建一个队列绑定死信交换机**

![bfe6a1cd9a6b4ec6a1f9a63fce9562e8.png][]

![0efc21f1665a40e49f64a431f55d0cb9.png][] **创建交换机绑定队列**

![423cf40f3f80432b87ca9392eb5ee7e5.png][]

![6dd9e1733b784075914a52882f5ec782.png][]

**绑定队列** ![fdf2741ddbfb441094cf9f934839af63.png][]

**在创建一组队列和交换机**

![a1110d9632d642ce83e96f6161e17708.png][]

![ac9a4cab87d9458b8c39c1d1f25a0d83.png][] **绑定队列**

![76c8ec139f1c466a9b1a68069bb026a8.png][]

生产者发送消息

@Test
        public void SendTTLMessage() {
            // 队列名称
            String queueName = "simple.direct";
            // 消息
            Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8))
                    .setExpiration("10000")//单位毫秒
                    .build();
            // 发送消息
            rabbitTemplate.convertAndSend(queueName,"hello", message);
        }

消费者接受消息

@RabbitListener(queues = "dlx.queue")
        public void listenDlxQueueMessage(Map<String, Object> msg) throws InterruptedException {
            System.out.println("消费者接收到object.queue消息：【" + msg + "】");
        }

![940ded75b58944a1a318941ee2c98851.png][]

十秒钟后消息到死心队列

![fd2f274e48304b698877c3f84d062da3.png][]

#### 二、延迟消息插件 ####

基于死信队列虽然可以实现延迟消息，但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果

官方文档说明： [Scheduling Messages with RabbitMQ | RabbitMQ - Blog][Scheduling Messages with RabbitMQ _ RabbitMQ - Blog]

插件下载地址： [GitHub - rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ][GitHub - rabbitmq_rabbitmq-delayed-message-exchange_ Delayed Messaging for RabbitMQ] 由于我们安装的MQ是`3.8`版本，因此这里下载`3.8.17`版本

![45048e1b80ba4fc9bae1f401e76befe6.png][]

因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。

docker volume inspect mq-plugins

![7759345acb7740c0b3eb2e78732ea224.png][]

插件目录被挂载到了`/var/lib/docker/volumes/mq-plugins/_data`这个目录，我们上传插件到该目录下

**开启插件**

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

![9fcf08987f5240cb80ed229ffe65a268.png][]

##### 声明延迟交换机 #####

基于注解方式：

@RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue", durable = "true"),
            exchange = @Exchange(name = "delay.direct", delayed = "true"),
            key = "delay"
    ))
    public void listenDelayMessage(String msg){
        log.info("接收到delay.queue的延迟消息：{}", msg);
    }

基于`@Bean`的方式：

package com.itheima.consumer.config;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Slf4j
    @Configuration
    public class DelayExchangeConfig {
    
        @Bean
        public DirectExchange delayExchange(){
            return ExchangeBuilder
                    .directExchange("delay.direct") // 指定交换机类型和名称
                    .delayed() // 设置delay的属性为true
                    .durable(true) // 持久化
                    .build();
        }
    
        @Bean
        public Queue delayedQueue(){
            return new Queue("delay.queue");
        }
        
        @Bean
        public Binding delayQueueBinding(){
            return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
        }
    }

##### 发送延迟消息 #####

发送消息时，必须通过x-delay属性设定延迟时间：

@Test
    void testPublisherDelayMessage() {
        // 1.创建消息
        String message = "hello, delayed message";
        // 2.发送消息，利用消息后置处理器添加消息头
        rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                // 添加延迟消息属性
                message.getMessageProperties().setDelay(5000);
                return message;
            }
        });
    }

[9fcd7a4b906542de9ff7c0c2ff2686db.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/8274ba3948d74323a1f2a83ec61216ee.png
[bfe6a1cd9a6b4ec6a1f9a63fce9562e8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/d133793805c74c79b32b37429d9335fb.png
[0efc21f1665a40e49f64a431f55d0cb9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/e68e28fea2154b20a8a47c619374971d.png
[423cf40f3f80432b87ca9392eb5ee7e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/1943f853f05e4c99bc5f079fb7b8d3c0.png
[6dd9e1733b784075914a52882f5ec782.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/9fe95b60fa1e4369a150bbde96b1d8b3.png
[fdf2741ddbfb441094cf9f934839af63.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/5f23da79d34a4a66a70b83fcfd126062.png
[a1110d9632d642ce83e96f6161e17708.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/c4233083fecb4f64908f0bbc59fa8939.png
[ac9a4cab87d9458b8c39c1d1f25a0d83.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/c4c892623e084f989967ece66bfe4bed.png
[76c8ec139f1c466a9b1a68069bb026a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/e32eedc603604eef9bafccae51f15648.png
[940ded75b58944a1a318941ee2c98851.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/ef0ea04074f547488373bb5bee5fa9da.png
[fd2f274e48304b698877c3f84d062da3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/626dce11493a4f42bd7128c3db991457.png
[Scheduling Messages with RabbitMQ _ RabbitMQ - Blog]: https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
[GitHub - rabbitmq_rabbitmq-delayed-message-exchange_ Delayed Messaging for RabbitMQ]: https://github.com/rabbitmq/rabbitmq-delayed-message-exchange
[45048e1b80ba4fc9bae1f401e76befe6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/2dfdce518ca749fb877a6065b377e821.png
[7759345acb7740c0b3eb2e78732ea224.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/a9ce889ba2424e61b814e1a9bb42314c.png
[9fcf08987f5240cb80ed229ffe65a268.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/71c9d1c166fd4669acff4fb34b7d4494.png