Skip to content
中间件
RabbitMQ
死信与延迟队列

死信与延迟队列

死信队列（DLQ）的四种触发条件、完整架构链路、Policy 配置方式与毒丸消息处理；以及延迟队列的两种实现方案——TTL + DLX 组合与延迟消息插件。

死信队列（Dead Letter Queue, DLQ）

什么是死信队列

死信队列是 RabbitMQ 中的重要容错机制。当消息在正常队列中无法被成功消费时，会被自动转发到一个专门的"死信"交换机和队列，以便进行后续的分析、重试或告警。

消息成为"死信"的四种情况：

触发条件 说明 触发场景
消息被拒绝 消费者使用 Requeue=false 拒绝消息 消息格式错误、业务规则校验失败
消息 TTL 过期 消息在队列中存活超过设定的 TTL 超时未处理的订单、过期通知
队列溢出 队列中的消息数量超过最大长度限制 流量突增、消费者处理过慢
投递次数超限 Quorum Queue 中消息投递次数超过 delivery-limit 毒丸消息反复导致消费者崩溃

死信队列架构

先理解正常流程和死信流程的完整链路——死信不是一个独立的功能，而是在正常队列上"挂载"了一个兜底出口：

                          ┌──────────────────────┐
                          │     正常消费链路       │
                          │                      │
  生产者 ──→ 业务交换机 ──→ 业务队列 ──→ 消费者     │
            (order.ex)   (order.q)    ┌──────────┐│
                                      │ 成功 → Accept
                                      │ 失败 → Requeue(重试)
                                      │ 格式错误 → Discard ───┐
                                      └──────────┘           │
                          └──────────────────────┘           │
                                                             ▼
                          ┌──────────────────────┐    ┌──────────────┐
                          │      死信链路（兜底）  │    │   死信交换机   │
                          │                      │    │  (dlx.order)  │
                                     消息 TTL 过期 ─────→│              │
                                     队列溢出 ──────────→│              │
                                    投递次数超限 ───────→│              │
                          │                      │    └──────┬───────┘
                          └──────────────────────┘           │
                                                             ▼
                                                     ┌──────────────┐
                                                     │   死信队列     │
                                                     │ (dlq.order)   │
                                                     │              │
                                                     │ 消费者分析    │
                                                     │ 告警/人工处理  │
                                                     └──────────────┘

关键理解：

  1. 正常链路：生产者 → 交换机 → 业务队列 → 消费者。消费者成功处理就 Accept，临时失败就 Requeue 重试，认为消息无价值就 Discard 丢弃。
  2. 死信链路：业务队列在声明时配置了两个关键属性——x-dead-letter-exchange（死信交换机名）和 x-dead-letter-routing-key（可选的路由键）。当消息满足死信条件时，Broker 自动将该消息从业务队列取出，重新发布到指定的死信交换机，死信交换机再路由到绑定的死信队列。
  3. 四种触发条件不是互斥的——比如一条消息既超过了 TTL 又导致队列溢出，以先发生的为准。

死信消息里多了什么：进入 DLQ 的消息会携带额外的元数据（x-death 头），记录了"它从哪个队列来的、为什么死、死了几次、什么时候死的"，方便排查。

配置死信队列（AMQP 1.0 Go）

在 AMQP 1.0 Go 客户端中，队列级别的死信配置通过消息注解（Message Annotations）和队列声明参数传递：

// ─── 步骤 1：声明死信交换机和死信队列 ───
err := conn.Management().DeclareExchange(ctx,
    &rmq.DirectExchangeSpecification{
        Name:      "dlx.order",
        IsDurable: true,
    })
if err != nil {
    log.Fatalf("声明死信交换机失败: %v", err)
}

// 声明死信队列
dlqInfo, err := conn.Management().DeclareQueue(ctx,
    &rmq.QuorumQueueSpecification{
        Name: "dlq.order.processing",
    })
if err != nil {
    log.Fatalf("声明死信队列失败: %v", err)
}

// 绑定死信队列到死信交换机
err = conn.Management().BindQueue(ctx, &rmq.BindingSpecification{
    SourceExchange:   "dlx.order",
    DestinationQueue: "dlq.order.processing",
    BindingKey:       "order.dlq",  // 死信路由键
})
if err != nil {
    log.Fatalf("绑定死信队列失败: %v", err)
}

// ─── 步骤 2：声明业务队列，关联死信交换机 ───
// AMQP 1.0 中，通过 QuorumQueueSpecification 的参数传递 DLX 配置
// 具体字段取决于库版本的 QueueSpecification 接口暴露程度
// 如果库不支持直接传参，推荐使用 RabbitMQ Policy 方式配置

