RabbitMQ 延迟消息:DLX 与延迟插件的取舍

一、从订单超时取消说起

电商里有个绕不开的需求:下单后 30 分钟未支付,自动关闭订单并释放库存。

需求本身很简单,难的是「30 分钟后」这个时间点怎么触发。常见的三种做法:

方案	机制	问题
定时任务轮询	每分钟扫一次 `status=待支付 AND create_time < NOW()-30min`	时间精度受扫描周期限制;数据量大时全表扫描压力大;任务单点
延迟消息	下单时发一条消息,30 分钟后投递,消费者收到就关单	依赖 MQ 支持延迟投递
时间轮(内存)	应用内维护时间轮	重启丢任务,多实例需协调

延迟消息是最贴合语义的一种:「30 分钟后提醒我一下」这件事,本来就应该由消息中间件承担。 它把定时逻辑从业务里彻底移走,不用扫描、不用分片、天然支持多消费者。

问题在于:RabbitMQ 原生并不支持延迟消息。 没有 delay 参数,没有延迟队列类型。要用它做延迟投递,只有两条路——死信队列(DLX),或者延迟插件。

二、DLX 方案:用「过期」实现「延迟」

原理

DLX 方案的巧妙之处在于:它没有实现延迟,而是借用了消息过期(TTL)机制。

回忆一下死信队列的触发条件——消息变成死信有三种原因:

消费者 basicNack / basicReject 且 requeue=false;
消息 TTL 过期;
队列长度超限,队头消息被挤掉。

第 2 条就是延迟的入口。一条消息在队列里放够 TTL 时间、没人消费,它就会"过期"变成死信;而死信会被转发到指定的死信交换机,再路由到死信队列。 于是:

让消息在 A 队列里等 30 分钟 → 过期 → 自动死信转发到 B 队列 → 消费者监听 B 队列

对消费者来说,它就是「30 分钟后收到了这条消息」。

配置

// 1. 声明延迟用的死信交换机与真正的消费队列
channel.exchangeDeclare("order.dlx.exchange", "direct", true);
channel.queueDeclare("order.dlx.queue", true, false, false, null);
channel.queueBind("order.dlx.queue", "order.dlx.exchange", "order.timeout");

// 2. 声明"等待队列":没有消费者,只负责让消息待够 TTL
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "order.dlx.exchange");     // 过期后去哪
args.put("x-dead-letter-routing-key", "order.timeout");       // 重写 routing key
args.put("x-message-ttl", 30 * 60 * 1000);                    // 队列级 TTL:30 分钟
channel.queueDeclare("order.delay.queue", true, false, false, args);
channel.queueBind("order.delay.queue", "order.delay.exchange", "order.create");

// 3. 下单成功:只发到延迟队列,没有消费者会动它
channel.basicPublish("order.delay.exchange", "order.create", null, body);

关键点是 order.delay.queue 上不能有任何消费者。一旦有人消费,消息在过期前就被取走了,延迟效果消失。

队列级 TTL vs 消息级 TTL

TTL 可以设在两个地方,差别很大:

	队列级 TTL(`x-message-ttl`)	消息级 TTL(`expiration`)
设置位置	声明队列时的参数	每条消息的属性
影响范围	队列中所有消息	该条消息
灵活性	一个队列只能有一个延迟时长	每条消息可以不同
是否阻塞	是(见下文)	是,而且更严重

如果延迟时长需要按业务变化(比如「普通订单 30 分钟、秒杀订单 5 分钟」),队列级 TTL 就不够用了,只能建多个队列,或者改用消息级 TTL:

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .expiration("1800000")        // ⚠️ 字符串,单位毫秒
        .deliveryMode(2)
        .build();
channel.basicPublish("order.delay.exchange", "order.create", props, body);

注意 expiration 是字符串类型,而且必须显式转成字符串,写成 setExpiration(1800000) 会编译不过。这是新手常见的第一个坑。

队头阻塞:DLX 方案的致命问题

只要有不同延迟时长的消息进同一个队列,就会遇到这个问题:

RabbitMQ 的队列是严格 FIFO 的,消息只能从队头出队。队列中前面的消息还没过期,后面的消息即使已经到期,也必须排队等着。

举个例子,同一个延迟队列里先后进来两条消息:

时刻 T+0:    消息 A 入队,TTL = 30 分钟
时刻 T+1:    消息 B 入队,TTL = 5 分钟      ← B 应该在 T+6 到期

