《Java 消息中间件·可靠消息与消费治理》1.3 消费端：手动提交与位移管理

位移不是「处理了多少条」

Kafka 消费端的位移有两个不同的值：

• position：下一次 poll() 会从哪里开始拉，由客户端维护，随拉取推进；
• committed offset：消费组向 broker 承诺「这个位置之前我已经处理完了」，由提交动作写入 __consumer_offsets。

自动提交和手动提交的区别，不在于「多久提交一次」，而在于提交的是哪个值。自动提交提交的是 position，也就是客户端已经拉到的位置；它不知道你的业务处理到了第几条。手动提交让你自己决定提交哪个位置，通常应该是「最后一条处理成功的记录 offset + 1」。

这个区别在正常运行时看不出来，因为处理速度通常快过拉取。一旦处理变慢、批次变大、或者进程恰好在批次中间退出，两个值就会分叉。本篇把分叉做出来。

4.3.0 的默认值与实现路径

先看默认值（Temurin 25.0.5+8 / kafka-clients 4.3.0 / 2026-10-29，ConsumerConfig.configDef().defaultValues()）：

consumer.enable.auto.commit=true
consumer.auto.commit.interval.ms=5000
consumer.max.poll.records=500
consumer.isolation.level=read_uncommitted
consumer.group.protocol=classic
consumer.session.timeout.ms=45000
consumer.auto.offset.reset=latest

两个关键点：

1. 默认开启自动提交，间隔 5 秒；
2. group.protocol=classic，所以 4.3.0 默认走经典消费者实现 ClassicKafkaConsumer 加 ConsumerCoordinator，不是 KIP-848 的 AsyncKafkaConsumer。选择逻辑在 ConsumerDelegateCreator：只有显式配成 group.protocol=consumer 才返回 AsyncKafkaConsumer，否则返回 ClassicKafkaConsumer。下面引用的源码路径只对 classic 成立；把协议改成 consumer 后，自动提交的实现位置会变，不能照搬结论。

自动提交在 poll() 里按间隔触发，但还有一条更容易被忽略的路径：close()。ConsumerCoordinator.close 会先做一次同步自动提交：

