消息队列高级——死信队列、延迟消息、消息积压 / Advanced Messaging with Dead Letters, Delays, and Backlogs

📅 创建时间：2026-05-08 🏷️ 标签：#死信队列 #延迟消息 #消息重试 #消息积压 #幂等消费 📚 前置知识：[[00-backend-overview]] [[02-mq-kafka]] [[03-mq-others]] 📚 相关知识：[[04-distributed-system]]（分布式事务） [[11-architecture-patterns]]（Saga）

---

场景：消息处理失败了，系统进入死循环

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  凌晨 3 点，监控报警：Kafka 消费 lag 暴增。             │
│                                                             │
│  排查发现：                                              │
│  订单服务消费消息时遇到 NPE（空指针异常）              │
│  → 消息被重复消费                                        │
│  → 又遇到 NPE                                            │
│  → 又重复消费                                            │
│  → 死循环了！                                            │
│                                                             │
│  更糟糕的是：                                            │
│  → 消息被消费了 1000 次                                   │
│  → 重复创建了 1000 个订单                                 │
│                                                             │
└─────────────────────────────────────────────────────────────┘

这一章，我们解决消息队列的三个高级问题：死信队列、延迟消息、消息积压。

---

第1节：死信队列——消息的"垃圾回收站"

什么是死信

┌─────────────────────────────────────────────────────────────┐
│                 消息变成死信的条件                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 消费失败达到最大重试次数                             │
│  → 重试 3 次后仍然失败 → 进入死信队列                   │
│                                                             │
│  2. 消息超时未消费                                        │
│  → 延迟消息 TTL 到期 → 进入死信队列                     │
│                                                             │
│  3. 消息被拒绝（Reject）而不 requeue                    │
│  → 消费者明确拒绝消息 → 进入死信队列                   │
│                                                             │
│  4. 队列满了                                            │
│  → 超出队列容量 → 旧消息进入死信队列                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Kafka 死信队列实现

from kafka import KafkaConsumer, KafkaProducer
import json

# 正常消费主题
main_topic = "order-created"

# 死信主题
dlq_topic = "order-created-dlq"

# 最大重试次数
MAX_RETRIES = 3

