目录
一、两个方向相反的事故，同一个根因二、MQ 真正不可替代的是背压，不是「解耦」三、三种投递语义，只有一种能当默认值四、双写问题：为什么「先写库再发消息」一定是错的五、幂等消费：at-least-once 的对偶面六、顺序：99% 的需求不是全局有序七、小结参考来源
3488 字
17 分钟
消息队列不生产可靠性：at-least-once、双写问题与幂等消费实战
2026-10-14
互联网
消息队列
/
Kafka
/
RabbitMQ
/
分布式系统
/
后端
/
架构

一、两个方向相反的事故，同一个根因#

先看两个现场。

事故 A：券没发出去。 用户下单成功、支付成功，但优惠券始终没到账。排查发现代码顺序是先提交订单事务、再 producer.send()，而那次发布恰好发生在这两行之间。消息从来没有被发出去过——没报错、没告警，因为站在订单服务的视角看，事务是成功的。

事故 B：库存扣了两遍。 库存服务重启，一条尚未 ack 的扣减消息被重新投递，同一笔订单扣了两次库存，最后超卖了。

一个丢、一个重，方向相反，但根因相同：把「消息发出去了」和「消息只被处理一次」当成了 MQ 的默认保证。这两件事 MQ 都不保证。broker 保证的是”一旦收下就不会丢”，它管不着你有没有发；而绝大多数 broker 的默认投递语义是 at-least-once——重复是常态，不是异常。

这篇文章的主线就一句话：MQ 不生产可靠性，它只是把「同步调用的失败」换成了「异步投递的失败」。同步失败会抛异常，异步失败会静默地丢或者静默地重复，所以这笔账必须由你自己的数据库来兜。

二、MQ 真正不可替代的是背压，不是「解耦」#

“解耦、异步、削峰”是介绍 MQ 的三件套，但其中只有一件是 MQ 独有的。解耦和异步不用 MQ 也能做（本地队列、setImmediate、线程池都是），它们更像副产品。真正不可替代的是背压：把 5 秒钟内涌入的 10 万次写入，摊平成 10 分钟里稳定的消费速率，让下游按自己的节奏处理；顺带把下游故障从”接口 500”隔离成了”消息堆积”。

但这份价值是有账单的：异步链路上没有调用栈。同步调用失败会在日志里留下异常和完整栈；异步消费失败留下的只是消息堆积，而堆积的原因可能是一年前的某次 schema 不兼容。所以选 MQ 时真正要评估的不是”能不能解耦”，而是”这套失败语义我兜不兜得住”。

三、三种投递语义，只有一种能当默认值#

语义怎么实现丢/重的表现适用
at-most-once生产者 acks=0、消费者自动 ack会丢，绝不重日志、指标、埋点
at-least-once生产者重试 + 处理完再提交 offset / 手动 ack不丢，可能重几乎所有业务场景
exactly-once幂等生产者 + 事务理想态仅在特定闭环内成立

at-most-once 就是”发出去就不管了”：Kafka 生产者 acks=0、RabbitMQ 消费端 autoAck=true，broker 把它写进 socket 就忘掉，消费者进程一崩，在途消息就没了。at-least-once 是默认值，也是唯一现实的默认值：生产端重试（网络抖动时自动重发）加消费端”处理成功后再确认”，两半合起来保证不丢，代价就是重复——重试会重复，确认丢失后重投也会重复。

exactly-once 需要看清它成立的范围。Kafka 的 EOS 是把输出和 offset 提交绑在同一个事务里，所以它只在**“从 Kafka 读、处理、写回 Kafka”这个闭环内**成立：

const producer = kafka.producer({
  transactionalId: 'order-processor-1', // 持久化且全局唯一，用于僵尸实例隔离
  idempotent: true,                     // 事务的前提
});


await producer.connect();
await producer.initTransactions();      // 必须在循环外只调一次


await producer.beginTransaction();
try {
  await producer.send({ topic: 'order.events', messages: [{ value: JSON.stringify(out) }] });
  // 把消费位点一起算进这个事务，而不是手动提交
  await producer.sendOffsetsToTransaction({ topics, consumerGroupId: 'order-processor' }, consumer);
  await producer.commitTransaction();
} catch (err) {
  await producer.abortTransaction();
  throw err;
}

一旦越过这个边界——去写 MySQL、去调支付网关、去投递到另一个集群——事务就管不着了，落到 at-least-once。所以端到端的”恰好一次”没有银弹，它的唯一实现方式是：at-least-once 投递 + 幂等消费，两者合成出来的。下一节和第五节分别解决这两半。

（顺带两个坑：消费端不设 isolation.level=read_committed 的话，事务里尚未提交的消息照样能读到；而幂等生产者要求 acks=all、retries>0、max.in.flight.requests.per.connection ≤ 5，配错任一项就静默降级。）