public void close(final Timer timer, CloseOptions.GroupMembershipOperation membershipOperation) {
    // we do not need to re-enable wakeups since we are closing already
    client.disableWakeups();
    try {
        maybeAutoCommitOffsetsSync(timer);

maybeAutoCommitOffsetsSync 提交的内容是：

private void maybeAutoCommitOffsetsSync(Timer timer) {
    if (autoCommitEnabled) {
        Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();

subscriptions.allConsumed() 返回的是已拉取到的位置，不是已处理完的位置。ClassicKafkaConsumer.close 在关闭协调器时调用这段逻辑（ClassicKafkaConsumer.java 第 1160 行附近），所以一次正常的 consumer.close() 就可能把没处理完的记录提交掉。

这就是下面要复现的核心：不是崩溃，不是网络故障，而是优雅关闭。

实验：自动提交在 close 时跳过 15 条

实验环境同前：Temurin 25.0.5+8、kafka-clients 4.3.0、apache/kafka:4.3.0，topic r13.commit 为单分区单副本。先发 40 条：

topic=r13.commit acks=all retries=3
ok=40 fail=0 offsetMin=0 offsetMax=39

消费者配置 max.poll.records=20，第一次 poll() 拉 20 条；程序只处理前 5 条，然后调用 close()。为了排除「处理期间正好赶上周期提交」的干扰，auto.commit.interval.ms 显式设为 60000；close() 的同步提交不受这个间隔限制。

运行输出：

mode=auto-close group=r13-auto-close polled=20
processed off=0 value=v0
processed off=1 value=v1
processed off=2 value=v2
processed off=3 value=v3
processed off=4 value=v4
processedCount=5 position=20 remainingInBatch=15
closed=true autoCommitOnClose=true

processedCount=5、position=20、remainingInBatch=15。关闭后查消费组：

group=r13-auto-close topic=r13.commit
committed=20 logEnd=40 lag=20

已提交位移是 20，不是 5。用同一个组、关闭自动提交、从已提交位置继续读，看看重启后还能读到什么：

group=r13-auto-close reread=20
firstOffset=20 lastOffset=39

重启后从 20 开始，读 20 条。offset 5 到 19 这 15 条，从未被处理，也永远不会再被这个组读到：

监控上没有任何异常：消费组 lag 从 40 降到 20，进程正常退出，提交成功。丢掉的 15 条不会触发任何告警。

对照一：同场景改手动提交

同样的 20 条、只处理前 5 条，改成 enable.auto.commit=false，处理完 5 条后显式提交位移 5：

mode=manual-close group=r13-manual-close polled=20
processedCount=5 position=20 remainingInBatch=15
closed=true manualCommitOffset=5

group=r13-manual-close topic=r13.commit
committed=5 logEnd=40 lag=35
group=r13-manual-close reread=35
firstOffset=5 lastOffset=39

重启后从 5 开始，35 条全部可重读。处理过的 5 条会重复，但没有任何一条被跳过。这就是 at-least-once：宁可重，不可丢。

对照二：自动提交但进程被杀

同样的 20 条、处理 5 条后直接 halt(9)，不走 close()：

mode=auto-kill group=r13-auto-kill polled=20
processedCount=5 position=20 remainingInBatch=15
halt(9) without close/commit

group=r13-auto-kill topic=r13.commit
committed=-1 logEnd=40 lag=40
group=r13-auto-kill reread=40
firstOffset=0 lastOffset=39

自动提交没有机会运行，已提交位移是 -1，重启后 40 条全部重读，前 5 条重复。三种写法放在一起：

这张表说明自动提交的两个方向：它既可能丢，也可能重，取决于进程是怎么退出的。优雅退出会提交拉取位置，造成跳过；崩溃退出会丢掉未提交的进度，造成重放。它唯一不会做的事，是保证「提交的位置等于处理完的位置」。

先处理后提交 vs 先提交后处理

两种顺序对应两档语义，代码形态只差一行。

先处理后提交，是 at-least-once：

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        handle(r);                 // 先处理
    }
    consumer.commitSync();         // 再提交整个批次
}

失败窗口是「处理成功、提交失败」：重启后这批记录会重读，已处理的会再处理一次。重复范围最多一个 poll 批次，默认 max.poll.records=500。

先提交后处理，是 at-most-once：

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    consumer.commitSync();         // 先提交
    for (ConsumerRecord<String, String> r : records) {
        handle(r);                 // 再处理；这里崩溃就永久跳过
    }
}

失败窗口是「提交成功、处理失败」：重启后从提交位置继续，未处理的记录被跳过。它适合「丢了也无所谓」的场景，比如可重建的缓存刷新；不适合订单、借阅、库存这类有业务后果的写入。

如果想缩小重复范围，可以按记录提交：

Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
for (ConsumerRecord<String, String> r : records) {
    handle(r);
    offsets.put(
        new TopicPartition(r.topic(), r.partition()),
        new OffsetAndMetadata(r.offset() + 1));   // 注意 +1
}
consumer.commitSync(offsets);

代价是提交请求变多，协调器压力变大。折中是每 N 条提交一次，或者按处理耗时提交。

+1 这个细节值得单独说：Kafka 的提交语义是「下一次从哪个 offset 开始读」。处理完 offset=4 之后，要提交 5，不是 4。提交 4 会让 offset=4 再被处理一次；这类 off-by-one 在重复量统计里经常被误认为「Kafka 又重复投递了」。

提交失败怎么重试

commitSync 相关的异常可以分成三类，处理方式完全不同。

CommitFailedException 的 javadoc 写得很明确：提交无法应用，通常因为再平衡已经完成，部分分区可能已经分配给别的成员。这时候继续用同一个消费者实例提交，只会继续失败；正确做法是让 poll() 重新加入组，或直接在再平衡监听器里处理。

可重试异常则应该退避重试，但要受 max.poll.interval.ms（默认 300000）约束：重试期间不调用 poll()，超过这个时限组会把你踢出，下一次提交就变成 CommitFailedException。

一个可以直接套用的重试骨架：

private void commitWithRetry(KafkaConsumer<String, String> consumer,
                             Map<TopicPartition, OffsetAndMetadata> offsets) {
    long backoffMs = 100;
    while (true) {
        try {
            consumer.commitSync(offsets);
            return;
        } catch (CommitFailedException e) {
            throw e;                       // 已再平衡，原地重试没有意义
        } catch (WakeupException e) {
            throw e;                       // 停机信号
        } catch (RetriableException e) {
            try {
                Thread.sleep(backoffMs);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(ie);
            }
            backoffMs = Math.min(backoffMs * 2, 2000);
        }
    }
}

实际代码还应该给这个循环加总时限，避免在协调器长时间不可达时无限退避；重试的总时长同样受 max.poll.interval.ms 约束。

commitAsync 则是另一种陷阱：它不会自动重试，失败只通过回调告诉调用方。如果一次提交 20 失败、随后又提交了一个更小的位移 10 并成功，已提交位移会往回退。异步提交适合「下一次提交会覆盖这一次」的场景，不适合作为唯一提交手段。它的细节在第 12.2 节展开。

位移管理的五个细节

一、提交的是「下一个要读的位置」

处理完 offset=4，提交 5。所有手动提交的 map 都要用 record.offset() + 1。批量提交时，取批次里每个分区最后一条记录的 offset 加 1。

二、按分区维护位移

一个 poll 批次可能包含多个分区。提交时要么用 commitSync() 提交所有已拉取位置（简单但粒度粗），要么为每个分区算出「最后处理成功的位置 + 1」。不要用一个全局变量记录「处理到第几条」，分区之间的位移是独立的。

三、再平衡监听器里不要做重活

ConsumerRebalanceListener.onPartitionsRevoked 在再平衡前调用，适合提交当前已处理位移、释放资源。这里做长时间处理会拖慢再平衡，甚至触发下一轮超时。提交本身也可能抛 CommitFailedException，要捕获并记录，不要让异常冒泡打断再平衡流程。

四、max.poll.interval.ms 是处理时长的上限

两次 poll() 之间的间隔超过 max.poll.interval.ms，消费者会被踢出组，分区被回收。默认 300 秒，看起来很长，但如果一个批次 500 条、每条处理 1 秒，就会超。调参方向是减小 max.poll.records 或把处理移到工作线程，而不是盲目加大这个值。

五、手动提交不等于不重复

处理完、提交前崩溃，仍会重读。手动提交只是把「提交时机」的控制权拿回来，重复窗口依然存在。要消除重复的副作用，需要业务幂等、去重表或状态机，这是第 2 章的内容。

本节判据

回答四个问题：

1. 自动提交提交的是处理进度吗？——不是。它提交 position，即客户端已拉取到的位置。close() 时的同步自动提交会把整个批次的拉取位置写进去。
2. 自动提交在什么情况下会丢？——进程优雅关闭、且批次只处理了一部分时。本机实测 20 条处理 5 条，close() 提交 20，15 条永久跳过。
3. 自动提交在什么情况下会重？——进程崩溃、未到提交间隔或未走 close() 时。本机实测 halt(9) 后已提交位移为 -1，40 条全部重读。
4. CommitFailedException 能原地重试吗？——不能。它表示组已再平衡或分区已不属于当前消费者，必须重新加入组；只有 RetriableException 才适合退避重试。

下一节进入第 2 章：既然 at-least-once 一定会重复，就先弄清楚重复到底有几种来源，再决定用唯一键、状态机还是去重表来接。

阅读导航：上一节 1.2 生产端：acks、重试与幂等 ｜ 下一节 2.1 重复的四种来源