消息幂等:重复投递的来源与去重

消费端造成的重复

实测 · 本机执行输出

  消费者 A 处理了 order-1001,redelivered=False
  消费者 B 处理了 order-1001,redelivered=True
  同一条订单被业务处理了 2 次

处理完业务、ack 之前崩溃(或 ack 丢失),broker 会重投该消息,标志 redelivered=True。

发布端造成的重复

实测 · 本机执行输出

  第一次发布(msg-2)的结果: StreamLostError(发布端据此认为“没发成功”)
  这时队列里实际已经有: 1 条
  发布端重连后重发同一条(同样的 message_id=msg-2)并收到 Ack;队列里现在有: 2 条
  取出来: [('order-1002', False, 'msg-2'), ('order-1002', False, 'msg-2')]  ——两份都是 redelivered=False:redelivered 标志抓不住发布端造成的重复

消息已到 broker 但 Ack 没回来,发布端重试,队列里就有两份;这两份的 redelivered 都是 False,所以不能靠这个标志去重。

broker 不去重

实测 · 本机执行输出

  收到 3 条(message_id 只是一个普通属性,RabbitMQ 不会用它去重)

message_id 只是属性,RabbitMQ 不用它去重。

幂等消费

实测 · 本机执行输出

  朴素消费者:收到 1100 次投递,业务执行 1100 次,余额 = 1100(应为 1000)
  幂等消费者(message_id 唯一约束):收到 1100 次投递,业务执行 1000 次,余额 = 1000
  耗时: 朴素 36 ms,幂等 37 ms(幂等版多一次 insert,sqlite 内存库,两者差别在噪声内)

用 message_id(或业务上唯一的键)做唯一约束,和业务操作放在同一个数据库事务里;重复的消息插入失败,业务就不会再执行。

代价

实测 · 本机执行输出

  100 万个 36 字符的 message_id:插入 787 ms,库大小约 28.3 MiB(sqlite,without rowid)

去重记录要保留到重复可能出现的窗口之外,100 万个 id 约 28 MiB,需要定期清理。

分步演示

消息幂等

本步的对象

本文用到的工具与边界

本机 macOS 27 arm64 / RabbitMQ 4.3.6 / Erlang/OTP 28.5 / pika 1.4.4(Homebrew 与临时虚拟环境);所有节点都是本机自己起的 RabbitMQ 进程(127.0.0.1,端口 9500 以上,每个实验独立的数据目录与节点名,不连任何外部服务);命令输出来自真实的 rabbitmqctl 与管理 API,发布与消费用 pika 的真实连接;计时用 time.perf_counter,数字每次运行会有小幅抖动。第二节经自己写的 TCP 代理:S→C 方向延迟 300 ms,100 ms 后 cut 连接;幂等实验用 sqlite3 内存库模拟业务库;耗时是整个消费 1100 条的时间,单次
没有验证:stream 队列的发布端去重(需要 stream 协议的 producer name);数据库事务与消息 ack 之间的更严格一致(见第 35 篇 Outbox);去重表在分布式数据库、Redis 上的实现;去重窗口的清理策略;多个消费者并发处理同一个 message_id 时的竞争(依赖唯一约束)

下一篇 · RABBITMQ 26

消息顺序:多消费者下顺序怎么保持