到 T+6 时,B 已经过期了,但它前面还压着 A。RabbitMQ 检查队头发现 A 还没到期,整条队列就停在这里。B 必须等到 T+30、A 出队之后,才轮到自己出队。

结果:B 实际延迟了 29 分钟,而不是 5 分钟。

这个偏差不是几毫秒的抖动,而是「短延迟消息被长延迟消息无限拖后」。哪怕只有一条 24 小时的消息堵在队头,后面所有 5 分钟的消息都会延迟近 24 小时。

所以 DLX 方案的使用前提非常明确:

场景	是否适用 DLX
所有消息延迟时长相同(如全部 30 分钟)	✅ 适用,队头阻塞不存在
延迟时长不同,但分别用不同的队列	✅ 适用,每个队列一个 TTL
延迟时长不同且混在同一个队列	❌ 不适用,必须换方案

第三条是很多实现翻车的地方——压测时只发同一种延迟的消息,自然测不出问题;上线后不同业务共用一个延迟队列,才开始出现「说好 5 分钟,实际 40 分钟」的投诉。

三、延迟插件方案:x-delayed-message

原理

rabbitmq-delayed-message-exchange 是一个官方插件,它引入了一种新的交换机类型 x-delayed-message。

与 DLX 完全不同,插件方案不依赖队列和 TTL,而是在交换机内部维护延迟逻辑:

生产者 ──带 x-delay=1800000──→ [x-delayed-message 交换机]
                                      │  消息暂存在交换机内部(基于 Mnesia)
                                      │  到时间后
                                      ▼
                                  正常路由到队列 → 消费者

消息到达交换机后,交换机不会立刻路由,而是根据 x-delay 头把消息存在自己的存储里,到期后再走正常的路由流程。因为延迟判断发生在交换机层、而不是队列里,每条消息的延迟完全独立,不存在队头阻塞。

使用

安装插件后需重启节点:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

声明交换机时类型写 x-delayed-message,并用 x-delayed-type 指定实际的路由类型(direct / topic / fanout):

Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");        // 延迟之外的真正路由方式
channel.exchangeDeclare("order.delayed.exchange",
        "x-delayed-message", true, false, args);

channel.queueDeclare("order.timeout.queue", true, false, false, null);
channel.queueBind("order.timeout.queue", "order.delayed.exchange", "order.timeout");

发送时在 headers 里带上 x-delay,单位毫秒:

Map<String, Object> headers = new HashMap<>();
headers.put("x-delay", 30 * 60 * 1000);

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .headers(headers)
        .deliveryMode(2)
        .build();

channel.basicPublish("order.delayed.exchange", "order.timeout", props, body);

注意这里有个设计上的不一致:x-delay 是整数毫秒,而 DLX 的 expiration 是字符串毫秒。 两者没有对齐,迁移时容易写错。

两种方案对比

维度	DLX(死信队列)	延迟插件
队头阻塞	有,同队列不同延迟会互相拖累	无,每条消息独立计时
延迟精度	受限,取决于队头	较高
是否需要装插件	不需要,原生能力	需要安装并重启节点
运维成本	零	集群每个节点都要装,版本必须与 RabbitMQ 兼容
延迟存储	Broker 的普通队列(消息在队列里等)	交换机内部存储(基于 Mnesia 表)
高延迟消息	长期占用队列空间	也占用存储,但只影响延迟交换机
消息量级	队列堆积大量等待消息,影响内存水位	大量延迟消息同样会占用存储和内存
适用场景	统一延迟时长、不想动运维	多种延迟时长、对精度有要求

两条实践结论:

延迟时长统一时,老老实实用 DLX。 它不需要改运维、不引入插件依赖,而且效果完全等价。
延迟时长多样、或有高延迟(小时级以上)消息时,用插件。 但要提前确认集群版本兼容性,并且把「插件安装」写进部署文档——这是一个很容易在扩容新节点时被遗漏的步骤,新节点没装插件会导致交换机声明失败。

另外,插件方案还有一个容易被忽略的边界:延迟消息在交换机内部的存储不是无限的,大量长时间延迟的消息会推高内存和磁盘占用。用延迟插件投递「7 天后执行」的任务时,要评估同时存在的消息总量。

四、消费者收到消息后必须查库

两种方案都配好了,消息也准时到了,接下来是最容易出错的一步。

一个直观但错误的实现是:

@RabbitListener(queues = "order.dlx.queue")
public void onTimeout(Message message, Channel channel) throws IOException {
    OrderEvent event = JSON.parseObject(message.getBody(), OrderEvent.class);

    // ❌ 直接按消息内容关单
    orderMapper.closeOrder(event.getOrderNo());
    stockService.release(event.getOrderNo());
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

这段代码的问题在于:它把消息当成"事实",而不是"提醒"。

消息里携带的是下单那一刻的快照。从发送到投递的 30 分钟里,世界已经变了:

情况	消息说的	数据库里的真实状态
用户已经支付	待支付	已支付
用户已主动取消	待支付	已取消
消息重复投递	待支付	已取消(第一条已处理)
消息提前到达	待支付	待支付,但还没到超时时间

第一条最严重:用户付了钱,订单被消息关掉了。 紧接着 release 释放库存,可能导致超卖——一次误操作同时造成资损和超卖。

所以正确的做法是:收到消息后先查数据库,以数据库状态为准做判断。

@RabbitListener(queues = "order.dlx.queue")
public void onTimeout(Message message, Channel channel) throws IOException {
    long tag = message.getMessageProperties().getDeliveryTag();
    OrderEvent event = JSON.parseObject(message.getBody(), OrderEvent.class);

    try {
        // 1. 查库:数据库是业务的事实源
        Order order = orderMapper.selectByOrderNo(event.getOrderNo());
        if (order == null) {
            channel.basicAck(tag, false);           // 订单不存在,无需处理
            return;
        }

        // 2. 校验状态:只有仍然"待支付"且确实已超时才关闭
        if (!"WAITING_PAY".equals(order.getStatus())) {
            log.info("订单状态已变更,忽略超时消息 orderNo={} status={}",
                    order.getOrderNo(), order.getStatus());
            channel.basicAck(tag, false);           // ✅ 直接确认,不要重新入队
            return;
        }
        if (System.currentTimeMillis() - order.getCreateTime().getTime()
                < 30 * 60 * 1000L) {
            // 消息提前到达(时钟偏差等),重新入队稍后再试,或直接忽略
            channel.basicNack(tag, false, false);
            return;
        }

        // 3. 状态合法,执行关单 + 释放库存
        orderService.closeTimeoutOrder(order.getOrderNo());

        channel.basicAck(tag, false);
    } catch (Exception e) {
        channel.basicNack(tag, false, true);
    }
}

这里有三个设计要点:

第一,查库本身就是幂等保障。 订单已经是「已取消」时,第二次收到同样的超时消息,第一个 if 判断就会拦住它。不需要额外的去重表——用"当前状态是否允许这次操作"来判重,是成本最低的幂等实现。 当然,如果需要更严格的保证(比如跨多个服务),可以再叠加一个基于订单号的分布式唯一标识。

第二,关单和释放库存必须在同一个事务里。 否则可能出现「订单关了、库存没释放」,库存被永久锁定。这也是为什么释放库存应该由 orderService 内部的本地事务完成,而不是在消费者里调两次远程接口。

第三,closeTimeoutOrder 内部仍然要带条件更新。

UPDATE `order`
SET status = 'CANCELED', cancel_time = NOW()
WHERE order_no = ?
  AND status = 'WAITING_PAY';        -- 条件更新,防止并发下的状态覆盖

因为第 1 步的查询和第 3 步的更新之间存在时间窗口,用户可能恰好在这个窗口里完成了支付。加上 AND status = 'WAITING_PAY' 后,支付成功的那条更新先执行,关单这条就会影响 0 行,业务代码据此判断「关单失败,不做后续库存释放」。这是查库校验之外的第二道防线,不能省。

五、小结

RabbitMQ 原生不支持延迟消息,只有 DLX 和延迟插件两条路
DLX 的原理是「队列级/消息级 TTL 过期 → 变死信 → 转发到死信队列」,延迟队列上不能有任何消费者
DLX 的硬伤是队头阻塞:同队列内短延迟消息会被前面的长延迟消息拖住,只有延迟时长统一(或按 TTL 分队列)时才适用
延迟插件用 x-delayed-message 交换机在交换机内部计时,每条消息独立,没有队头阻塞,代价是要装插件、动运维、注意集群版本兼容
expiration 是字符串毫秒,x-delay 是整数毫秒,两者不通用
消费者收到超时消息后必须查库:消息是下单时的旧快照,数据库才是事实源;不查库会把已支付的订单关掉
查库校验 + 条件更新(WHERE status = 'WAITING_PAY')构成两道幂等防线,后者用来覆盖「查询与更新之间的支付窗口」