consumer = KafkaConsumer(
    main_topic,
    bootstrap_servers=['kafka:9092'],
    group_id='order-processor',
    enable_auto_commit=False,
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

dlq_producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

for message in consumer:
    order = message.value
    retries = message.headers.get('retry-count', 0)

    try:
        process_order(order)
        consumer.commit()

    except RetryableError as e:
        # 可重试错误：重新入队
        if retries < MAX_RETRIES:
            # 重新发送消息，带上重试次数
            order['_retry_count'] = retries + 1
            producer.send(main_topic, order)
            producer.flush()
            consumer.commit()
        else:
            # 超过最大重试次数 → 进入死信队列
            order['_dlq_reason'] = str(e)
            order['_dlq_time'] = time.time()
            dlq_producer.send(dlq_topic, order)
            dlq_producer.flush()
            consumer.commit()

    except UnretryableError as e:
        # 不可重试错误：直接进入死信队列
        order['_dlq_reason'] = str(e)
        order['_dlq_time'] = time.time()
        dlq_producer.send(dlq_topic, order)
        dlq_producer.flush()
        consumer.commit()

RocketMQ 死信队列（原生支持）

┌─────────────────────────────────────────────────────────────┐
│              RocketMQ 死信队列                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  RocketMQ 原生支持死信队列，无需手动实现：             │
│                                                             │
│  消息重试次数超过阈值 → 自动进入 %DLQ% 前缀的队列      │
│                                                             │
│  原生主题：order-created                                  │
│  死信主题：%DLQ%order-created                            │
│                                                             │
│  消费配置：                                              │
│  consumer.setMaxReconsumeTimes(3)  # 最大重试 3 次   │
│                                                             │
│  死信消息特性：                                          │
│  • 不会自动删除，保留 3 天                              │
│  • 需要人工处理或定时清理                               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

死信队列的监控和处理

# 死信队列消费者：人工处理 / 自动修复
dlq_consumer = KafkaConsumer(
    dlq_topic,
    bootstrap_servers=['kafka:9092'],
    group_id='dlq-processor',
    auto_offset_reset='earliest'
)

for message in dlq_consumer:
    dlq_msg = json.loads(message.value)
    reason = dlq_msg.get('_dlq_reason', 'unknown')
    retry_count = dlq_msg.get('_retry_count', 0)

    # 分类处理
    if 'JsonParseException' in reason:
        # 数据格式错误：无法修复，直接记录日志
        log.error(f"DLQ 数据格式错误，无法处理: {dlq_msg}")

    elif 'DatabaseConnectionException' in reason:
        # 数据库连接问题：可能是临时故障，等待后重试
        time.sleep(60)
        kafka.send(main_topic, dlq_msg)  # 重新放回主队列

    elif 'BusinessLogicError' in reason:
        # 业务逻辑错误：需要人工介入
        send_alert(f"DLQ 业务错误，需要人工处理: {reason}")

    else:
        # 未知错误：统一记录
        log.error(f"DLQ 未知错误: {reason}")

---

第2节：延迟消息——订单 30 分钟未支付自动取消

场景：订单超时取消

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  用户下单 → 订单状态 = "待支付"                          │
│  如果 30 分钟内未支付 → 自动取消订单                      │
│                                                             │
│  方案 A：定时轮询数据库（浪费资源）                      │
│  每分钟扫描：SELECT * FROM orders                         │
│            WHERE status='pending' AND create_time < NOW()-30min│
│  → 数据库压力大                                          │
│                                                             │
│  方案 B：延迟消息（推荐）                                │
│  下单时发送一条延迟消息（30 分钟后投递）                 │
│  → 30 分钟后消息被消费 → 检查订单状态 → 未支付则取消  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

RocketMQ 延迟消息

// RocketMQ 原生支持延迟消息（18 个延迟级别）
public class OrderService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void createOrder(OrderDTO order) {
        // 1. 创建订单
        Order createdOrder = orderRepository.save(order);

        // 2. 发送延迟消息（延迟 30 分钟）
        rocketMQTemplate.asyncSend("order-timeout-topic", createdOrder,
            new MessageDelayLevel(30, TimeUnit.MINUTES));
    }
}

// 消费者：处理超时订单
@RocketMQMessageListener(topic = "order-timeout-topic", consumerGroup = "order-cancel")
public class OrderTimeoutConsumer implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        Order current = orderRepository.findById(order.getId());

        if ("pending".equals(current.getStatus())) {
            // 未支付，取消订单
            current.setStatus("cancelled");
            current.setCancelReason("超时未支付");
            orderRepository.save(current);

            // 恢复库存
            inventoryService.restore(order.getProductId(), order.getQuantity());
        }
        // 如果已支付，忽略
    }
}

RabbitMQ 延迟队列（TTL + DLX）

┌─────────────────────────────────────────────────────────────┐
│              RabbitMQ 延迟队列实现                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐  │
│  │                                                     │  │
│  │  order.delay.queue (TTL=30分钟)                    │  │
│  │      ↓ (消息过期)                                   │  │
│  │  order.dlx.exchange (死信交换机)                   │  │
│  │      ↓                                              │  │
│  │  order.cancel.queue (实际处理队列)                 │  │
│  │                                                     │  │
│  └─────────────────────────────────────────────────────┘  │
│                                                             │
│  1. 下单时，消息发送到 order.delay.queue                 │
│  2. 消息在队列中等待 TTL（30 分钟）                     │
│  3. 消息过期后，进入 DLX                                  │
│  4. DLX 路由到 order.cancel.queue                       │
│  5. 消费者处理取消逻辑                                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Redis ZSet 实现延迟队列

import time
import json

# 使用 ZSet 实现延迟队列
# Score = 过期时间戳

def add_delay_task(task_id, task_data, delay_seconds):
    """添加延迟任务"""
    execute_at = time.time() + delay_seconds
    redis.zadd("delay:queue", {json.dumps(task_data): execute_at})

