bigywater.tech
Spring · 140

Spring Kafka:监听容器与 offset 提交方式

Spring Kafka 的偏移量到底什么时候提交?AckMode 选哪个?进程被杀之后会重复处理几条?MANUAL 里 ack 了为什么偏移量没动?concurrency 设成 10 就能 10 倍并发吗?这篇在 Spring Kafka 4.0.7 和真实的 Kafka 上逐项实测。

MANUAL 不立刻提交
1.0 秒时已处理完 3 条并调用了 ack,但已提交偏移量仍然是 None;MANUAL_IMMEDIATE 才是 3
4 对 1
kill -9 之后,BATCH 与 RECORD 分别被重复处理的消息条数
3 个线程
concurrency=5 但只有 3 个分区时,真正干活的消费者线程数

提交时机

实测 · 本机执行输出 
  [BATCH(默认)] 开始处理后 1.0 秒:已处理完 3 条,已提交偏移量 None;全部处理完(10 条)之后已提交偏移量 10
  [RECORD] 开始处理后 1.0 秒:已处理完 3 条,已提交偏移量 3;全部处理完(10 条)之后已提交偏移量 10
  [MANUAL(处理完调用 ack.acknowledge())] 开始处理后 1.0 秒:已处理完 3 条,已提交偏移量 None;全部处理完(10 条)之后已提交偏移量 10
  [MANUAL_IMMEDIATE(处理完调用 ack.acknowledge())] 开始处理后 1.0 秒:已处理完 3 条,已提交偏移量 3;全部处理完(10 条)之后已提交偏移量 10
  [MANUAL 但代码里忘了确认] 开始处理后 1.0 秒:已处理完 3 条,已提交偏移量 None;全部处理完(10 条)之后已提交偏移量 None

BATCH、RECORD、MANUAL、MANUAL_IMMEDIATE 的提交时刻各不相同。

崩溃后的重复

实测 · 本机执行输出 
  [BATCH(默认)] 被杀前开始处理 [0, 1, 2, 3]、处理完 [0, 1, 2],此时已提交偏移量 None;重启后开始处理 [0, 1, 2, 3, 4, 5, 6, 7, 8, 9];被重复处理的 [0, 1, 2, 3]
  [RECORD] 被杀前开始处理 [0, 1, 2, 3]、处理完 [0, 1, 2],此时已提交偏移量 3;重启后开始处理 [3, 4, 5, 6, 7, 8, 9];被重复处理的 [3]
  [MANUAL(处理完确认)] 被杀前开始处理 [0, 1, 2, 3]、处理完 [0, 1, 2],此时已提交偏移量 None;重启后开始处理 [0, 1, 2, 3, 4, 5, 6, 7, 8, 9];被重复处理的 [0, 1, 2, 3]
  [MANUAL_IMMEDIATE(处理完确认)] 被杀前开始处理 [0, 1, 2, 3]、处理完 [0, 1, 2],此时已提交偏移量 3;重启后开始处理 [3, 4, 5, 6, 7, 8, 9];被重复处理的 [3]

提交越勤重复越少,正在处理的一条一定重复。

并发度

实测 · 本机执行输出 
  concurrency=1:30 条消息在分区上的分布 {'p0': 8, 'p1': 12, 'p2': 10};参与的消费者线程 1 个,每个线程处理的分区与条数 [(['p0', 'p1', 'p2'], 30)]
  concurrency=3:30 条消息在分区上的分布 {'p0': 8, 'p1': 12, 'p2': 10};参与的消费者线程 3 个,每个线程处理的分区与条数 [(['p0'], 8), (['p1'], 12), (['p2'], 10)]
  concurrency=5:30 条消息在分区上的分布 {'p0': 8, 'p1': 12, 'p2': 10};参与的消费者线程 3 个,每个线程处理的分区与条数 [(['p0'], 8), (['p1'], 12), (['p2'], 10)]

上限是分区数。

分步演示

Kafka 监听容器
本步的对象

    本文用到的工具与边界

    • 本机 macOS 27 arm64 / OpenJDK 21.0.12(Homebrew) / Spring Boot 4.0.8(Spring Framework 7.0.9、Tomcat 11.0.24) / Spring Cloud 2025.1.3;用 javac 编译、java 直接运行,依赖由 Maven 从 Maven Central 解析;数字每次运行会有小幅抖动。Kafka 4.3.1(Homebrew)单节点 KRaft,自己起的实例;已提交偏移量用 confluent-kafka 的 Consumer.committed 读取;为了让 kill -9 后能很快重启,session.timeout.ms 调成了 6 秒、heartbeat.interval.ms 2 秒
    • 没有验证:默认 session.timeout.ms(45 秒)下 kill -9 后到底要等多久才会重新分配(我是按文档调小的,没有实测默认值);max.poll.interval.ms 超时被踢出消费组的行为;TIME、COUNT、COUNT_TIME 三种 AckMode;批量监听(batch listener);ConsumerAwareRebalanceListener 与重平衡时的提交;事务(read_committed);spring.kafka.consumer.enable-auto-commit=true 时与容器提交的关系;静态成员(group.instance.id)
    京ICP备2026067502号-1京公网安备11010502063487号