一文搞懂 RabbitMQ 延时队列（订单定时取消为例）

叁歲伎倆 2023-10-01 20:03 68阅读 0赞

### 1. 死信及死信队列 ###

#### 1.1 什么是死信 ####

一般来说，生产者将消息投递到队列中，消费者从队列取出消息进行消费，但某些时候由于特定的原因导致队列中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了**死信**(Dead Letter)，所有的死信都会放到死信队列中。

**为什么为有死信**？消息变成死信一般是以下三种情况：

1.  **消息被拒绝**，即`basicReject/basicNack`，并且设置 requeue 参数为 false，这种情况一般消息丢失 。
2.  **消息过期（TTL）**，TTL全称为Time-To-Live，表示的是消息的有效期，默认情况下 Rabbit 中的消息不过期，但是可以设置队列的过期时间和消息的过期时间以达到消息过期的效果 ，消息如果在队列中一直没有被消费并且存在时间超过了TTL，消息就会变成了"死信" ，后续无法再被消费。
3.  **队列达到最大长度**，一般当设置了最大队列长度或大小并达到最大值时。

#### 1.2 死信交换器 DLX ####

在消息的拒绝操作都是在`requeue = true`情形下，如果为 false 可以发现当发生异常确认后，消息丢失了，这肯定是不能容忍的，所以提出了死信交换器（dead-letter-exchange）的概念。

死信交换器仍然只是一个普通的交换器，创建时并没有特别要求和操作。在创建队列的时候，声明该交换器将用作保存被拒绝的消息即可，相关的参数是 `x-dead-letter-exchange`。当这个队列中有死信时，RabbitMQ 就会自动的将这个消息重新发布到设置的 Exchange 上去，进而被路由到另一个队列。

**举个栗子**

1、生产者生产 3 条消息

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class DlxProducer {
        
    
        public final static String EXCHANGE_NAME = "dlx_exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            //建立连接
            Connection connection = RabbitMQUtils.getConnection();
            // 创建一个信道
            Channel channel = connection.createChannel();
            // 指定转发
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            String[] routekeys = {
        "rabbit", "cat", "dog"};
            for (int i = 0; i < 3; i++) {
        
                String routekey = routekeys[i % 3];
                String msg = "Hello,RabbitMq" + (i + 1);
                channel.basicPublish(EXCHANGE_NAME, routekey, null, msg.getBytes());
                System.out.println("Sent " + routekey + ":" + msg);
            }
            // 关闭频道和连接
            channel.close();
            connection.close();
        }
    }

2、普通消费者消费消息，但是不能消费全部的消息，并把不能消费得消息投递到死信队列。如果是我们还想做点其他事情，我们可以在死信交换的时候改变死信消息的路由键，具体的相关的参数是 `x-dead-letter-routing-key`。

import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 类说明：普通的消费者,但是自己无法消费的消息，将投入死信队列
     */
    public class NormalDlxConsumer {
        
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            //建立连接
            Connection connection = RabbitMQUtils.getConnection();
            // 创建一个信道
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(DlxProducer.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            //绑定死信交换器
            //声明一个队列，并绑定死信交换器
            String queueName = "dlx_queue";
            Map<String, Object> argos = new HashMap<String, Object>();
            argos.put("x-dead-letter-exchange", DlxConsumer.DLX_EXCHANGE_NAME);
            //死信路由键，会替换消息原来的路由键
            //args.put("x-dead-letter-routing-key", "dead");
            channel.queueDeclare(queueName, false, true, false, argos);
            //绑定，将队列和交换器通过路由键进行绑定
            channel.queueBind(queueName, DlxProducer.EXCHANGE_NAME, "#");
            System.out.println("waiting for message........");
            final Consumer consumer = new DefaultConsumer(channel) {
        
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
        
                    String message = new String(body, "UTF-8");
                    //如果是cat的消息确认
                    if (envelope.getRoutingKey().equals("cat")) {
        
                        System.out.println("Received[" + envelope.getRoutingKey() + "]" + message);
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    } else {
        
                        //如果是其他的消息拒绝（queue=false），成为死信消息
                        System.out.println("Will reject[" + envelope.getRoutingKey() + "]" + message);
                        channel.basicReject(envelope.getDeliveryTag(), false);
                    }
                }
            };
            channel.basicConsume(queueName, false, consumer);
        }
    }