def process_delay_tasks():
    """定时扫描并执行到期的任务"""
    while True:
        now = time.time()
        # 获取所有到期的任务（score <= now）
        tasks = redis.zrangebyscore("delay:queue", 0, now)

        for task_data in tasks:
            task = json.loads(task_data)
            try:
                process_order_timeout(task)
                # 执行成功后删除
                redis.zrem("delay:queue", task_data)
            except Exception as e:
                log.error(f"处理延迟任务失败: {e}")
                # 可选择重试或进入死信队列

        time.sleep(1)  # 每秒检查一次

---

第3节：消息积压——消费者跟不上生产速度

场景：消费者挂了 1 小时

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  消费者服务凌晨发布重启。                                 │
│                                                             │
│  发布时间：2:00 - 2:30（30 分钟消费者不可用）           │
│                                                             │
│  积压量：                                              │
│  生产速度：1000 条/秒 × 1800 秒 = 180 万条              │
│                                                             │
│  恢复后：                                              │
│  消费者需要处理 180 万条积压                             │
│  → 按正常速度需要 30 分钟才能追平                      │
│  → 这 30 分钟内新消息还在继续积压                     │
│                                                             │
│  后果：                                                 │
│  → 延迟越来越大                                        │
│  → 超时订单越来越多                                    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

解决方案一：增加消费者

# Kafka 消费者组：增加分区数和消费者数
# 分区数 = 最大并行度

# 当前：3 个分区，3 个消费者
# 积压后：扩容到 6 个分区，6 个消费者
# 并行度翻倍，消费速度翻倍

# 消费者配置
consumer = KafkaConsumer(
    'order-topic',
    bootstrap_servers=['kafka:9092'],
    group_id='order-processor',  # 同一消费者组
    # 消费者数量 <= 分区数量
)

解决方案二：消息丢弃 + 补偿（极端情况）

# 当积压超过阈值时，丢弃旧消息，优先处理新消息
def consume_with_backpressure():
    # 检查积压量
    lag = consumer.end_offset() - consumer.position()

    if lag > 100000:  # 积压超过 10 万
        log.warning(f"消息积压严重 {lag}，启动紧急处理")

        # 跳过旧消息，从最新开始消费
        consumer.seek_to_end()

        # 记录跳过的消息
        skipped_count = lag
        metrics.increment("messages_skipped", skipped_count)

        # 发送告警
        send_alert(f"跳过 {skipped_count} 条积压消息，请人工核对")

    else:
        # 正常消费
        for message in consumer:
            process_message(message)

解决方案三：消费者优先级

# 优先处理新消息（快速追平），旧消息异步补偿
class PriorityConsumer:
    def __init__(self):
        self.main_consumer = KafkaConsumer('order-topic', group_id='main')
        self.compensation_consumer = KafkaConsumer('order-topic', group_id='compensation')

    def run(self):
        lag = self.get_lag()

        if lag > 100000:
            # 积压严重：主消费者从最新开始，补偿消费者处理旧消息
            self.main_consumer.seek_to_end()
            self.compensation_consumer.seek_to_beginning()

            # 并行处理
            Thread(target=self.consume_main).start()
            Thread(target=self.consume_compensation).start()
        else:
            # 正常消费
            for msg in self.main_consumer:
                self.process(msg)

    def consume_compensation(self):
        # 异步处理积压消息
        for msg in self.compensation_consumer:
            # 标记为"补偿处理"
            order = json.loads(msg.value)
            order['_compensation'] = True
            self.process(order)

---

第4节：消息幂等消费的完整方案

消息幂等的五种方案

