《Java 消息中间件·可靠消息与消费治理》2.1 重复的四种来源

「重复」到底指什么

讨论重复之前先统一口径。Kafka 的日志里允许出现两条内容相同的记录，这不是故障；真正的故障是同一个业务事件被业务逻辑执行了两次。前者是存储事实，后者是副作用事实。

这两件事经常被混为一谈，于是排查时会陷入「日志里明明只有一条，为什么库存扣了两次」或者「日志里有两条，为什么账是对的」。要分清它们，得先知道重复是从哪一段进来的。

一条消息从生产到落库，至少经过四段：生产者、broker、消费者、下游存储。能制造重复的位置有四个，机制各不相同：

这张表是本节的主线。下面用同一套 libraryd 主题，把这四种来源逐个构造出来，并记录它们各自留下的痕迹。下面的输出均为本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29），单次运行，随调度波动，不作基准。

先排除三种「看起来像重复」的情况

排查重复的第一步不是找去重方案，而是确认它真的是重复。至少三种情况会被误判：

• 同一业务主键的两次合法事件：同一本书被两个会员先后借走，bookId 相同、loanId 不同。这是两笔借阅，不是重复。去重键如果选 bookId，会把正常业务丢掉。
• 重试时重新生成了事件 ID：上游每次重试都 UUID.randomUUID()，offset 不同、eventId 也不同，但业务上是同一笔。这不是传输重复，是事件 ID 生成位置错了。
• 指标口径重复：同一条记录被两个消费者组各消费一次，监控把两个组的计数相加。Kafka 侧没有重复，是统计方式重复。

这三种情况的共同点是：判断重复的键选错了。键必须来自业务事实，且在重试时保持不变。下面的四类来源都假定键已经选对。

实验环境与构造方式

环境：Temurin 25.0.5+8，kafka-clients 4.3.0，apache/kafka:4.3.0 单节点 KRaft，宿主机 4 vCPU / 7.7 GiB。三个主题：

• library.loan.created：2 个分区，7 条借阅创建事件，用于再平衡与位移回退；
• library.loan.retry：1 个分区，6 条事件，其中第 3 条被业务层重发一次；
• library.loan.replay：1 个分区，5 条事件，用于下游重放。

生产者统一 acks=all、enable.idempotence=true、linger.ms=0。消费者统一手动提交、auto.offset.reset=earliest。每次运行用带时间戳后缀的消费组名，避免上一轮提交的位移干扰结果。代码在实验目录的 DuplicateSourcesLab.java，下面只摘关键段落。

来源一：生产重试

生产侧的重复有两种，必须分开说。

第一种是客户端重试：请求发出去了但响应丢了，客户端用同一个批次、同一个序列号重发。这种重复由 broker 按 (PID, 分区) 的序列号吸收，卷一 3.2 已经用批次头里的 producerId 与 baseSequence 说明过。开启 enable.idempotence=true 后，它不会变成两条记录。

第二种是业务层重试：调用方没拿到结果，把整个「构造事件 + send」的流程又跑了一遍。这时事件体相同，但批次是新的，序列号也是新的，幂等生产者没有理由拒绝它。本实验构造的是第二种：

for (int i = 1; i <= 6; i++) {
    send(p, RETRY_TOPIC, null, "book-" + i, event("loan-" + i, "book-" + i, 2000 + i));
    if (i == 3) {
        // 业务代码没拿到响应，整段重试：eventId 不变
        send(p, RETRY_TOPIC, null, "book-3", event("loan-3", "book-3", 2003));
    }
}

生产结果的 offset 分布：

== produce app-retry == records=7 [loan-1@0, loan-2@1, loan-3@2, loan-3-retry@3, loan-4@4, loan-5@5, loan-6@6]

再用 kafka-dump-log.sh 看落盘记录：

docker exec mq-rel2-kafka sh -c 'unset JMX_PORT; /opt/kafka/bin/kafka-dump-log.sh \
  --files /tmp/kafka-logs/library.loan.retry-0/00000000000000000000.log --print-data-log'

关键两行（省略了时间戳与 CRC）：

baseOffset: 2 lastOffset: 2 count: 1 baseSequence: 2 lastSequence: 2 producerId: 11 producerEpoch: 0
| offset: 2 key: book-3 payload: eventId=2003|loanId=loan-3|bookId=book-3|type=LOAN_CREATED
baseOffset: 3 lastOffset: 3 count: 1 baseSequence: 3 lastSequence: 3 producerId: 11 producerEpoch: 0
| offset: 3 key: book-3 payload: eventId=2003|loanId=loan-3|bookId=book-3|type=LOAN_CREATED

两条记录属于同一个 producerId=11，序列号分别是 2 和 3，eventId 都是 2003。同一个生产者、连续序列号、不同 offset、相同 eventId，这是业务层重试的指纹。broker 不会把它当成重试批次，因为序列号没有重复。

识别方法：拿处理日志里的 (topic, partition, offset) 与 eventId 对照。offset 不同但 eventId 相同，基本可以判定为生产侧重复；如果 offset 不同、eventId 也不同，但业务主键相同，那是上游生成了两个事件 ID，属于事件建模问题，不是传输重复。

