RabbitMQ死信队列与延迟队列

RabbitMQ 死信队列、TTL延迟队列、插件延迟队列

前言

死信队列：是异常处理的利器

死信队列（DLX）是 RabbitMQ 中最核心的异常处理机制，它的价值远不止实现延迟消息：

• 消息重试：消费失败的消息自动进入死信队列，实现延迟重试
• 消息兜底：多次重试仍失败的消息，转人工处理或告警
• 消息审计：统一收集消费失败的消息，便于监控和分析
• 流量削峰：队列满时，溢出消息进入死信队列延迟处理
  生产环境中，死信队列是保障消息可靠性的最后一道防线。

RabbitMQ 的延迟消息方案

RabbitMQ 本身没有原生的延迟队列，但我们可以通过以下三种方案实现：

方案	特点	适用场景
死信队列（DLX）	实现延迟 + 异常处理，生产环境首选	消息重试、兜底、审计、延迟消息
TTL延迟队列	多级TTL队列，支持多种固定延迟时间	延迟时间种类少、要求高可靠性
插件延迟队列	官方插件，支持任意延迟时间	延迟时间灵活、配置简单

---

一、死信队列（Dead Letter Queue）

1.1 什么是死信

死信（Dead Letter）是指消息在队列中无法被正常消费时的状态。RabbitMQ 中有三种情况会产生死信：

1. 消息被拒绝：消费者调用 basicReject 或 basicNack，且 requeue=false
2. 消息 TTL 过期：消息在队列中存活时间超过设置的 TTL
3. 队列达到最大长度：队列设置了 x-max-length，新消息进入时队列已满，头部消息被丢弃
   

1.2 核心配置

配置死信队列的关键在于：业务队列声明时指定死信交换机（deadLetterExchange）和死信路由键（deadLetterRoutingKey）。

@Bean
public Queue bizQueue() {
    return QueueBuilder.durable("biz-queue")
            .deadLetterExchange("dlx-exchange")       // 死信交换机
            .deadLetterRoutingKey("dlx-routing-key")  // 死信路由键
            .ttl(10000)       // TTL 10秒
            .maxLength(5)     // 队列最大长度
            .build();
}

死信交换机和死信队列的声明与普通交换机/队列无异：

@Bean
public DirectExchange deadLetterExchange() {
    return ExchangeBuilder.directExchange("dlx-exchange").build();
}

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("dlx-queue").build();
}

@Bean
public Binding deadLetterBinding(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
    return BindingBuilder.bind(deadLetterQueue)
            .to(deadLetterExchange)
            .with("dlx-routing-key");
}

1.3 消费者：拒绝消息

消费者使用 手动 ACK 模式，根据业务逻辑决定是否拒绝消息：

@RabbitListener(queues = "biz-queue", ackMode = "MANUAL")
public void onMessage(Message message, Channel channel) throws IOException {
    String body = new String(message.getBody());
    long deliveryTag = message.getMessageProperties().getDeliveryTag();

    if (body.contains("拒绝")) {
        // 拒绝消息，requeue=false → 进入死信队列
        channel.basicNack(deliveryTag, false, false);
    } else {
        channel.basicAck(deliveryTag, false);
    }
}

1.4 死信消费者

监听死信队列，处理死信消息：

