消息重复与乱序:消费端怎么处理

重复与丢失

实测 · 本机执行输出

  确认方式                      | 消费者被杀之后 broker 里            | 处理过的不同消息 | 丢失  | 丢失的编号   | 被处理多次      
  --------------------------+-----------------------------+----------+-----+---------+------------
  先处理、再 ack(至少一次)           | 被杀时队列:ready 200 / unacked 0 | 300      | 0   | -       | 1(消息 [100])
  先 ack、再处理(至多一次)           | 被杀时队列:ready 199 / unacked 0 | 299      | 1   | 100~100 | 0          
  auto-ack(broker 一发出去就算成功) | 被杀时队列:ready 0 / unacked 0   | 101      | 199 | 101~299 | 0          
  先处理再 ack:被杀时正在处理的那一条没有 ack,broker 把它重新投递给新消费者,它被处理了两次;先 ack 再处理:被杀时那一条已经确认却没处理,丢了;auto-ack:broker 已经把一大批消息推给了这个消费者,它们全丢

先处理再 ack 保证不丢,崩溃会重复;先 ack 再处理崩溃会丢;auto-ack 把一大批已推送的消息一起丢掉。

乱序的来源

实测 · 本机执行输出

  (a) 单个消费者(prefetch=1),第 10 条被 nack 并重新入队一次:处理顺序 [8, 9, 10, 11, 12, 13]...,乱序的消息 0 条(重新入队的消息回到原来的位置,所以仍然是 10 先于 11)
  (b) 两个消费者竞争同一个队列(一个每条 20 ms,一个每条 1 ms),200 条:处理完成的顺序前 20 个 [0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 1, 17, 19, 20],乱序的消息 12 条
  (c) 两个消费者(每条 5 ms),其中一个拿到第 50 条后卡了 100 ms 才 nack 重新入队:第 50 条是第 67 个处理完的,排在它前面处理完的编号比 50 大的有 16 条(最大到 66),乱序的消息共 28 条

队列本身先进先出,乱序来自多个消费者并行(快慢不同)和处理失败后重新处理;单消费者的重新入队不乱序。

Kafka 的顺序

实测 · 本机执行输出

  写法           | 收到   | 每个实体平均分布在几个分区 | 同一实体内顺序被打乱的实体数(共 20) | 实体内乱序的消息数 | 相对发布顺序整体乱序的消息数
  -------------+------+---------------+----------------------+-----------+---------------
  按实体 id 做 key | 2000 | 1.0           | 0                    | 0         | 1499          
  不设 key(随机分区) | 2000 | 3.0           | 20                   | 1322      | 1357          
  按实体 id 做 key:同一实体的消息都进同一个分区,实体内部有序;不同实体之间(跨分区)没有顺序可言。不设 key:同一实体的消息被撒到多个分区,实体内部的顺序也没有了

只有同一个分区内有序。按实体 id 做 key,同一实体进同一分区,实体内有序;不设 key 则实体内也乱。

版本号

实测 · 本机执行输出

  测试流:100 个实体,每个实体 20 条更新;每条消息带两部分内容——“把 status 设成 S<版本>”(SET 类,后到的新版本覆盖旧的)和“counter 加 1”(累加类,每条消息只能算一次);发布顺序是各实体的版本轮流,每 300 条之内随机重排(相当于相邻 3 个版本的消息混在一起),再随机插入 200 条(10%)重复,共 2200 条;正确的终态:每个实体 status = S20、counter = 20
  消费端的做法                   | status 最终不是 S20 的实体 | counter 最终不是 20 的实体
  -------------------------+---------------------+--------------------
  朴素:来一条改一条                | 58 / 100            | 92 / 100           
  只用版本号(WHERE version < ?) | 0 / 100             | 92 / 100           
  版本号能挡住乱序(旧版本晚到时被 WHERE 条件拒绝),status 全对;但累加类的 counter 不是“谁新谁覆盖”,重复消息每来一次就多加一次,版本号管不了

SET 类字段带上单调递增的版本号,处理时 WHERE version < ? 拒绝旧版本,乱序就不会覆盖新状态。

版本号 + 去重

实测 · 本机执行输出

  消费端的做法       | status 最终不是 S20 的实体 | counter 最终不是 20 的实体
  -------------+---------------------+--------------------
  朴素           | 58 / 100            | 92 / 100           
  只做去重(seen 表) | 55 / 100            | 0 / 100            
  只用版本号        | 0 / 100             | 92 / 100           
  去重 + 版本号     | 0 / 100             | 0 / 100            
  去重解决重复(累加类的 counter 对了),版本号解决乱序(SET 类的 status 对了);两种语义的字段各用各的办法,一起用才同时对

累加类字段靠去重(每条消息只算一次),SET 类靠版本号,两者一起才同时对。

分步演示

重复与乱序

本步的对象

本文用到的工具与边界

本机 macOS 27 arm64 / MySQL 26.7.0 与 PostgreSQL 16.15(Homebrew,每个实验自己 initdb、自己起实例,端口 9150 以上,只监听 127.0.0.1)/ Python 3.14.8 + PyMySQL 1.2.3 + psycopg 3.3.6 / Redis 8.10.2 / Kafka 4.3.1 / RabbitMQ 4.3.6 / etcd 3.7.2(按各篇注明用到哪些);不连任何外部服务,不碰本机已有的数据库实例;故障用 kill -9 进程、让连接断开或在代码里的指定位置抛异常来制造;计时用 time.perf_counter,数字每次运行会有小幅抖动。本篇用 RabbitMQ 4.3.6(pika 1.4.4,消费者是可以被 kill -9 的子进程)、Kafka 4.3.1 单节点 3 分区(confluent-kafka 2.16;不设 key 的情形由我给每条消息随机指定分区)、MySQL 26.7.0;乱序重排和重复是用固定种子的随机数在测试流里造的,重排窗口 300 条、重复 10% 是我设的
没有验证:Kafka 生产者重试在 max.in.flight 大于 1 且未开幂等时造成的乱序;RabbitMQ 的 quorum 队列、single active consumer 对顺序的保证;消费者组再平衡时的重复(只测了 RabbitMQ 的崩溃重投);Kafka 事务 / exactly-once 语义;真实业务里版本号怎么生成(这里是现成的)

下一篇 · 分布式事务 16

最终一致:对账与修复