// 方式 A：通过消息注解声明（当库支持 QueueSpecification 参数时）
queueInfo, err := conn.Management().DeclareQueue(ctx,
    &rmq.QuorumQueueSpecification{
        Name: "order.queue",
        // 部分版本支持以下参数（取决于库版本）:
        // DeadLetterExchange:   "dlx.order",
        // DeadLetterRoutingKey: "order.dlq",
        // MessageTTL:           30000,  // 30 秒
        // MaxLength:            10000,  // 最大消息数
        // DeliveryLimit:        5,      // 投递次数限制（Quorum Queue）
    })

推荐方式：使用 RabbitMQ Policy

RabbitMQ 强烈建议使用 Policy 而非在代码中硬编码死信配置，因为 Policy 可以动态更新无需重新部署：

# 通过命令行设置 Policy
rabbitmqctl set_policy DLX-POLICY "^order\." \
  '{"dead-letter-exchange":"dlx.order", "dead-letter-routing-key":"order.dlq", "delivery-limit":5}' \
  --apply-to queues --priority 7
// Go 代码中只需要正常声明队列即可，死信配置由 Policy 自动关联
queueInfo, err := conn.Management().DeclareQueue(ctx,
    &rmq.QuorumQueueSpecification{
        Name: "order.queue", // 匹配 "^order\." 模式的 Policy 自动生效
    })

死信消费者（处理失败消息） 

// 消费死信队列中的消息，进行分析、告警或人工处理
dlqConsumer, err := conn.NewConsumer(ctx, "dlq.order.processing",
    &rmq.ConsumerOptions{InitialCredits: 1})
if err != nil {
    log.Fatalf("创建 DLQ 消费者失败: %v", err)
}
defer dlqConsumer.Close(context.Background())

for {
    delivery, err := dlqConsumer.Receive(ctx)
    if err != nil {
        if errors.Is(err, context.Canceled) {
            return
        }
        log.Printf("DLQ 接收错误: %v", err)
        continue
    }

    msg := delivery.Message()
    body := string(msg.Data()[0])

    // 读取死亡信息（AMQP 1.0 使用 x-opt-deaths 注解）
    // 其中记录了死亡原因、原队列、死亡次数等
    log.Printf("[DLQ] 死信消息: %s", body)
    log.Printf("[DLQ] 原因: %v", msg.MessageAnnotation("x-opt-death-reason"))

    // 根据死亡原因分类处理
    // - "rejected": 消息被消费者明确拒绝
    // - "expired": 消息 TTL 过期
    // - "delivery_limit": 投递次数超限（毒丸消息）
    // - "maxlen": 队列溢出

    // 处理后确认（避免 DLQ 堆积）
    delivery.Accept(ctx)
}

毒丸消息处理（Delivery Limit）

Quorum Queue 特有的 delivery-limit 参数可以自动检测"毒丸消息"——反复导致消费者崩溃的消息：

// 消费者中正确处理毒丸消息
for {
    delivery, _ := consumer.Receive(ctx)
    msg := delivery.Message()

    // 检查投递次数
    deliveryCount := msg.DeliveryCount() // AMQP 1.0 内置字段

    err := processMessage(msg)
    if err != nil {
        log.Printf("处理失败 (第 %d 次投递): %v", deliveryCount, err)

        // 投递次数接近 delivery-limit 时，主动丢弃避免循环
        if deliveryCount >= 4 {
            log.Printf("毒丸消息，主动丢弃: %s", string(msg.Data()[0]))
            delivery.Discard(ctx) // 丢弃，不重新入队
        } else {
            delivery.Requeue(ctx) // 重新入队等待重试
        }
        continue
    }

    delivery.Accept(ctx)
}

延迟队列（Delayed Queue）

两种实现方案

RabbitMQ 本身不直接支持延迟队列，但可以通过以下两种方式实现：

方案 原理 优点 缺点
TTL + DLX 组合 消息 TTL 过期后自动转发到死信队列 原生支持，无需插件 延迟精度取决于 TTL 设置，无法灵活修改
延迟消息插件 使用 rabbitmq_delayed_message_exchange 插件 灵活，每条消息可设不同延迟 需要安装额外插件

方案一：TTL + DLX 组合（Go 实现） 

// ─── 架构说明 ───
// 生产者 → delay.order.queue (TTL 30秒，无消费者)
//       → (过期后) → dlx.order → dlq.order.queue (消费者实际消费)