3、申明一个消费者，负责消费死信队列

mport com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 类说明：普通的消费者，负责消费死信队列dlx_accept
     */
    public class DlxConsumer {
        
    
        public final static String DLX_EXCHANGE_NAME = "dlx_accept";
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            //建立连接
            Connection connection = RabbitMQUtils.getConnection();
            // 创建一个信道
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(DLX_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            String queueName = "dlx_accept";
            channel.queueDeclare(queueName, false, false, false, null);
            channel.queueBind(queueName, DLX_EXCHANGE_NAME, "#");
            System.out.println("waiting for message........");
            //声明了一个死信消费者
            final Consumer consumer = new DefaultConsumer(channel) {
        
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
        
                    String message = new String(body, "UTF-8");
                    System.out.println("Received dead letter[" + envelope.getRoutingKey() + "]" + message);
                }
            };
            //消费者正式开始在指定队列上消费消息
            channel.basicConsume(queueName, true, consumer);
        }
    }

测试结果：

![image-20211019141510180][]

**DLX和备用交换器的区别**

1.  备用交换器是主交换器无法路由消息，那么消息将被路由到这个新的备用交换器，而死信交换器则是接收过期或者被拒绝的消息。
2.  备用交换器是在声明主交换器时发生联系，而死信交换器则声明队列时发生联系。

场景分析：备用交换器一般是用于生产者生产消息时，确保消息可以尽量进入 RabbitMQ，而死信交换器主要是用于消费者消费消息产生死信的场景（比如消息过期，队列满了，消息拒绝且不重新投递）。

### 2. 什么是延时队列 ###

延时队列，首先，它是一种队列，队列意味着内部的元素是有序的，元素出队和入队是有方向性的，元素从一端进入，从另一端取出。

其次，延时队列，最重要的特性就体现在它的延时属性上，跟普通的队列不一样的是，**普通队列中的元素总是等着希望被早点取出处理，而延时队列中的元素则是希望被在指定时间得到取出和处理**，所以延时队列中的元素是都是带时间属性的，通常来说是需要被处理的消息或者任务。

简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

RabbitMQ 是没有延时属性可以设置的，但是可以通过**DLX+TTL**的方式来实现 RabbitMQ 的延时队列。

### 3. 延时队列的使用场景 ###

那么什么时候需要用延时队列呢？考虑一下以下场景：

1.  订单在十分钟之内未支付则自动取消。
2.  新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
3.  账单在一周内未支付，则自动结算。
4.  用户注册成功后，如果三天内没有登陆则进行短信提醒。
5.  用户发起退款，如果三天内没有得到处理则通知相关运营人员。
6.  预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。

这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，如：发生订单生成事件，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭；发生店铺创建事件，十天后检查该店铺上新商品数，然后通知上新数为 0 的商户；发生账单生成事件，检查账单支付状态，然后自动结算未支付的账单；发生新用户注册事件，三天后检查新注册用户的活动数据，然后通知没有任何活动记录的用户；发生退款事件，在三天之后检查该订单是否已被处理，如仍未被处理，则发送消息给相关运营人员；发生预定会议事件，判断离会议开始是否只有十分钟了，如果是，则通知各个与会人员。

看起来似乎使用定时任务，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？如果数据量比较少，确实可以这样做，比如：对于“如果账单一周内未支付则进行自动结算”这样的需求，如果对于时间不是严格限制，而是宽松意义上的一周，那么每天晚上跑个定时任务检查一下所有未支付的账单，确实也是一个可行的方案。但对于数据量比较大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，活动期间甚至会达到百万甚至千万级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。

