消息队列基础：模型、选型与可靠投递

消息队列不是"把 HTTP 调用换成异步调用"这么简单。一旦业务链路经过 MQ，就要回答三件事：消息会不会丢、会不会重复、顺序还是不是我要的。这三件事的答案取决于生产者、Broker、消费者三段各自的配置，任何一段配错，整套可靠性就是假的。这篇文章从模型讲起，给出选型对比，然后把可靠投递拆成三段逐一落实，最后覆盖幂等、死信、顺序、积压与消息轨迹。示例以 Node.js 生态为主，命令行为以官方文档为准。

为什么需要消息队列

MQ 解决的是三个真实问题，每个都有对应的代价：

代价要提前接受：一致性问题会变复杂（上游成功了，下游最终失败要补偿）、排查难度上升（一条业务流分散在多个系统日志里）、重复不可避免（下文会讲为什么 exactly-once 在跨系统场景基本是营销话术）。如果这三个代价换不回上面的三个收益（比如流量根本没峰、下游只有一个），不要为了"架构先进"上 MQ。

两大模型：点对点与发布订阅

现代 MQ 把两者组合成消费组模型：一个 topic 下多个 partition，一个消费组内的消费者瓜分 partition——组与组之间互不影响（发布订阅），组内每条消息只给一个消费者（点对点）。这个模型决定了顺序、扩容、积压三个核心问题的上限，选型时先理解它。

选型：RabbitMQ / Kafka / Redis Streams

经验法则：需要复杂路由、任务分发、小规模异步任务 → RabbitMQ；日志管道、事件流、大数据吞吐、要回放 → Kafka；已有 Redis 且量级可控、想省一套运维 → Redis Streams。选型最大的坑是"按品牌选"——Kafka 不会自动给你高吞吐（分区数和消费并发要配够），RabbitMQ 也不会自动给你可靠（默认消息不持久化）。选完型，可靠性还得按下面三段自己配。

可靠投递三段论：生产者、Broker、消费者

一条消息的可靠 = 生产段确认 + Broker 持久化 + 消费段确认，三段缺一不可。

生产者确认：不知道 Broker 收没收到，就等于丢了

RabbitMQ 用 confirm 模式：Channel 开启 confirm() 后，每条消息等 Broker 的 ack；persistent: true 让消息落盘标志生效；开启返回监听捕捉"路由不到队列"的消息：

import amqp from 'amqplib';

const conn = await amqp.connect('amqp://127.0.0.1:5672');
// publisher confirm 必须用 createConfirmChannel：普通 channel 没有 waitForConfirms
const ch = await conn.createConfirmChannel();
await ch.assertExchange('order.events', 'topic', { durable: true }); // exchange 也要 durable
await ch.publish(
  'order.events',
  'order.created',
  Buffer.from(JSON.stringify({ orderId: 5001, items: [] })),
  { persistent: true, messageId: crypto.randomUUID(), headers: { traceId: 't-abc' } },
);
await ch.waitForConfirms();  // 拿不到确认就按失败处理：重试或落补偿表

Kafka 侧对应 acks=all：分区 leader 与所有 ISR 副本确认才算成功，配上幂等 producer 防止重试导致的 Broker 端重复：

import { Kafka } from 'kafkajs';

const kafka = new Kafka({ clientId: 'order-service', brokers: ['127.0.0.1:9092'] });
const producer = kafka.producer({
  idempotent: true,        //Broker 端去重（要求 acks=all）
  acks: 'all',
  retry: { retries: 5 },
});
await producer.connect();
await producer.send({
  topic: 'order.events',
  messages: [{ key: '5001', value: JSON.stringify({ orderId: 5001 }), headers: { traceId: 't-abc' } }],
});

生产者侧还有一个常见漏洞：发送在代码里 fire-and-forget（不 await 确认、不捕获返回）。上线后日志一片绿色，Broker 挂一次才知道消息一直在静默丢失。

Broker 持久化：交换机、队列、消息三个开关

RabbitMQ 的持久化是三件套：exchange durable: true、queue durable: true、消息 persistent: true。任何一件没配，Broker 重启都会丢——最坑的是队列：不 durable 的队列重启后直接消失，绑定的 exchange 也找不到路由目标。高可用用仲裁队列（quorum queue）替代老镜像队列。

Kafka 侧：replication.factor >= 3，topic 配 min.insync.replicas=2，acks=all 才有意义；副本数 1 的 topic 磁盘坏了就是全丢。

消费者确认：先处理完再 ack