func setupDelayedQueue(ctx context.Context, conn *rmq.Connection) error {
    // 1. 声明死信交换机
    err := conn.Management().DeclareExchange(ctx,
        &rmq.DirectExchangeSpecification{
            Name:      "dlx.order",
            IsDurable: true,
        })
    if err != nil {
        return err
    }

    // 2. 声明实际消费队列（死信队列）
    _, err = conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{
            Name: "order.process.queue",
        })
    if err != nil {
        return err
    }

    // 3. 绑定
    err = conn.Management().BindQueue(ctx, &rmq.BindingSpecification{
        SourceExchange:   "dlx.order",
        DestinationQueue: "order.process.queue",
        BindingKey:       "order.process",
    })
    if err != nil {
        return err
    }

    // 4. 声明延迟队列（核心：设置 TTL + 死信转发）
    // 这个队列没有消费者，消息到期后自动转发
    _, err = conn.Management().DeclareQueue(ctx,
        &rmq.QuorumQueueSpecification{
            Name: "delay.order.30s",
            // 注意：具体参数名取决于库版本
            // DeadLetterExchange:   "dlx.order",
            // DeadLetterRoutingKey: "order.process",
            // MessageTTL:           30000,
        })
    if err != nil {
        return err
    }

    return nil
}

// 生产者：发送延迟消息
func sendDelayedMessage(ctx context.Context, publisher *rmq.Publisher,
    body []byte, delayMs int) error {

    // 发送到延迟队列（不是最终消费队列！）
    msg := rmq.NewMessage(body)
    msg.SetContentType("application/json")

    // 消息级别 TTL（覆盖队列级别 TTL）
    // 注意：TTL 单位是毫秒
    msg.SetMessageAnnotation("x-message-ttl", delayMs)

    // 也可以直接设置消息 TTL
    msg.SetTTL(delayMs)

    result, err := publisher.Publish(ctx, msg)
    if err != nil {
        return err
    }

    switch result.Outcome.(type) {
    case *rmq.StateAccepted:
        return nil
    default:
        return fmt.Errorf("延迟消息发布失败: %v", result.Outcome)
    }
}

// 消费者：监听实际处理队列
func consumeDelayedMessages(ctx context.Context, conn *rmq.Connection) error {
    consumer, err := conn.NewConsumer(ctx, "order.process.queue",
        &rmq.ConsumerOptions{InitialCredits: 5})
    if err != nil {
        return err
    }
    defer consumer.Close(context.Background())

    for {
        delivery, err := consumer.Receive(ctx)
        if err != nil {
            if errors.Is(err, context.Canceled) {
                return nil
            }
            log.Printf("接收错误: %v", err)
            continue
        }

        msg := delivery.Message()
        body := string(msg.Data()[0])

        // 检查消息的原始 TTL（如果设置了的话）
        log.Printf("收到延迟消息: %s (原始TTL: %dms)", body, msg.TTL())
        processOrder(body)

        delivery.Accept(ctx)
    }
}

方案二：延迟消息插件（需要先安装插件） 

# 安装延迟消息插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
// 声明 x-delayed-message 类型交换机
// 注意：AMQP 1.0 Go 客户端中，声明方式取决于库版本
// 如果库不直接支持自定义 exchange type，需要通过底层连接发送 AMQP 帧
func setupDelayedExchange(ctx context.Context, conn *rmq.Connection) error {
    // 方式一：如果库支持自定义 exchange 类型
    // err := conn.Management().DeclareExchange(ctx,
    //     &rmq.CustomExchangeSpecification{
    //         Name:       "delayed.exchange",
    //         Type:       "x-delayed-message",
    //         IsDurable:  true,
    //         IsAutoDelete: false,
    //         Arguments: map[string]interface{}{
    //             "x-delayed-type": "direct",
    //         },
    //     })

    // 方式二：使用 Policy（推荐）
    // 先声明一个普通的 direct 交换机，然后通过 Policy 覆盖其类型
    err := conn.Management().DeclareExchange(ctx,
        &rmq.DirectExchangeSpecification{
            Name:      "delayed.exchange",
            IsDurable: true,
        })
    return err
}

// 发送带延迟的消息
func sendWithDelay(ctx context.Context, publisher *rmq.Publisher,
    body []byte, delayMs int) error {

    msg := rmq.NewMessage(body)
    msg.SetContentType("application/json")

    // 关键：通过消息注解设置延迟时间（毫秒）
    msg.SetMessageAnnotation("x-delay", delayMs)

    result, err := publisher.Publish(ctx, msg)
    if err != nil {
        return err
    }

    switch result.Outcome.(type) {
    case *rmq.StateAccepted:
        log.Printf("延迟消息已发送 (delay=%dms): %s", delayMs, string(body))
        return nil
    default:
        return fmt.Errorf("延迟消息发送失败: %v", result.Outcome)
    }
}

延迟队列常见使用场景

场景 延迟时间 示例
订单超时取消 30 分钟 下单后 30 分钟未支付自动取消
短信重发 递增间隔 1分钟、5分钟、30分钟阶梯重试
定时提醒 指定时间点 会议开始前 15 分钟提醒
延迟重试 递增间隔 调用第三方 API 失败后延迟重试
Stream 流队列优先级、TTL 与惰性队列