### 4. 延时队列的实现 ###

在 RabbitMQ 中一般采用的是 `TTL+DLX` 的方式来实现延时队列，DLX 上面已经介绍了，通过在创建队列的时候设置队列的`x-dead-letter-exchange`属性，而 TTL 也同样可以设置属性`x-message-ttl`，如下：

Map<String, Object> args = new HashMap<String, Object>();
    args.put("x-message-ttl", 5000);
    channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

这样所有被投递到该队列的消息都最多不会存活超过 5s。

集齐了延时队列的两大要素，现在来看看一条延时消息的处理过程，我们以机票订单为例，一般机票下单之后会要求在 30 分钟之内支付，如果 30 分钟之后没有支付则取消该订单（为什么要取消，因为你站位了又不支付，不能影响别人买票），在我之前的项目中处理方式就是通过定时任务直接查库来处理的，因为当时的订单量并不是很大。

当下单后我们把订单信息发送到 MQ 的延时队列中，并设置 30 分钟过期，30 分钟以后延时队列的数据在转发到死信队列中去，然后我们从死信队列中获取订单信息，并判断它的支付状态，如果已经支付，不做任何处理，如果未支付，则取消订单。

#### 4.1 实现原理 ####

生产者发送消息到延迟队列，对延迟队列或消息设置过期时间（TTL），过期之后通过死信交换机（DLX）把消息重新发送到需要消费的队列（死信队列）中去进行消费。

![image-20220224192355290][]

1.  创建死信队列；
2.  创建死信交换机；
3.  死信队列和死信交换机绑定；
4.  创建延时队列，通过`TTL+DLX`，并配置`x-dead-letter-exchange`和`x-message-ttl`属性；
5.  创建延时交换机；
6.  延时队列和延时交换机绑定。

#### 4.2 配置队列和交换机 ####

常量

public class DelayConstant {
        
    
        /**
         * 延迟队列 TTL 名称
         */
        public static final String ORDER_DELAY_QUEUE = "ticket.order.delay.queue";
    
        /**
         * 延时队列
         * 延时消息就是发送到该交换机的
         */
        public static final String ORDER_DELAY_EXCHANGE = "ticket.order.delay.exchange";
    
        /**
         * routing key 名称 路由键
         * 具体延时消息发送在该 routingKey 的
         */
        public static final String ORDER_DELAY_ROUTING_KEY = "order_delay";
    
        /**
         * 死信队列
         */
        public static final String DEAD_ORDER_QUEUE_NAME = "dead.ticket.order.queue";
    
        /**
         * 死信队列交换机 DLX，dead letter发送到的 exchange
         */
        public static final String DEAD_ORDER_EXCHANGE_NAME = "dead.ticket.order.exchange";
    
        /**
         * 路由
         */
        public static final String DEAD_ORDER_ROUTING_KEY = "dead.order";
    }

队列配置