┌─────────────────────────────────────────────────────────────┐
│              消息幂等五种方案                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  方案 1：唯一消息 ID（推荐）                             │
│  生产者生成唯一 ID，消费者去重                            │
│  ┌─────────────────────────────────────────────────────┐  │
│  │  producer: msg["idempotency_id"] = UUID()        │  │
│  │  consumer:                                          │  │
│  │    if redis.setnx(f"processed:{id}", 1, ex=86400): │  │
│  │        process(msg)                                 │  │
│  │    else:                                            │  │
│  │        pass  # 已处理过                             │  │
│  └─────────────────────────────────────────────────────┘  │
│                                                             │
│  方案 2：数据库唯一索引                                  │
│  ┌─────────────────────────────────────────────────────┐  │
│  │  INSERT INTO orders (...)                           │  │
│  │  ON CONFLICT (idempotency_key) DO NOTHING         │  │
│  └─────────────────────────────────────────────────────┘  │
│                                                             │
│  方案 3：Redis SET NX                                    │
│  SET order:process:idempotency_key unique_value NX EX 86400 │
│                                                             │
│  方案 4：状态机幂等                                      │
│  UPDATE orders SET status='paid'                          │
│  WHERE id=? AND status='pending'                        │
│  -- 如果已经支付，这条 SQL 影响 0 行                     │
│                                                             │
│  方案 5：Kafka 事务                                      │
│  消费消息 + 业务处理 + 提交 offset 在同一事务中         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

完整幂等消费实现

class IdempotentConsumer:
    def __init__(self, redis_client, db):
        self.redis = redis_client
        self.db = db
        self.consumer = KafkaConsumer('order-topic', group_id='order-processor')

    def process(self, message):
        msg = json.loads(message.value)
        idempotency_key = msg['idempotency_key']

        # 1. Redis 去重（快速）
        if not self.redis.setnx(f"processed:{idempotency_key}", 1):
            log.info(f"消息已处理，跳过: {idempotency_key}")
            return

        # 2. 设置过期时间（防止 Redis 数据膨胀）
        self.redis.expire(f"processed:{idempotency_key}", 86400)

        # 3. 业务处理
        try:
            self.create_order(msg)

            # 4. 提交 offset
            self.consumer.commit()

        except DuplicateOrderException:
            # 订单重复（数据库唯一约束），幂等成功
            self.consumer.commit()

        except Exception as e:
            # 业务异常，删除 Redis 标记，允许重试
            self.redis.delete(f"processed:{idempotency_key}")
            raise e

    def create_order(self, msg):
        # 数据库幂等插入
        self.db.execute("""
            INSERT INTO orders (id, user_id, product_id, idempotency_key, ...)
            VALUES (?, ?, ?, ?, ...)
            ON CONFLICT (idempotency_key) DO NOTHING
        """, ...)

---

升华：MQ 高级模式的最佳实践

┌─────────────────────────────────────────────────────────────┐
│              MQ 高级模式 Checklist                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  死信队列：                                              │
│  ✅ 消费失败的消息必须进入 DLQ，不能无限重试             │
│  ✅ DLQ 消息需要监控和定期清理                           │
│  ✅ 不同错误类型路由到不同的 DLQ                         │
│                                                             │
│  延迟消息：                                             │
│  ✅ RocketMQ 推荐（原生支持 18 个延迟级别）              │
│  ✅ Redis ZSet 适合自建延迟队列                         │
│  ✅ 延迟精度要求高时用时间轮算法                        │
│                                                             │
│  消息积压：                                             │
│  ✅ 监控 consumer lag，设置告警阈值                      │
│  ✅ 积压时快速扩容消费者                                 │
│  ✅ 极端情况可丢弃旧消息，优先处理新消息                │
│                                                             │
│  幂等消费：                                             │
│  ✅ 每条消息必须有唯一标识（idempotency_key）            │
│  ✅ 幂等去重 + 数据库唯一约束 双保险                    │
│  ✅ 处理失败要删除幂等标记，允许重试                    │
│                                                             │
│  一句话总结：                                              │
│  MQ 的可靠性 = 消息不丢 + 处理不重 + 失败可查。          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

"AI 可查 vs 必须理解"清单

AI 可查：
✅ Kafka DLQ 的具体配置方式
✅ RocketMQ 延迟消息的参数配置
✅ 各语言的 Kafka 消费者 API

必须理解：
🔴 消息变成死信的四种条件
🔴 延迟消息的实现原理（TTL+DLX vs ZSet vs 时间轮）
🔴 消费者积压时的应急处理策略
🔴 五种幂等消费方案及各自适用场景

---

学习状态：🟡 开始学习