来源二：消费者再平衡

再平衡导致重复的条件是：消费者处理了记录，但还没提交位移，分区就被收走了。新 owner 从已提交位移开始读，未提交的那一段自然重来一遍。

实验用 2 个分区、2 个消费者。消费者 A 先订阅，拿到全部分区，先处理 7 条记录，但只提交 p0=2、p1=1：

List<ConsumerRecord<String, String>> first = pollUntil(a, 7, Duration.ofSeconds(10));
// 处理了 7 条，只提交一部分
a.commitSync(Map.of(
        new TopicPartition(CLEAN_TOPIC, 0), new OffsetAndMetadata(2),
        new TopicPartition(CLEAN_TOPIC, 1), new OffsetAndMetadata(1)));

随后消费者 B 加入同一个组。默认的 range 分配器把 p0 留给 A、p1 给 B。raw 输出：

== rebalance == A processed=[A@p0@o0, A@p0@o1, A@p0@o2, A@p0@o3, A@p1@o0, A@p1@o1, A@p1@o2] committed={p0=2,p1=1}
== rebalance == afterRebalance processed=[B@p1@o1, B@p1@o2, A@p0@o2, A@p0@o3] aAssign=[library.loan.created-0] bAssign=[library.loan.created-1]
== rebalance == duplicateOffsets=[p0@o2, p0@o3, p1@o1, p1@o2] count=4 byConsumer=[B@p1@o1, B@p1@o2, A@p0@o2, A@p0@o3]

四个重复分两种情形：

• p1 从 A 搬到了 B，B 从 committed=1 开始，重读 p1@o1、p1@o2；
• p0 仍归 A，但 eager 再平衡会先撤销全部分区再重新分配，A 的位置也被重置到 committed=2，于是重读 p0@o2、p0@o3。

第二点最容易被忽略：很多人以为「分区没搬走就不会重复」。eager 协议下，即使分区留在原地，位置也会回到已提交位移；cooperative 协议只撤销要搬走的分区，留在原地的分区不受影响。这个差异在卷一 6.2 用协议时间线拆过，这里只强调它对重复的影响。

提交时机决定重复范围

再平衡本身不制造重复，制造重复的是「处理了但没提交」。同样一次再平衡，提交策略不同，重复范围差很多：

onPartitionsRevoked 里做一次同步提交可以把范围压到「回调之前刚处理完、还没来得及提交」的那几条。但回调本身可能失败，所以它只能缩小重复，不能消除重复。真正的兜底仍然是消费端幂等。

识别方法：重复项的 (topic, partition, offset) 完全相同，但处理者不同（或同一处理者的两次运行）。如果处理日志里带了消费者实例 ID 或 rebalance generation，会看到 generation 变化。没有 generation 时，看 offset 是否相同就够了：offset 相同、eventId 相同，说明是消费侧重读，不是生产侧重发。

需要说明实验的构造方式：两个 KafkaConsumer 跑在同一个 JVM 里，对协调器来说就是两个成员，协议行为与两个进程一致，只是省掉了进程启动。组名带时间戳，确保从零位移开始。

来源三：位移回退

位移回退是再平衡的近亲，但触发动作不同：不是分区换 owner，而是有人把位置往回拨。生产环境里常见于三种操作：代码里 seek()、运维执行 --reset-offsets、消费组被删除后按 auto.offset.reset 重新开始。

实验用 assign 直接控制位置，先 seek 到 1 读三条，再 seek 回 1 读同样的三条：

c.assign(List.of(P0));
c.seek(P0, 1);
List<String> pass1 = pollUntil(c, 3, Duration.ofSeconds(8));
c.seek(P0, 1);
List<String> pass2 = pollUntil(c, 3, Duration.ofSeconds(8));

输出：

== seek-rollback == pass1=[p0@o1, p0@o2, p0@o3] pass2=[p0@o1, p0@o2, p0@o3]
== seek-rollback == duplicateOffsets=[p0@o1, p0@o2, p0@o3] count=3

两次读到的是同一批 offset。Kafka 没有产生任何新记录，log-end-offset 不变，producerId 也不变。重复完全发生在消费侧。

识别方法：与再平衡一样看 (topic, partition, offset)；区别在于位移回退通常伴随一次显式的位移变更记录。如果代码里没有 seek，就要查运维操作记录、消费组重置记录，以及 auto.offset.reset 是否被触发过。监控上，current-offset 突然变小而 log-end-offset 没变，就是一个强信号。

四种常见的位移回退

最后一种最隐蔽：消费者并没有主动回退，是它要读的位移已经不在日志里了。Kafka 只能按 auto.offset.reset 重新定位，earliest 会把整个主题重读一遍。保留期短于消费者停机时间时，这种情况迟早会发生。

不要把它和「消费者重启」混为一谈。正常重启会从已提交位移继续，不产生重复；只有「已处理、未提交」的那一段才会重来。位移回退则是把已提交的位置也往回拨，重复范围更大，可能覆盖已经确认过的历史。