@Configuration
    public class DelayRabbitConfig {
        
    
    
        /**
         * 1.死信队列
         */
        @Bean
        public Queue orderQueue() {
        
            return new Queue(DelayConstant.DEAD_ORDER_QUEUE_NAME, true);
        }
    
        /**
         * 2.死信交换机
         * 通过死信交换机把死信消息发送到指定的队列中去
         * 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。
         */
        @Bean
        public TopicExchange orderTopicExchange() {
        
            return new TopicExchange(DelayConstant.DEAD_ORDER_EXCHANGE_NAME);
        }
    
        /**
         * 3.死信队列（绑定交换机）
         */
        @Bean
        public Binding orderBinding() {
        
            // TODO 如果要让延迟队列之间有关联,这里的 routingKey 和 绑定的交换机很关键
            return BindingBuilder.bind(orderQueue()).to(orderTopicExchange()).with(DelayConstant.DEAD_ORDER_ROUTING_KEY);
        }
    
    
        /**
         * 4.延时队列配置
         * <p>
         * 1、第一种方式是直接设置 Queue 延迟时间 但如果直接给队列设置过期时间,这种做法不是很灵活,（当然二者是兼容的,默认是时间小的优先）
         * params.put("x-message-ttl", 5 * 1000);
         * 2、第二种就是每次发送消息动态设置延迟时间,这样我们可以灵活控制
         */
        @Bean
        public Queue delayOrderQueue() {
        
            Map<String, Object> params = new HashMap<>();
            // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，即死信消息转发到那个队列
            params.put("x-dead-letter-exchange", DelayConstant.DEAD_ORDER_EXCHANGE_NAME);
            // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
            params.put("x-dead-letter-routing-key", DelayConstant.DEAD_ORDER_ROUTING_KEY);
            return new Queue(DelayConstant.ORDER_DELAY_QUEUE, true, false, false, params);
        }
    
        /**
         * 5.延时队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。
         * @return DirectExchange
         */
        @Bean
        public DirectExchange orderDelayExchange() {
        
            return new DirectExchange(DelayConstant.ORDER_DELAY_EXCHANGE);
        }
    
        /**
         * 6.延时队列绑定交换机
         */
        @Bean
        public Binding dlxBinding() {
        
            return BindingBuilder.bind(delayOrderQueue()).to(orderDelayExchange()).with(DelayConstant.ORDER_DELAY_ROUTING_KEY);
        }
    }

#### 4.3 创建订单消息 ####

@Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class Order implements Serializable {
        
    
        /**
         * 订单id
         */
        private String orderId;
    
        /**
         * 订单名称
         */
        private String name;
    
    
        /**
         * 订单状态 0：未支付，1：已支付，2：订单已取消
         */
        private Integer orderStatus;
    
        /**
         * 下单时间
         */
        private Date orderTime;
    
        /**
         * 订单金额
         */
        private BigDecimal amount;
    }

#### 4.4 消息生产者 ####

import com.javatv.bean.Order;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /**
     * @description : 消息生产者
     */
    @Component
    @Slf4j
    public class RabbitmqOrderProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void publish(Order order, String messageId, String exchangeName, String key) {
        
            /* 确认的回调 确认消息是否到达 Broker 服务器 其实就是是否到达交换器
             * 如果发送时候指定的交换器不存在 ack 就是 false 代表消息不可达
             */
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        
                log.info("correlationData：{} , ack:{}", correlationData.getId(), ack);
                if (!ack) {
        
                    System.out.println("进行对应的消息补偿机制");
                }
            });
            /* 消息失败的回调
             * 例如消息已经到达交换器上，但路由键匹配任何绑定到该交换器的队列，会触发这个回调，此时 replyText: NO_ROUTE
             */
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
        
                log.info("message:{}; replyCode: {}; replyText: {} ; exchange:{} ; routingKey:{}",
                        message, replyCode, replyText, exchange, routingKey);
            });
            // 在实际中ID 应该是全局唯一 能够唯一标识消息 消息不可达的时候触发ConfirmCallback回调方法时可以获取该值，进行对应的错误处理
            CorrelationData correlationData = new CorrelationData(messageId);
            rabbitTemplate.convertAndSend(exchangeName, key, order, message -> {
        
                /**
                 * 如果配置了 params.put("x-message-ttl", 60 * 1000 * 30);
                 * 那么这一句也可以省略,具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间
                 * 这里为了演示设置为 10 s
                 */
                message.getMessageProperties().setExpiration(1000 * 10 + "");
                return message;
            }, correlationData);
        }
    }

#### 4.5 消息消费者 ####