const ch2 = await conn.createChannel();
await ch2.assertQueue('order.notifications', { durable: true });
await ch2.prefetch(10);            //未确认消息上限，防止单消费者吃爆内存
await ch2.consume('order.notifications', async (msg) => {
  if (!msg) return;                // 消费者被 cancel 时回调会收到 null
  try {
    await handleOrderEvent(JSON.parse(msg.content.toString()), msg.properties.headers?.traceId);
    ch2.ack(msg);                   // 处理成功才确认
  } catch (error) {
    ch2.nack(msg, false, false);    // requeue=false，交给死信，避免无限循环
  }
});

三条纪律：ack 放在业务成功之后（先 ack 再处理，等于把"处理失败"变成"消息丢失"）；异常 requeue 要克制——毒消息（解析就报错）requeue 后还是报错，必须靠重试上限 + 死信终结；prefetch 必须设置，否则 Broker 会把队列一口气推给一个消费者。

Kafka 侧的对应纪律是位点手动提交：自动提交（按时间间隔）意味着"处理没完成也可能提交了位点"，崩溃时消息就丢了。改为处理成功后手动提交：

// kafka 变量沿用上文生产者示例
const consumer = kafka.consumer({ groupId: 'order-notifier' });
await consumer.connect();
await consumer.subscribe({ topic: 'order.events', fromBeginning: false });
await consumer.run({
  autoCommit: false,                        // 关闭自动提交位点
  eachBatch: async ({ batch, commitOffsetsIfNecessary }) => {
    for (const message of batch.messages) {
      await handleOrderEvent(JSON.parse(message.value.toString()), message.headers?.traceId);
    }
    commitOffsetsIfNecessary();             // 整批处理完才提交：崩溃时整批重投，靠幂等消化
  },
});

两种确认模型的对拍：RabbitMQ 是逐条 ack（broker 追踪每条未确认消息）；Kafka 是位点提交（消费者组记录"消费到哪了"）。位点模型没有逐条状态，所以失败恢复的最小单位是整批——这既是 Kafka 高吞吐的来源，也是"消费端必须幂等"的又一个理由。

重复消费与幂等

三段全配好，仍然会重复：生产者等确认超时但消息其实已入队（重试产生重复）；消费者处理完业务、ack 之前进程崩溃（重启后重投）。所以可靠投递的终点是 at-least-once + 消费端幂等。

幂等键的选择取决于业务：订单号、业务流水号、或生产者生成的 messageId。去重落地是数据库唯一约束（去重表）或 Redis SETNX，完整方案（唯一约束兜底、状态机、去重表与 Redis 的取舍）见本站《幂等设计：从接口到消息消费的防重体系》。

对 Kafka 的"exactly-once"要说清边界：幂等 producer + 事务只覆盖同一 Kafka 生态内 source topic 到 sink 的精确一次；消费端要把结果写进 MySQL、要调第三方接口，跨出生态就退回 at-least-once，幂等仍然要自己做。

死信队列与重试梯度

直接 requeue 的问题：立刻重投 → 再失败 → 再重投，失败消息风暴会拖垮消费。成熟做法是梯度重试：

消费失败
  │
  ├── 第 1 次失败 ──> retry.10s 队列（TTL 10s，到期死信回主队列）
  ├── 第 2 次失败 ──> retry.1m 队列（TTL 1m）
  ├── 第 3 次失败 ──> retry.10m 队列（TTL 10m）
  └── 超过梯度上限 ──> dead 队列 ──> 告警 + 人工/自动补偿任务

// 用 x-death 头里的计数决定下一跳（示意）
async function declareRetryTopology(ch: amqp.Channel) {
  await ch.assertQueue('retry.10s', {
    durable: true,
    arguments: {
      'x-message-ttl': 10_000,
      'x-dead-letter-exchange': '',           // 死信回到默认 exchange
      'x-dead-letter-routing-key': 'order.notifications', // 路由回主队列
    },
  });
  await ch.assertQueue('dead.order.notifications', { durable: true });
}

要点：死信队列必须带告警，定期有人看，否则只是把问题换个地方堆着；主队列本身要配 DLX，消费者 nack(requeue=false) 的消息才有去处；梯度节奏（10s/1m/10m/1h）按下游恢复预期定，不是拍脑袋指数退避到底。

顺序消息

顺序的粒度要分清：全局有序（整个 topic 严格按序）代价极高——Kafka 单分区、RabbitMQ 单队列单消费者，吞吐被锁死；实际需要的是分区/业务键内有序：同一个订单的消息按序，不同订单之间无所谓。

