RabbitMQ 消费幂等实战：把去重记录和业务更新放进同一个事务

结论：ACK 不能替代业务幂等

接入腾讯云消息队列 RabbitMQ 版后，消费者最容易踩到的坑不是“收不到消息”，而是数据库已经提交，进程却在发送 ACK 前退出。连接关闭后，未确认消息可能再次投递；如果业务处理只是执行一次余额累加，重复消费就会变成重复入账。

腾讯云官方《RabbitMQ Message Reliability Use Cases》明确说明，消费者确认提供至少一次消费语义，客户端仍需实现幂等。本文只讨论消费端数据库副作用，不把生产者确认、消息持久化和消费成功混为一谈。

可靠的顺序是：校验事件 → 在同一个数据库事务内写去重记录并更新业务 → 提交事务 → 手动 ACK。下面用 Bun 内置 SQLite 做一个可直接运行的故障窗口演示，不需要云账号，也不声称它已经验证了真实 Broker 的网络行为。

一、先定义事件身份

去重键必须描述“同一件业务事件”，而不是“这次投递”。示例使用 event_id；生产者重试发送时必须复用它，不能每次生成新 ID。多租户、多个消费者共享表时，应把租户和消费处理器名称纳入唯一键，避免误吞其他业务事件。

还要约定：相同 ID 的载荷不可变。如果收到同一个 ID，却带来不同账户或金额，不能当作成功重复而静默忽略，应报警并进入人工处理或隔离流程。

二、运行一个最小事务实验

安装 Bun 后，将下列代码保存为 demo.js，执行 bun demo.js。金额使用整数最小货币单位，避免浮点误差。内存数据库只是为了让实验每次从干净状态开始；生产环境需要持久化数据库。

import { Database } from "bun:sqlite";

const db = new Database(":memory:");
db.exec(`
  CREATE TABLE account(id TEXT PRIMARY KEY, cents INTEGER NOT NULL);
  CREATE TABLE processed(
    event_id TEXT PRIMARY KEY, account_id TEXT NOT NULL, delta INTEGER NOT NULL
  );
  INSERT INTO account VALUES ('a', 0);
`);
const consume = db.transaction((eventId, accountId, delta) => {
  const result = db.query(
    "INSERT INTO processed VALUES (?, ?, ?) ON CONFLICT(event_id) DO NOTHING"
  ).run(eventId, accountId, delta);
  if (result.changes === 0) {
    const old = db.query(
      "SELECT account_id, delta FROM processed WHERE event_id=?"
    ).get(eventId);
    if (old.account_id !== accountId || old.delta !== delta) {
      throw new Error("event_id payload conflict");
    }
    return "duplicate";
  }
  const updated = db.query(
    "UPDATE account SET cents=cents+? WHERE id=?"
  ).run(delta, accountId);
  if (updated.changes !== 1) throw new Error("account missing");
  return "applied";
});

console.log(consume("evt-1", "a", 100));
// 模拟已提交但 ACK 丢失：再次处理同一事件。
console.log(consume("evt-1", "a", 100));
console.log(db.query("SELECT cents FROM account WHERE id='a'").get().cents);
try {
  consume("evt-1", "a", 200);
} catch (err) {
  console.log(err.message);
}
db.close();

三、把事务接到手动确认回调

示例中的 consume 函数正常返回，代表本次事务已提交，或者已核实这是完全相同的历史事件。这两种情况都可以确认当前投递。SQL 抛异常、连接断开，或者无法确定提交是否成功时，都不能抢先 ACK；让消息在连接恢复后重投，并通过去重表判断实际结果。

ACK 使用当前通道的 delivery_tag；它不是业务事件 ID，不可保存后跨通道复用。最小实现应单条确认，避免批量确认顺带确认尚未完成的其他消息。还必须关闭自动确认，否则回调执行失败时，消息可能已经被认为消费完成。

对于暂时性数据库故障，可以采用有退避和次数上限的重试策略。对于格式错误、ID 冲突等永久性错误，不要无限立即重新入队；应按已验证的死信或隔离机制处理。不能假设一次 reject 就会自动留存消息，是否有死信去向取决于实际队列配置。

四、验收要覆盖故障窗口

至少验证三条路径：事务提交前失败，业务更新和去重记录一起回滚；事务提交后、ACK 前失败，重复投递不再次产生副作用；同一事件 ID 携带不同载荷，明确报错而不是静默成功。

下面的程序覆盖后两条数据库行为。真实接入时，还应在事务提交后暂停消费者，再终止进程，观察重新启动后消息重投而业务金额不变。不要只看队列深度下降：还要检查事件记录、业务金额和错误处理结果是否一致。

五、边界比代码更重要

• 去重记录与业务表必须处于同一个可原子提交的数据库事务。先写 Redis 去重标记再更新数据库，不能提供这个保证。
• 示例以单连接顺序执行演示逻辑。并发上线需要数据库唯一约束兜底，并按目标数据库语义处理冲突、锁等待和事务重试，不能只依赖“先查询再插入”。
• 清理去重记录要覆盖消息最大重放周期，包括积压、死信重放和人工补偿。清得过早，旧消息就会再次生效。
• 发邮件、调用支付接口等外部副作用不在本地事务内。需要外部接口的幂等键，或结合事务发件箱等机制，不能把本例包装成跨系统的 exactly-once。

这套方案保证的是既定去重范围内，数据库业务副作用不会因为同一事件重投而重复提交，而不是保证消息永远只投递一次。

参考

• 腾讯云：RabbitMQ 消息可靠性实践
• 腾讯云：什么是消息队列 RabbitMQ 版