import com.javatv.bean.Order;
    import com.javatv.constant.DelayConstant;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    
    @Component
    public class DelayConsumer {
        
    
        @RabbitListener(queues = {
        DelayConstant.DEAD_ORDER_QUEUE_NAME})
        public void orderDelayQueue(Order order, Message message, Channel channel) {
        
            System.out.println("###########################################");
            System.out.println("【orderDelayQueue 监听的消息】 - 【消费时间】 - ["
                    +new Date()+"]- 【订单内容】 - ["+order.toString()+"]");
    
            if(order.getOrderStatus() == 0) {
        
                order.setOrderStatus(2);
                System.out.println("【该订单未支付，取消订单】" + order.toString());
            } else if(order.getOrderStatus() == 1) {
        
                System.out.println("【该订单已完成支付】");
            } else if(order.getOrderStatus() == 2) {
        
                System.out.println("【该订单已取消】");
            }
            System.out.println("###########################################");
        }
    }

#### 4.6 测试 ####

我们模拟两条订单消息，如下：

@Test
    public void sendDelay() {
        
        Order order1 = new Order();
        String id1 = String.valueOf(Math.round(Math.random() * 10000));
        order1.setOrderId(id1);
        order1.setOrderStatus(0);
        order1.setName("杭州-北京");
        Order order2 = new Order();
        String id2 = String.valueOf(Math.round(Math.random() * 10000));
        order2.setOrderId(id2);
        order2.setOrderStatus(0);
        order2.setName("北京-深圳");
        orderProducer.publish(order1, id1, DelayConstant.ORDER_DELAY_EXCHANGE, DelayConstant.ORDER_DELAY_ROUTING_KEY);
        orderProducer.publish(order2, id2, DelayConstant.ORDER_DELAY_EXCHANGE, DelayConstant.ORDER_DELAY_ROUTING_KEY);
    }

当发布消息后，我们在客户端去查看数据：

1、刚开始存在于延时队列里面，如下：

![image-20220224194621196][]

2、当消息过期之后，则存在死信队列中，如下：

![image-20220224194712936][]

然后我们在开启消费者服务，监控如下：

![image-20220224195058492][]

#### 4.7 不足之处 ####

在正常情况下，我们等待的时间都是一样的，假如都是 30 分钟，但如果**一个队列存在不同的延时消息**怎么办？

第一种就是不同的时间用不同的队列；

第二种如果是同一队列的话则存在以下问题：

如果我发送两条延时消息，第 1 条延时时间设为 60 s，第二条消息设为 2 s，且第 1 条消息优先发送，常规消息处理应该是第 2 条消息先过期并进入死信队列然后进行消费，但实际情况是 RabbitMQ 只会检查第 1 条消息是否过期，如果过期则丢到死信队列，如果第 1 条消息的延时时长很长，而第二个消息的延时时长很短，则第二个消息并不会优先得到执行。（可自行测试一下，这里不演示）

### 5. 延时队列插件 ###

对于上面存在的问题，如果不能实现在消息粒度上添加 TTL，并使其在设置的 TTL 时间及时死亡，就无法设计成一个通用的延时队列。

#### 5.1 插件安装 ####

RabbitMQ 3.5.7版本以后支持延迟插件，通过插件可以很好的解决上面的问题，进入插件官网：

https://www.rabbitmq.com/community-plugins.html

找到 `rabbitmq_delayed_message_exchange`插件，然后解压放置到RabbitMQ的插件目录。

![image-20220224220256019][]

以前都是一个压缩包，现在直接是解压好的文件，下载之后把`.ez`文件拷贝到 RabbitMQ 安装目录下的 plugins 目录中，如下：

![image-20220224221617367][]

然后进入 RabbitMQ 的安装目录下的 sbin 目录，执行下面命令让该插件生效，然后重启RabbitMQ。

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

执行成功后如下：

![image-20220224221856186][]