四、双写问题：为什么「先写库再发消息」一定是错的#

事故 A 的写法在代码里长这样：

await db.tx((tx) => tx.none(`UPDATE orders SET status = 'PAID' WHERE id = $1`, [id]));
// ← 进程在这里重启、或 MQ 抖动，下面这行永远不会执行
await producer.send({ topic: 'order.events', messages: [{ value: JSON.stringify({ id }) }] });

反过来先发消息再写库，问题变成发完消息事务回滚——下游收到一条查不到订单的幽灵消息。这就是双写问题：两个没有共同事务边界的系统，你无论如何都会在两者之间留下一个”可能不一致”的窗口。

有人会说”那我用分布式事务”。2PC/XA 确实能解决，代价是同步阻塞、协调者单点，而且 Kafka 这类 broker 根本不支持 XA。RocketMQ 的事务消息（半消息 + 事务回查）是另一条路，但要换 broker，且只解决这一个方向的问题。

通用的正解是事务性发件箱（Transactional Outbox）：不发消息给 MQ，而是把”要发的消息”当成一行数据，写进同一个数据库的同一张表，和业务变更在同一个事务里提交。

-- 关键前提：业务表和 outbox 表在同一个库里，才能共享一个本地事务
CREATE TABLE outbox (
  id             BIGSERIAL PRIMARY KEY,  -- 它同时也是事件的去重键
  aggregate_type TEXT        NOT NULL,
  aggregate_id   TEXT        NOT NULL,
  type           TEXT        NOT NULL,   -- 如 order.created
  payload        JSONB       NOT NULL,
  created_at     TIMESTAMPTZ NOT NULL DEFAULT now()
);
async function payOrder(orderId) {
  return db.tx(async (tx) => {
    await tx.none(`UPDATE orders SET status = 'PAID' WHERE id = $1`, [orderId]);
    // 消息的"发送"退化成一次本地 INSERT，和业务变更同生共死
    await tx.none(
      `INSERT INTO outbox (aggregate_type, aggregate_id, type, payload)
       VALUES ('order', $1, 'order.paid', $2)`,
      [String(orderId), JSON.stringify({ orderId })]
    );
  });
}

这样一来，“要么都发生要么都不发生”由数据库的本地事务保证了。它之所以成立，是因为它把”两个系统的双写”降级成了”一个系统的单写”，剩下的投递工作交给一个叫**中继（relay）**的角色异步完成：

// 轮询发布者：最朴素的中继，简单、不依赖 CDC 基础设施
async function relayOnce(producer) {
  await db.tx(async (tx) => {
    // SKIP LOCKED 让多个中继实例可以并行跑，互不阻塞
    const rows = await tx.manyOrNone(
      `SELECT * FROM outbox ORDER BY id LIMIT 100 FOR UPDATE SKIP LOCKED`
    );
    for (const row of rows) {
      await producer.send({
        topic: 'order.events',
        messages: [{ key: row.aggregate_id, value: JSON.stringify(row.payload),
                     headers: { 'event-id': String(row.id) } }],
      });
    }
    await tx.none(`DELETE FROM outbox WHERE id = ANY($1)`, [rows.map((r) => r.id)]);
  });
}

中继只有两种实现路线：轮询发布者（定时扫表，如上）和事务日志尾随（Debezium 的 Outbox Event Router 读数据库 binlog/WAL，延迟更低、对业务库压力更小）。前者十几行代码就能上，后者适合量大的场景。

注意最后那段：投递成功之后、DELETE 之前崩溃，消息就会重复投递。outbox 保证的是”不丢”，它换来的仍然是 at-least-once。正因如此，上面把 outbox.id 塞进了消息头——消费端得有东西可去重。

五、幂等消费：at-least-once 的对偶面#

既然重复不可避免，消费端就必须自己扛住。两种写法，按性价比排序。

写法一：去重表 + 唯一索引。 把”这条消息处理过了”也当成一行数据，和业务变更放在同一个事务里：

