TDMQ CMQ 消费幂等：用事务堵住“业务成功、删消息失败”的窗口

结论：收到一次，不等于只执行一次

CMQ 的消费确认不能替代业务幂等。消费者完成数据库更新后，可能在删除消息前崩溃；消息重新可见后，另一消费者会再次收到它。正确顺序是：用稳定业务事件 ID 去重，将去重记录与业务变更放进同一个数据库事务，提交成功后再删除消息。

本文使用 Bun 内置 SQLite 演示这个原子性边界。示例不连接云端，不声称已经完成 CMQ 联调；目标是先在本地验证最容易写错的重复投递路径。

1. 先确认三个接口语义

根据 TDMQ for CMQ 官方 HTTP 数据流接口文档：

• 调用 ReceiveMessage 后，消息进入 inactive 状态，不可见时长由队列属性 visibilityTimeout 指定。超过时间仍未删除，消息会重新变为 active。
• 返回的 receiptHandle 是本次消费的句柄，不是长期有效的业务标识。删除时使用最近一次消费得到的句柄；超过不可见时间后，旧句柄失效。
• 至少一次消费不保证业务幂等，去重需要由业务实现。

因此不要用 receiptHandle 当幂等键。生产者应为同一个业务事件保留相同的 event_id，重试发送时也不能重新生成。不同事件则必须使用不同 ID，例如订单的“创建”和“退款”不能只共用一个订单号作为去重键。

2. 最小可复现：积分只增加一次

安装支持 bun:sqlite 的 Bun 运行时后，保存为 demo.js，执行 bun demo.js，无需安装第三方包。

import { Database } from "bun:sqlite";

const db = new Database(":memory:");
db.exec(`
CREATE TABLE account(id INTEGER PRIMARY KEY, balance INTEGER NOT NULL);
INSERT INTO account VALUES(1, 0);
CREATE TABLE processed(event_id TEXT PRIMARY KEY, points INTEGER NOT NULL);
`);
const apply = db.transaction((eventId, points, fail = false) => {
  const r = db.query(
    "INSERT INTO processed VALUES (?, ?) ON CONFLICT(event_id) DO NOTHING"
  ).run(eventId, points);
  if (r.changes === 0) {
    const old = db.query("SELECT points FROM processed WHERE event_id=?").get(eventId);
    if (old.points !== points) throw new Error("event ID payload conflict");
    return;
  }
  db.query("UPDATE account SET balance=balance+? WHERE id=1").run(points);
  if (fail) throw new Error("failure before commit");
});
function check(balance, count) {
  const a = db.query("SELECT balance FROM account").get();
  const p = db.query("SELECT COUNT(*) AS n FROM processed").get();
  if (a.balance !== balance || p.n !== count) throw new Error("unexpected state");
}
try { apply("event-1", 10, true); } catch (e) {
  if (e.message !== "failure before commit") throw e;
}
check(0, 0);
apply("event-1", 10);
apply("event-1", 10);
check(10, 1);
console.log("balance=10, processed=1");
db.close();

预期输出为 balance=10, processed=1：余额增加 10，去重记录只有一条。第二次调用模拟“数据库已提交，但删除消息未完成”后的重复投递；fail=true 则模拟事务提交前发生异常。

示例用内存数据库隔离实验，每次启动都会清空。生产环境必须使用持久化业务数据库，并由所有消费者共享同一去重约束；不能把每个进程独立的 SQLite 文件当作分布式幂等设施。

3. 为什么必须在同一个事务里

如果先单独写入去重记录，再更新业务，中途失败就会留下“已处理”的假记录，重试会跳过尚未完成的业务。反过来，先提交积分更新、再写去重记录，则会留下重复加分的窗口。

示例的主键冲突只处理 event_id 重复，其它 SQL 错误仍然抛出。事务回滚时，积分变化和去重记录一起撤销。不能把数据库异常一律捕获后当作重复成功，否则连接故障、约束错误也可能被错误确认。

示例额外保存 points，发现同一个事件 ID 携带不同积分时直接报错。真实系统可以保存关键业务字段或规范化载荷摘要，防止上游误复用 ID 被静默吞掉。

4. 接入 CMQ 时的确认顺序

消费者收到消息后，先解析并校验业务事件，再执行事务。事务返回成功，或确认是载荷一致的已处理事件，才使用本次 receiptHandle 调用 DeleteMessage。删除失败不能反向撤销已经提交的业务，也不能无条件再加一次积分；应记录错误，允许后续重复投递进入同一幂等路径。

visibilityTimeout 应覆盖正常处理耗时并留出网络与提交余量，但调大它不是幂等方案。超时可能让两个消费者同时处理同一个事件，最终仍要依赖共享数据库的唯一约束和事务。实际数据库的锁等待、死锁与重试规则需要单独验证。

5. 上线前做三项故障演练

• 事务提交前终止处理：重试后业务应完成一次，失败尝试不留下去重记录。
• 提交后、删除前终止进程：重新消费后业务值不再增长，随后用新句柄完成删除。
• 处理时间超过不可见窗口：即使出现并发重复投递，业务累计结果仍只有一次。

还有两个边界：去重记录不能随意提前清理，否则历史消息重放可能再次生效；跨数据库、发邮件或调用外部支付接口，不在这个本地事务的保护范围内，需要下游幂等键或事务发件箱等额外设计。

参考文档

• 腾讯云：Consuming Message
• 腾讯云：Deleting Message

本文按上述 HTTP 数据流接口说明讨论，接入其它 SDK 或消息产品时，应重新核对字段映射与确认语义，不直接套用接口名称。