GU𝐆𝐫𝐨𝐰 𝐔𝐩
返回文章列表
高并发项目
MQRabbitMQ延迟消息死信队列

09MQ延迟消息

1.延迟消息的含义与作用

  • 延迟消息：发送者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。
  • 延迟任务：设置在一定时间之后才执行的任务。

以电商下单业务为例，交易服务下单完成后扣减库存，但用户未必会支付。若用户长时间不付款，会长期占用资源，影响其他交易。解决方案是设置支付超时时间（如 30 分钟），超时未支付则取消订单并恢复库存。

延迟消息实现订单超时检查

交易服务下单后保存订单、扣减库存，并向 MQ 发送一条延迟消息（如延迟 30 分钟）；30 分钟后交易服务收到消息，再去查询订单支付状态，据此决定是否取消订单并恢复库存。

实现延迟消息有两种常见方法：死信交换机和延迟消息插件。

2.死信交换机

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

  • 消费者使用 basic.reject 或 basic.nack 声明消费失败，并且消息的 requeue 参数设置为 false。
  • 消息是一个过期消息（达到了队列或消息本身设置的过期时间），超时无人消费。
  • 要投递的队列消息堆积满了，最早的消息可能成为死信。

如果队列通过 dead-letter-exchange 属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中。这个交换机称为死信交换机（Dead Letter Exchange，简称 DLX）。

死信交换机的作用：

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

实现延迟消息的思路是：让消息先进入一个设置了 TTL 和 dead-letter-exchange 的队列（消费者不监听它），消息过期后成为死信，被投递到死信交换机，再路由到真正被消费的队列中。例如设置消息 TTL 为 30 秒，相当于让消息延迟 30 秒后才到达消费者。

死信交换机需要自行维护 TTL 与绑定关系，使用较为繁琐，因此更推荐使用延迟消息插件。

3.延迟消息插件

3.1延迟消息插件原理

这个插件可以将普通交换机改造为支持延迟消息功能的交换机，当消息投递到交换机后可以暂存一定时间，到期后再投递到队列。相比死信交换机，不再需要为每类延迟时长单独建立带 TTL 的队列，配置更简洁。

3.2声明延迟交换机

使用 @RabbitListener 注解声明延迟交换机和队列：

@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);
}

也可以基于 ExchangeBuilder 声明：

@Bean
public DirectExchange delayExchange(){
    return ExchangeBuilder
            .directExchange("delay.direct")
            .delayed() // 设置delay的属性为true
            .durable(true) // 持久化
            .build();
}

3.3发送延迟消息

发送消息时需要通过消息头 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;
        }
    });
}

4.超时取消订单

4.1业务流程

用户下单完成后，发送 15 分钟延迟消息，在 15 分钟后接收消息，检查支付状态：

  • 已支付：更新订单状态为已支付。
  • 未支付：更新订单状态为关闭订单，恢复商品库存。

取消超时订单流程

整体流程是：下单业务创建订单后发送延迟消息；15 分钟后延迟消息处理逻辑查询支付状态，若已支付则标记为已支付，否则取消订单并恢复库存。

4.2固定延迟的不足

设置 30 分钟后检测订单支付状态，实现起来非常简单，但是存在两个问题：

  • 如果并发较高，30 分钟可能堆积消息过多，对 MQ 压力很大。
  • 大多数订单在下单后 1 分钟内就会支付，但是却需要在 MQ 内等待 30 分钟，浪费资源。

固定 30 分钟延迟存在的问题

4.3多级延迟优化

针对固定延迟的问题，可以采用多级延迟方案：根据未支付的时长，使用不同的延迟时间再次发送延迟消息。例如依次使用 10s、10s、10s、15s、15s、30s、30s、60s、60s、2m、5m、10m、10m 等递增的延迟时间检查订单状态。

改造后的流程：

  1. 下单业务创建订单后，发送一条初始延迟消息（如延迟 10s）。
  2. 延迟消息处理逻辑收到消息后，查询支付状态。
  3. 若已支付，则标记订单为已支付，流程结束。
  4. 若未支付，根据当前是否还有下次延迟时间：
    • 有：再次发送延迟消息，等待下一个时间点检查。
    • 无：取消订单并恢复库存，流程结束。

这样既能尽早发现已支付的订单，避免长时间占用 MQ 资源，又能控制单条消息的最大延迟时长，减轻 MQ 压力。