@RabbitListener(queues = "dlx-queue")
public void onDeadLetterMessage(Message message, Channel channel) throws IOException {
    String body = new String(message.getBody());
    System.out.println("[死信队列] 收到死信消息: " + body);
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

1.5 注意事项

• basicNack 的第三个参数 requeue 必须为 false，否则消息会重新入队而非进入死信队列

---

二、延迟队列：TTL + 死信交换机（统一延迟时间）

2.1 实现原理

利用死信队列的 TTL 机制实现延迟效果：

核心思路：创建一个没有消费者的队列，设置 TTL，配置死信交换机。消息进入该队列后，由于没有消费者消费，TTL 到期后自动通过死信交换机转发到处理队列，由处理队列的消费者消费。

2.2 核心配置

// 延迟队列（无消费者，20秒TTL后进入处理队列）
@Bean
public Queue delayQueue() {           
    return QueueBuilder.durable("delay-queue")
            .ttl(20000)                                          // 20秒TTL
            .deadLetterExchange("delay-process-exchange")        // 死信交换机
            .deadLetterRoutingKey("delay-process-routing-key")   // 死信路由键
            .build();
}

@Bean
public DirectExchange delayExchange() {    // 生产者第一次发送消息到达的交换机
    return ExchangeBuilder.directExchange("delay-exchange").build();
}

@Bean
public Binding delayBinding(Queue delayQueue, DirectExchange delayExchange) {       
    return BindingBuilder.bind(delayQueue).to(delayExchange).with("delay-routing-key");
}

// 处理队列（消费者监听这里）
@Bean
public Queue delayProcessQueue() {             //延迟后的消息放到这里迅速被消费者消费
    return QueueBuilder.durable("delay-process-queue").build();
}

@Bean
public DirectExchange delayProcessExchange() {     //延迟队列到达时间后消息进入这个交换机然后进入处理队列
    return ExchangeBuilder.directExchange("delay-process-exchange").build();
}

@Bean
public Binding delayProcessBinding(Queue delayProcessQueue, DirectExchange delayProcessExchange) {
    return BindingBuilder.bind(delayProcessQueue)
            .to(delayProcessExchange)
            .with("delay-process-routing-key");
}

2.3 生产者与消费者

// 生产者：发送消息到延迟队列
@GetMapping("/send")
public String send() {
    String now = LocalDateTime.now().format(FORMATTER);
    String message = "延迟消息，发送时间 " + now + "，预计20秒后被消费";
    rabbitTemplate.convertAndSend(
            DelayQueueConfig.DELAY_EXCHANGE,
            DelayQueueConfig.DELAY_ROUTING_KEY,
            message);
    return "消息已发送（" + now + "），约20秒后到达处理队列";
}

// 消费者：监听处理队列
@RabbitListener(queues = DelayQueueConfig.DELAY_PROCESS_QUEUE)
public void onMessage(String message) {
    String now = LocalDateTime.now().format(FORMATTER);
    System.out.println("[延迟队列消费者] 消费时间 " + now + "，消息内容: " + message);
}

2.4 优缺点分析

优点：

• 无需安装任何插件，RabbitMQ 原生支持
• 实现简单，原理清晰

缺点：

• 延迟时间固定：所有消息的延迟时间相同（队列级别 TTL）
• 多种延迟时间需要多个队列：比如要支持 5分钟、10分钟、30分钟的延迟，就需要创建 3 个延迟队列
• 消息堆积：延迟队列无消费者，大量消息堆积在内存/磁盘中

---

三、插件延迟队列：rabbitmq_delayed_message_exchange

3.1 实现原理

RabbitMQ 官方提供了 rabbitmq_delayed_message_exchange 插件，支持为每条消息单独设置延迟时间。

3.2 安装插件

#查看插件列表
rabbitmq-plugins list

选择一个相近的版本下载
插件下载地址

在cd /usr/lib/rabbitmq/plugins目录下导入,没有路径就mikdir /usr/lib/rabbitmq/plugins
插件上传参考

#查看插件列表
rabbitmq-plugins list

#启动插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

#重启服务
service rabbitmq-server restart

3.3 核心配置

@Bean
public DirectExchange pluginDelayExchange() {
    return ExchangeBuilder
            .directExchange("plugin-delay-exchange")
            .durable(true)
            .delayed()    // 关键：声明为延迟交换机
            .build();
}

@Bean
public Queue pluginDelayQueue() {
    return QueueBuilder.durable("plugin-delay-queue").build();
}

@Bean
public Binding pluginDelayBinding(Queue pluginDelayQueue, DirectExchange pluginDelayExchange) {
    return BindingBuilder.bind(pluginDelayQueue)
            .to(pluginDelayExchange)
            .with("plugin-delay-routing-key");
}

ExchangeBuilder.directExchange().delayed().build() 底层会将交换机类型设为 x-delayed-message，比手动创建 CustomExchange 更简洁。

3.4 生产者：每条消息独立延迟

@GetMapping("/send")
public String send(@RequestParam(defaultValue = "5000") long delayMs) {
    String now = LocalDateTime.now().format(FORMATTER);
    String message = "插件延迟消息，发送时间 " + now + "，延迟 " + delayMs + "ms";

    rabbitTemplate.convertAndSend(
            PluginDelayQueueConfig.PLUGIN_DELAY_EXCHANGE,
            PluginDelayQueueConfig.PLUGIN_DELAY_ROUTING_KEY,
            message,
            msg -> {
                msg.getMessageProperties().setDelayLong(delayMs); // 每条消息独立延迟
                return msg;
            });

    return "消息已发送（" + now + "），约 " + delayMs + " 毫秒后被消费";
}

Spring AMQP 2.x+ 提供了 setDelayLong() 方法，旧版本需手动设置 header：msg.getMessageProperties().setHeader("x-delay", delayMs)。

3.5 消费者

@RabbitListener(queues = PluginDelayQueueConfig.PLUGIN_DELAY_QUEUE)
public void onMessage(String message) {
    String now = LocalDateTime.now().format(FORMATTER);
    System.out.println("[插件延迟队列消费者] 消费时间 " + now + "，消息内容: " + message);
}

3.6 优缺点分析

类型	说明
优点	每条消息可设置不同的延迟时间，灵活性极高
	配置简单，只需一个交换机 + 一个队列
缺点	需要安装并启用 RabbitMQ 插件
	插件内部会将延迟消息持久化到磁盘，有一定 IO 开销
	插件版本需与 RabbitMQ 版本兼容

四、三种方案对比

维度	死信队列	TTL+死信延迟队列	插件延迟队列
实现复杂度	低	中（需延迟队列+处理队列）	低（一个交换机+一个队列）
延迟时间	无延迟（消息路由机制）	统一延迟（队列级别TTL）	按消息设置（x-delay）
是否需要插件	否	否	是
支持多种延迟时间	不适用	需要多个队列（每种TTL一个）	天然支持
消息堆积处理	依赖队列配置	延迟队列无消费者，天然堆积	插件内部磁盘存储
适用场景	消息过滤、异常处理	固定延迟（如统一30分钟超时）	动态延迟（如不同用户不同超时）
性能	高	中等	中等（有磁盘IO）
RabbitMQ 版本要求	无	无	需 3.8+

---

五、总结与选型建议

1. 死信队列：是 RabbitMQ 的基础机制，主要用于消息的异常处理和路由。本身不提供延迟功能，但为延迟队列奠定了基础。

2. TTL + 死信延迟队列：适合延迟时间固定的场景。比如电商订单超时30分钟自动取消，所有订单延迟时间一样，用这种方案最合适。

3. 插件延迟队列：适合延迟时间动态变化的场景。比如不同会员等级有不同的超时时间，每条消息的延迟不同，用插件方案最灵活。

一句话选型：

• 延迟时间固定 → TTL + 死信
• 延迟时间动态 → 插件方案

---

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2301_79245606/article/details/162851079