#### 5.2 配置队列和交换机 ####

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.CustomExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.QueueBuilder;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class DelayMessageConfig {
        
     
    	public static final String DELAY_EXCHANGE_NAME = "plugin.delay.exchange";
    	
    	public static final String DELAY_QUEUE_NAME = "plugin.delay.queue";
    	
    	public static final String ROUTING_KRY = "plugin.delay.queue";
    	
    	/**
    	 * 声明一个延迟队列
    	 * @return
    	 */
    	@Bean
    	Queue delayQueue(){
        
    		return QueueBuilder.durable(DELAY_QUEUE_NAME).build();
    	}
    
    	/**
    	 * 声明一个交换机
    	 * @return
    	 */
    	@Bean
    	CustomExchange delayExchange(){
        
    		
    		Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
    		return new CustomExchange(DELAY_EXCHANGE_NAME, "x-delayed-message", true,false, args);
    		
    	}
    	/**
    	 * 绑定
    	 * @param delayQueue
    	 * @param delayExchange
    	 * @return
    	 */
    	@Bean
    	Binding queueBinding(Queue delayQueue, CustomExchange delayExchange){
        
    		
    	    return BindingBuilder.bind(delayQueue).to(delayExchange).with(ROUTING_KRY).noargs();
    		
    	}
    }

#### 5.3 消息生产者 ####

import com.javatv.bean.Order;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /**
     * @description : 消息生产者
     */
    @Component
    @Slf4j
    public class RabbitmqDelayOrderProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        /**
         *
         * @param order 消息
         * @param messageId 唯一id
         * @param exchangeName 交换机
         * @param key 路由键
         * @param delayTime 延迟时间
         */
        public void publish(Order order, String messageId, String exchangeName, String key,Integer delayTime) {
        
            /* 确认的回调 确认消息是否到达 Broker 服务器 其实就是是否到达交换器
             * 如果发送时候指定的交换器不存在 ack 就是 false 代表消息不可达
             */
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        
                log.info("correlationData：{} , ack:{}", correlationData.getId(), ack);
                if (!ack) {
        
                    System.out.println("进行对应的消息补偿机制");
                }
            });
            /* 消息失败的回调
             * 例如消息已经到达交换器上，但路由键匹配任何绑定到该交换器的队列，会触发这个回调，此时 replyText: NO_ROUTE
             */
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
        
                log.info("message:{}; replyCode: {}; replyText: {} ; exchange:{} ; routingKey:{}",
                        message, replyCode, replyText, exchange, routingKey);
            });
            // 在实际中ID 应该是全局唯一 能够唯一标识消息 消息不可达的时候触发ConfirmCallback回调方法时可以获取该值，进行对应的错误处理
            CorrelationData correlationData = new CorrelationData(messageId);
            rabbitTemplate.convertAndSend(exchangeName, key, order, message -> {
        
                // 设置延迟时间
                message.getMessageProperties().setDelay(delayTime);
                return message;
            }, correlationData);
        }
    }

#### 5.4 消息消费者 ####

import com.javatv.bean.Order;
    import com.javatv.constant.DelayConstant;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    
    @Component
    public class DelayPluginConsumer {
        
    
        @RabbitListener(queues = {
        "plugin.delay.queue"})
        public void orderDelayQueue(Order order, Message message, Channel channel) {
        
            System.out.println("###########################################");
            System.out.println("【orderDelayQueue 监听的消息】 - 【消费时间】 - ["
                    +new Date()+"]- 【订单内容】 - ["+order.toString()+"]");
    
            if(order.getOrderStatus() == 0) {
        
                order.setOrderStatus(2);
                System.out.println("【该订单未支付，取消订单】" + order.toString());
            } else if(order.getOrderStatus() == 1) {
        
                System.out.println("【该订单已完成支付】");
            } else if(order.getOrderStatus() == 2) {
        
                System.out.println("【该订单已取消】");
            }
            System.out.println("###########################################");
        }
    }

#### 5.5 测试 ####