来源四：下游重放

前三种重复都能在 Kafka 侧找到痕迹：要么多了一条记录，要么同一 offset 被读了两次。第四种在 Kafka 侧完全干净，只有下游存储知道。

典型场景是重建投影：新上线一个查询服务，需要从 library.loan.replay 从头消费一遍，把结果写进读库。如果写入逻辑是「追加一行」，重放一次就多一份；如果是「按主键覆盖」，重放不改变结果。

实验用两个消费者组分别从头读同一个主题，模拟两次独立重放，写入两个投影：

for (int pass = 1; pass <= 2; pass++) {
    try (KafkaConsumer<String, String> c = consumer("dup-replay-lab-" + RUN + "-" + pass)) {
        c.assign(List.of(tp));
        c.seekToBeginning(List.of(tp));
        for (ConsumerRecord<String, String> r : pollUntil(c, 5, Duration.ofSeconds(8))) {
            naive.add(field(r.value(), "loanId"));   // 追加式投影
            upsert.put(field(r.value(), "loanId"), r.value()); // 覆盖式投影
        }
    }
}

输出：

== replay == naiveEntries=10 distinctLoanId=5 upsertEntries=5 duplicateEntries=5

源主题只有 5 条记录，两次重放后追加式投影变成 10 行，覆盖式投影仍是 5 行。Kafka 侧没有新增记录，offset 也没变；重复只存在于下游的写入方式里。

识别方法：先确认 Kafka 侧有没有新记录。如果 log-end-offset 没涨、消费者组也没重置，但下游行数涨了，基本就是重放或重复调用。再看下游的写入语义：INSERT 还是 UPSERT，有没有唯一约束，有没有按 eventId 去重。重放本身不是错误，错误是下游没有为重放做好准备。

重放做对，靠的是写入语义

重放不可避免：新投影上线、补历史、换存储都需要。能做的是让重放不改变结果：

• 按业务主键 UPSERT，而不是 INSERT；
• 目标表加唯一约束，让第二次写入失败而不是产生第二行；
• 重放写入影子表，校验后原子切换；
• 重放通道单独限速，避免把在线流量打满。

前三项里至少要有两项。只有「重放前先清空目标表」这一种做法，在重放中断后无法判断清到了哪里，不建议单独使用。

一张排查判据表

把四种来源放在一起，判据其实只有三个问题：

1. Kafka 里有没有新增记录？
2. 两条重复的 (topic, partition, offset) 是否相同？
3. 两条重复的 eventId 是否相同？

这张表能解释大部分现场。剩下的边角情况要单独判：

• 生产重试但每次重新生成 eventId：offset 不同、eventId 也不同，只能靠业务主键识别。这是事件建模缺陷，应在生产端修，不是在消费端猜。
• 再平衡叠加生产重试：offset 不同且 eventId 相同，但重复出现在两个消费者上。先按生产重试处理，再看消费侧有没有重复处理同一 offset。
• 下游重放叠加非幂等写入：Kafka 侧完全正常，只有对账能发现。这类问题只能靠下游的唯一键或幂等表兜住。

判据能定位来源，但不能替代修复。四种来源对应的修复手段并不相同：生产重试要靠业务幂等或事件 ID 去重；再平衡与位移回退要靠「先处理后提交」与去重表；下游重放要靠覆盖式写入或重放专用通道。下一节把这三种幂等实现拆开，并给出一份可以运行的 IdempotentConsumer。

一个最小处理账本

要在事后区分四类重复，消费端至少要留下三列：eventId、(topic, partition, offset)、consumerId。写入业务库时，这三列和处理结果放在同一个事务里，账本才有意义。

CREATE TABLE consume_ledger (
  event_id    VARCHAR(64) PRIMARY KEY,
  topic       VARCHAR(128) NOT NULL,
  partition   INT NOT NULL,
  offset      BIGINT NOT NULL,
  consumer_id VARCHAR(64) NOT NULL,
  applied_at  TIMESTAMP NOT NULL
);

有了它，四类重复的查询方式不同：

• 按 event_id 分组计数大于 1：生产侧重试或下游重放；
• 按 (topic, partition, offset) 分组计数大于 1：再平衡或位移回退；
• 同一 event_id 对应多个 offset：生产侧重发；
• 同一 offset 对应多个 consumer_id：分区换过 owner。

账本本身也会成为瓶颈，所以生产环境通常只保留 TTL 窗口内的记录，或者把它并进业务表的唯一索引。无论哪种，判据不变。

回到 libraryd

libraryd 的六个事件里，重复的代价不一样，处理优先级也不一样：

一个共同点：这些 key 都来自业务，不来自消费者。消费者生成的 UUID 每次都不一样，起不到去重作用。事件 ID 必须在生产端生成，并随事件一起传输；消费端只负责「见过没有」。

下一节从这个前提开始，先讲唯一键怎么选，再把去重表、TTL、容量和状态机拼成一个完整的消费者。

阅读导航：上一节 1.3 消费端：手动提交与位移管理 ｜ 下一节 2.2 幂等消费：唯一键、状态机与去重表