CREATE TABLE processed_messages (
  message_id   TEXT PRIMARY KEY,
  processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
async function handle(msg) {
  const messageId = msg.headers['event-id']; // 稳定且唯一，不许在这里现生成 uuid
  await db.tx(async (tx) => {
    const r = await tx.result(
      `INSERT INTO processed_messages (message_id) VALUES ($1)
       ON CONFLICT (message_id) DO NOTHING`,
      [messageId]
    );
    if (r.rowCount === 0) return; // 已经处理过，直接跳过
    await tx.none(`UPDATE stock SET qty = qty - $1 WHERE sku = $2`, [msg.qty, msg.sku]);
  });
}

有两个容易踩的点。第一，去重记录和业务写入必须在同一个事务里——如果分开写，“插了去重记录但业务失败”会让这条消息被永久跳过，等于丢消息。第二，processed_messages 会无限增长，得配合分区表或定期裁剪（且保留窗口要长于消息的最大可能重投延迟；SQS 的去重窗口是固定 5 分钟，这本身就是个约束）。

写法二：把业务操作本身设计成幂等的。 如果业务是一个状态机，那”幂等”不需要额外的表：

-- 重放时 rowCount = 0，天然幂等，且不需要额外的去重表
UPDATE orders SET status = 'PAID' WHERE id = $1 AND status = 'PENDING';

更好的做法是让写操作只表达”目标状态”而不是”增量”。SET qty = qty - 1 天生不幂等，而 SET qty = 90（或用版本号做乐观锁）就是幂等的——这也是为什么在事件驱动系统里，事件最好携带状态而非指令。去重表是通用兜底，状态机幂等才是治本。

六、顺序：99% 的需求不是全局有序#

“消息要保证顺序”是最容易提错的需求。全局有序意味着单分区单消费者，等于放弃并行，吞吐会被压到一条线。而真实需求几乎都是同一业务实体的有序：同一订单的状态变更不能颠倒，不同订单之间毫无关系。

表达方式是分区键：Kafka 用 message key、SQS FIFO 用 MessageGroupId、RabbitMQ 用 routing key。同一 key 进同一分区，分区内天然有序，这就是分布式系统里”有序”的正确粒度。然后是三件必须知道的事：

1. 扩分区会打破 key → 分区的映射。 Kafka 默认策略是 hash(key) % 分区数，分区数一变，同一个 key 就会被路由到另一个分区，“扩容前”和”扩容后”的消息分居两地，顺序就此断裂。Confluent 文档直写 “If the number of partitions changes, this delivery guarantee may no longer hold”，Kafka 官方也为此专门修订了文档：同一 key 可能落到不同分区、既有 key 的顺序保证受影响，而且不会自动重分布已有数据，分区数只能加不能减。规避办法只有一个——在设计期就按未来两三年的量预留分区数，宁多勿少。

2. 生产者重试会乱序。 重试时如果 max.in.flight.requests.per.connection > 1，后发的批次可能先落盘。开幂等生产者（此时上限可放大到 5）才安全。但更隐蔽的是消费端：如果一个分区内的 handler 是并发执行的，分区有序就白搭了。要么单分区串行消费，要么在应用层把 key 哈希到固定数量的 worker 上，保证”同 key 落同一 worker”。

3. 一条毒消息能堵死整个分区。 在”分区内串行”的前提下，一条永远处理失败的消息会阻塞它后面所有的消息。RabbitMQ 里更要小心：nack 且 requeue=true 的消息会回到队头并被立即重投，一条必然失败的消息很快就能烧掉整个 CPU——这就是 hot loop。正确做法是给重试设上限：RabbitMQ 用死信交换机（DLX）配合 x-delivery-limit，Kafka 侧常用重试 topic 链加一个 DLT。

一句话：重试次数上限 + 退避 + 死信队列不是可选项，是必需品——没有它们，一次失败会升级成一次停摆。

七、小结#

  • MQ 不生产可靠性，它只是把同步调用的失败换成了异步投递的失败。异步失败不会抛异常，所以必须由你自己的数据库兜底。
  • at-least-once 是唯一现实的默认值：它由”生产端重试 + 消费端处理完再 ack”两半合成，代价就是重复。
  • Kafka 的 exactly-once 只在”读 Kafka、处理、写回 Kafka”这个闭环内成立，越过边界就退回 at-least-once。端到端的恰好一次 = at-least-once + 幂等消费。
  • 别双写。 把消息当成一行数据写进 outbox 表，和业务变更共享一个本地事务，再用轮询发布者或事务日志尾随把投递这件事异步化。
  • outbox 保证不丢，但不保证不重，所以 event-id 必须随消息传出去，供消费端去重。
  • 幂等去重表要和业务写入在同一个事务里，否则会从”重复”变成更糟的”丢失”。
  • 顺序用 key 表达，不用全局串行；扩分区会打破 key 到分区的映射，所以分区数要在设计期就预留够。
  • 重试必须有上限，否则一条毒消息就能让整个分区停摆。

参考来源#

消息队列不生产可靠性：at-least-once、双写问题与幂等消费实战
https://www.hehonglei.cn/posts/message-queue-delivery-semantics-and-outbox-pattern/
作者
Honglei He
发布于
2026-10-14
许可协议
CC BY-NC-SA 4.0
Agent 记忆系统设计：工作、语义、情景、程序四层，以及各自的失效语义
Open Graph 与结构化数据：让链接在聊天框和搜索结果里都好看
© 2026 Honglei He. All Rights Reserved. / RSS / Sitemap
沪ICP备2024081718号-1
Powered by Astro & Fuwari