@Test
    public void sendDelayPlugin() {
        
        Order order1 = new Order();
        order1.setOrderId("1");
        order1.setOrderStatus(0);
        order1.setName("杭州-北京");
        
        Order order2 = new Order();
        order2.setOrderId("2");
        order2.setOrderStatus(0);
        order2.setName("北京-深圳");
        
        // 发送一条延迟 60s 的消息
        delayOrderProducer.publish(order1, "1", DelayPluginConfig.DELAY_EXCHANGE_NAME, DelayPluginConfig.ROUTING_KRY,1000 * 60);
        // 发送一条延迟 5s 的消息
        delayOrderProducer.publish(order2, "2", DelayPluginConfig.DELAY_EXCHANGE_NAME, DelayPluginConfig.ROUTING_KRY,1000 * 5);
    }

执行测试类的 5s 内延迟队列中是没有消息的，如下：

![image-20220224224931433][]

在 5s 之后存在第 1 条消息，如下：

![image-20220224224640228][]

在 60s 之后存在第 2 条消息，如下：

![image-20220224224735212][]

然后我们在开启消费者的情况下，也不会因为延迟消息时间长的消息没进入队列而不消费延时时间短的消息，如下：

![image-20220224225206594][]

### 6. 总结 ###

如果你仔细看了这篇文章，你会发现在自己通过**TTL+DLX**实现延时队列的时候，我把消费之前的消息定义在为延时队列，而过期的消息存入的队列我把它称为了死信队列，因为这条消息是死信消息，但插件定义的延迟队列来放死信消息，其实两者的定义并不冲突，都是可以理解的，我的意思是这样的牛角不要钻。

在学习延时队列的时候也查阅了部分博客，看到了这样一个问题：

> RabbitMQ 的延时任务和消息确认机制冲突吗？即使消息发送成功了，`setReturnCallback` 这个回调还是被触发了。

就我目前演示的版本，`setReturnCallback`是没有被触发的，不知道是否之前的版本有过这样的问题就不得而知了。

源码：https://gitee.com/javatv/advanced-way/tree/master/spring-boot-rabbitmq

参考：https://www.cnblogs.com/mfrank/p/11260355.html

[image-20211019141510180]: https://img-blog.csdnimg.cn/img_convert/6864273bba779da33533aa4463294676.png
[image-20220224192355290]: https://img-blog.csdnimg.cn/img_convert/aaecbe5b03e880b553385199947bbf04.png
[image-20220224194621196]: https://img-blog.csdnimg.cn/img_convert/dd3936f362f98b92e6b89fcfc6d0afef.png
[image-20220224194712936]: https://img-blog.csdnimg.cn/img_convert/3be29db95f0abdeaf0e472d6ef577c7d.png
[image-20220224195058492]: https://img-blog.csdnimg.cn/img_convert/628efe6781aa13c7fcb1ca4b55801357.png
[image-20220224220256019]: https://img-blog.csdnimg.cn/img_convert/7be10de332d0f46963546aedf598e1f8.png
[image-20220224221617367]: https://img-blog.csdnimg.cn/img_convert/47ef6c71bc2537afdfe5ce4ea29b7fa9.png
[image-20220224221856186]: https://img-blog.csdnimg.cn/img_convert/58889efa6e088b9a83f328bd61f3f47f.png
[image-20220224224931433]: https://img-blog.csdnimg.cn/img_convert/0684e8600bae18b2f527a9bf8d9b252d.png
[image-20220224224640228]: https://img-blog.csdnimg.cn/img_convert/6f548ff91e9f7b8333e22c43f9b3906f.png
[image-20220224224735212]: https://img-blog.csdnimg.cn/img_convert/e9a0f4f95c0d866d6e49be64e39c6362.png
[image-20220224225206594]: https://img-blog.csdnimg.cn/img_convert/a390111e337cd52161f585bec35cec35.png