Kafka 用 key 分区天然获得这个性质：key = orderId 保证同单消息落同一分区。两个边界必须写进设计文档：消费端失败处理会破坏顺序——消息 A 失败进重试梯度时，消息 B 已被处理，A 回来后顺序已经颠倒。对严格顺序的业务（状态机流转），失败只能"暂停当前分区继续消费并告警"，不能跳过也不能后台重试。扩缩容会触发再平衡，分区在消费者之间迁移，短暂窗口内处理停顿，要评估再平衡超时参数。

积压治理

积压的第一个问题是发现得晚。把 consumer lag 做成核心告警指标（lag = 消费位点到最新位点的差值），Kafka 用管理脚本或指标系统：

# 查看消费组每个分区的积压量
kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 \
  --describe --group order-service

发现积压后的三档手段，按代价从低到高：

1. 批量消费：一次 poll 一批，数据库操作合并成批量写（一条 INSERT 变一次 VALUES 多行），吞吐立刻翻几倍——最常被忽略的选项。
2. 扩容消费者：加消费者实例分摊分区（Kafka 单分区吞吐有上限，分区数不够就加分区；加分区只对新消息生效，老分区继续积压）。
3. 降级与跳过：确认积压的是非核心消息（埋点、通知），按时间阈值丢弃或采样消费，并在事后对账补数。核心消息不能丢时，新建一个高分区数的 topic，写个"搬运消费者"把积压消息倒进去并行消化。

预案要提前写：哪类消息可丢、lag 到多少扩容、谁值守。积压时的凌晨决策质量一定低于白天写好的预案。

消息轨迹

消息问题的排查依赖轨迹：一条消息从哪来、什么时候被 Broker 确认、投给了谁、重试几次、最后 ack 还是死信。实践做法：

• 入口处生成 traceId，放进消息 headers，一路透传到下游服务的日志与后续消息。
• 生产、确认、消费开始、消费成功、失败重试、死信，每个节点打一条结构化日志，带 messageId、traceId、队列名、时间戳、耗时。
• Kafka 侧顺带记录 topic/partition/offset，RabbitMQ 侧记录 queue 与 deliveryTag，出现问题时能精确圈出消息落点。

const traceId = msg.properties.headers?.traceId ?? crypto.randomUUID();
logger.info('consume', {
  messageId: msg.properties.messageId,
  traceId,
  queue: 'order.notifications',
  deliveryTag: msg.fields.deliveryTag,
});
// 业务内部再发下游消息时，把 traceId 继续写进 headers

RocketMQ 的消息轨迹（生产/存储/消费全链路上报）是同类思想的官方集成版，自建 RabbitMQ/Kafka 时按上面的打点清单补齐即可。

注意事项与边界

• MQ 不是数据库：消息保留时间（Kafka 7 天默认、RabbitMQ 确认后即删）不是业务数据年限，重要状态必须落库，消息只是传输载体。
• 消息体格式要版本化：加字段可以、删改字段不行，新旧消费者要能在灰度期共存解析。
• 上游业务库与发消息之间没有原子性：库提交成功、发消息失败（或反过来）会出现"有单无事件"。常见做法是事务性发件箱：在同一个数据库事务中写入业务变更和待投递事件，再由后台投递器重试发送；投递可能重复，消费者仍要按消息 ID 幂等处理，见幂等设计。
• 重试会产生重复，任何消费逻辑先按"我会被投递多次"设计，再谈性能。
• 环境隔离：测试环境的队列与生产严格分开，一次误连清空生产队列的代价远高于多维护一套环境。

总结

消息队列的可靠不来自产品品牌，来自三段确认各自闭环：生产段等确认（confirm/acks=all）+ Broker 段三件套持久化 + 消费段先处理后 ack。在此之上，承认 at-least-once、用幂等键消化重复，用梯度重试与死信兜住失败，用分区键定义顺序粒度，用 lag 监控与预案治理积压。最后，把轨迹打点当上线前的必做项——排查消息问题时，没有轨迹就等于盲飞。

参考

• RabbitMQ: Publisher Confirms
• RabbitMQ: Dead Letter Exchanges
• RabbitMQ: Consumer Acknowledgements 与 prefetch
• Apache Kafka: Delivery Semantics（acks / 幂等 / 事务）
• Apache Kafka: Consumer Configs
• Redis Streams 官方文档
• Node.js amqplib 官方站点