进度不在消息里
前五章讲的是消息本身：它怎么被写进分区、怎么复制、元数据怎么传播。从这一章开始，视角换到读消息的一方。
Kafka 的消费者不删除消息。消息被读走之后仍然留在日志里，直到保留策略把它清掉。那么「这个消费组读到哪了」这件事，不可能记录在消息上，只能单独存一份。这份单独存下来的东西就是位移（offset）。
位移有两个容易混淆的形态：
- 消费位置（position）：消费者当前读到哪，存在于客户端内存里，
poll()返回一批记录后就前进了。 - 已提交位移（committed offset）：消费者主动告诉服务端「我已经处理到这里」，由服务端持久化。消费者重启、换机器、组内再平衡之后，从这份记录恢复。
这一节要回答三个问题：已提交位移存在哪、以什么结构存、它和消费位置差在哪。全部用本机跑出来的原始输出说话。
__consumer_offsets 是什么
它是一个普通的内部 topic，名字叫 __consumer_offsets。说它普通，是因为你能用 kafka-topics.sh 查看它、用 kafka-dump-log.sh 读它；说它特殊，是因为它的创建不由用户控制，而是第一个消费组提交位移时由协调器自动创建。
本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：
docker exec mq-b1 /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server 127.0.0.1:9092 --describe --topic __consumer_offsets
Topic: __consumer_offsets TopicId: rgen7YWqTmOgfgvaARi9gQ PartitionCount: 50 ReplicationFactor: 1 Configs: compression.type=producer,min.insync.replicas=1,cleanup.policy=compact,segment.bytes=104857600
Topic: __consumer_offsets Partition: 0 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
Topic: __consumer_offsets Partition: 1 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
Topic: __consumer_offsets Partition: 2 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
输出只截了前三行，实际有 50 个分区。三个配置值得逐条读：
PartitionCount: 50：由offsets.topic.num.partitions决定，默认 50。建好之后不能靠改配置扩容，要扩只能手工做分区重分配。cleanup.policy=compact：这是键值表的存储方式。同一个 key 的新值会覆盖旧值，日志清理后只保留每个 key 的最新一条。位移天生适合这种语义：关心「现在读到哪」，不关心「历史上提交过几次」。ReplicationFactor: 1：本机单节点，启动时用KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1覆盖了默认值 3。生产环境必须是 3，否则协调器所在 broker 一挂，所有组的位移都不可用。
注意 min.insync.replicas=1 出现在 Configs 里。这是 4.x 给内部 topic 写下的显式配置，和我们在 4.2 节讲的写入门槛是同一套机制。单副本时它只能取 1，没有讨论空间。
一个组的位移落在同一个分区
50 个分区不是按 topic-partition 分散写入的。同一个 group 的所有位移记录都落在同一个 __consumer_offsets 分区，分区号由组名决定。
本机做法是扫描 50 个段文件，找出含目标组名的分区。命令有点长，但很直接：
docker exec mq-b1 sh -c 'for d in /tmp/kafka-logs/__consumer_offsets-*; do
f=$(ls $d/*.log | head -1)
n=$(/opt/kafka/bin/kafka-dump-log.sh --offsets-decoder --print-data-log --files $f 2>/dev/null \
| grep -c "b1.offsets.group")
if [ "$n" != "0" ]; then echo "PARTITION $(basename $d) hits=$n"; fi
done'
本机实测输出：
PARTITION __consumer_offsets-22 hits=6
组 b1.offsets.group 的 6 条记录全部在 22 号分区。再拿另一个组 b1.proto.consumer.group 做同样的扫描，结果是 18 号分区。把两个组名代进 Java 的字符串哈希：
b1.offsets.group hashCode=1463411122 abs%50=22
b1.proto.consumer.group hashCode=-1396975668 abs%50=18
abs(groupId.hashCode()) % 50 与实测分区完全一致。这不是巧合，也不是本节自己发明的公式——两个样本、两个不同分区，足以说明定位规则。工程上的含义很实在：一个消费组的位移读写都压在一个分区、一个协调器上。组特别多时，50 个分区就是位移流量的并行上限；组特别大时，单组的提交频率会先撞到这个上限。
键和值长什么样
kafka-dump-log.sh 有一个 --offsets-decoder 选项，专门按 __consumer_offsets 的格式解码。下面是从 22 号分区解出来的原文（为了可读性，只保留与本节有关的四条）：
docker exec mq-b1 sh -c 'f=$(ls /tmp/kafka-logs/__consumer_offsets-22/*.log | head -1);
/opt/kafka/bin/kafka-dump-log.sh --offsets-decoder --print-data-log --files $f'
| offset: 0 ... key: {"type":"2","data":{"group":"b1.offsets.group"}} payload: {"version":"3","data":{"protocolType":"","generation":0,"protocol":null,"leader":null,"currentStateTimestamp":1793247616118,"members":[]}}
| offset: 1 ... key: {"type":"2","data":{"group":"b1.offsets.group"}} payload: {"version":"3","data":{"protocolType":"consumer","generation":1,"protocol":"range","leader":"console-consumer-32a94a99-9134-4cd1-a906-b572d5626d23","currentStateTimestamp":1793247619126,"members":[...]}}
| offset: 2 ... key: {"type":"1","data":{"group":"b1.offsets.group","topic":"b1.offsets.demo","partition":1}} payload: {"version":"4","data":{"offset":12,"leaderEpoch":0,"metadata":"","commitTimestamp":1793247619231,"topicId":"Rfl4WzuKSVOyZ2jXqNldDQ"}}
| offset: 3 ... key: {"type":"1","data":{"group":"b1.offsets.group","topic":"b1.offsets.demo","partition":0}} payload: {"version":"4","data":{"offset":0,"leaderEpoch":-1,"metadata":"","commitTimestamp":1793247619231,"topicId":"Rfl4WzuKSVOyZ2jXqNldDQ"}}
这张表只有两种 key 类型，含义完全不同：
type | key 里的字段 | 值里存什么 | 谁写入 |
|---|---|---|---|
1 | group + topic + partition | offset、leaderEpoch、metadata、commitTimestamp、topicId | 消费者提交位移 |
2 | group | 组的世代、协议、leader、成员列表 | 协调器维护组状态 |
type=1 才是位移。key 三元组 (group, topic, partition) 唯一确定一个位移，value 里的 offset 就是已提交位移。本机这次提交的三个分区分别是 12、0、0，和后面 --describe 的输出对得上。
两个细节容易被忽略：
leaderEpoch会写进去。4.x 的提交格式带leaderEpoch，用于在 leader 切换后判断位移是否还安全。partition=0那条是-1，因为它从没被消费过，提交的是起始位置，没有 leader 世代可记。topicId也在 value 里。Kafka 4.x 用 topic UUID 而不是 topic 名字来防「删了重建同名 topic 后旧位移被误用」。key 里仍然保留topic字符串是为了兼容旧客户端。
提交位移与消费位置不是一回事
光看段文件还不够直观：位置前进时，服务端的那条记录并不会跟着动。写一个最小程序把两个值同时打出来。
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.List;
import java.util.Properties;
public class PosVsCommit {
public static void main(String[] args) {
Properties p = new Properties();
p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:19211");
p.put(ConsumerConfig.GROUP_ID_CONFIG, "b1.pos.group");
p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
try (KafkaConsumer<String, String> c = new KafkaConsumer<>(p)) {
c.subscribe(List.of("b1.offsets.demo"));
c.poll(Duration.ofSeconds(5));
dump("poll#1", c);
c.poll(Duration.ofSeconds(3));
dump("poll#2", c);
c.commitSync();
dump("commit", c);
c.poll(Duration.ofSeconds(3));
dump("poll#3", c);
}
}
private static void dump(String tag, KafkaConsumer<String, String> c) {
for (TopicPartition tp : c.assignment()) {
OffsetAndMetadata cm = c.committed(java.util.Set.of(tp)).get(tp);
System.out.printf("%-8s %s position=%d committed=%s%n",
tag, tp, c.position(tp), cm == null ? "null" : String.valueOf(cm.offset()));
}
}
}
本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29），去掉配置日志后只留业务行：
[poll#1] records=5
poll#1 b1.offsets.demo-1 position=0 committed=null
poll#1 b1.offsets.demo-0 position=5 committed=null
poll#1 b1.offsets.demo-2 position=0 committed=null
[poll#2] records=5
poll#2 b1.offsets.demo-1 position=4 committed=null
poll#2 b1.offsets.demo-0 position=6 committed=null
poll#2 b1.offsets.demo-2 position=0 committed=null
[commitSync] done
commit b1.offsets.demo-1 position=4 committed=4
commit b1.offsets.demo-0 position=6 committed=6
commit b1.offsets.demo-2 position=0 committed=0
[poll#3] records=5
poll#3 b1.offsets.demo-1 position=9 committed=4
poll#3 b1.offsets.demo-0 position=6 committed=6
poll#3 b1.offsets.demo-2 position=0 committed=0
这张输出里有三件事：
poll#1和poll#2之间，position 变了，committed 还是 null。程序没有提交，服务端就没有任何记录，重启后这个组仍然从auto.offset.reset的位置开始。commitSync之后两者相等。提交动作把当时内存里的 position 写进__consumer_offsets。poll#3之后 position 到了 9，committed 还停在 4。这是常态：只要没提交，读得再远也不算数。
「消费位置」描述的是这一次运行的进度；「已提交位移」描述的是可恢复的进度。再平衡、重启、换机器，恢复到的是后者。前面那个 poll#3 如果此时进程被杀，重新分配后消费者会从 4 继续，而不是 9——那 5 条已经处理过的记录会被再读一遍。这是至少一次语义的来源，卷二会展开；本节只需要记住：提交是唯一的持久化动作。
–describe 的六列怎么读
日常看得最多的是这张表。本机对 b1.pos.group 的实测输出：
docker exec mq-b1 /opt/kafka/bin/kafka-consumer-groups.sh \
--bootstrap-server 127.0.0.1:9092 --describe --group b1.pos.group
Consumer group 'b1.pos.group' has no active members.
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
b1.pos.group b1.offsets.demo 0 6 6 0 - - -
b1.pos.group b1.offsets.demo 1 4 12 8 - - -
b1.pos.group b1.offsets.demo 2 0 12 12 - - -
逐列解释：
| 列 | 含义 | 读的时候注意 |
|---|---|---|
CURRENT-OFFSET | 已提交位移，读的是 __consumer_offsets 里的值 | 不是消费位置；不提交就永远不动 |
LOG-END-OFFSET | 该分区当前日志末端位移，即下一条待写消息的位置 | 只说明写入量，不说明消费量 |
LAG | LOG-END-OFFSET - CURRENT-OFFSET | 用的是已提交位移，不含「已拉取但未提交」的部分 |
CONSUMER-ID | 当前持有该分区的成员 | 没有活跃成员时是 - |
HOST | 成员连接协调器时用的地址 | 默认 bridge 网络里常是 172.17.0.1 |
CLIENT-ID | 客户端自报的标识 | 排查「是谁在消费」时比 CONSUMER-ID 好认 |
这里 p1 的 lag 是 8、p2 是 12，合计 20；而日志末端一共 30 条，已提交 10 条，账对得上。注意 CURRENT-OFFSET 用的是已提交位移，不是消费者内存里的 position。如果消费者正在跑、每秒拉取但不提交，lag 会看起来「不动」，这不代表它没在消费。
还有一个容易误读的地方：--describe 默认只列有位移记录的分区。如果一个组从没提交过，输出可能只有组名和「no active members」，没有任何分区行。想看成员和分配，用 --members --verbose；想看组状态，用 --state。
offsets.retention.minutes 的边界
已提交位移不是永久的。控制它寿命的是 broker 配置 offsets.retention.minutes，本机实测默认值：
docker exec mq-b1 /opt/kafka/bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 \
--describe --entity-type brokers --entity-name 1 --all | grep offsets.retention
offsets.retention.check.interval.ms=600000 sensitive=false synonyms={DEFAULT_CONFIG:offsets.retention.check.interval.ms=600000}
offsets.retention.minutes=10080 sensitive=false synonyms={DEFAULT_CONFIG:offsets.retention.minutes=10080}
10080 分钟是 7 天。语义是：当组里没有任何活跃成员，且最后一次提交已经超过这个时长，位移记录会被清理。清理不是实时的，后台任务按 offsets.retention.check.interval.ms（10 分钟）扫描一次，所以实际删除时间会在 7 天之后的某个扫描点。
这里有个值得动手验证的细节。我尝试把它动态改小：
docker exec mq-b1 /opt/kafka/bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 \
--alter --entity-type brokers --entity-name 1 --add-config offsets.retention.minutes=1
本机实测返回：
InvalidRequestException: Cannot update these configs dynamically: [offsets.retention.minutes]
也就是说，这个参数和 4.2 节的 replica.lag.time.max.ms 一样，不能在线改，必须写进 broker 配置后重启。这一点在很多「调参清单」里被写成可直接 --alter，实测不成立。想验证清理行为，只能接受「改配置 + 重启 + 等 10 分钟扫描」的成本；本节没有在本机等待这个窗口，所以不给「多久后消失」的数字。
还有两个相关参数容易混：
offsets.topic.segment.bytes（默认 100 MB）：内部 topic 的段大小。位移写入量很小，一般很久才滚一次段。offsets.topic.compression.codec（默认 0，不压缩）：位移记录的压缩方式。
它们影响的是存储与 IO，不影响「位移保留多久」。
一次完整的复现路径
把本节实验按顺序串起来，命令如下。前提是已经起好单节点 broker，宿主机端口 19211，容器内 PLAINTEXT 端口 9092。注意容器内命令用 9092 而不是 19092：19092 是 EXTERNAL 监听，advertised 成宿主机地址 19211，在容器里连它会拿到一个自己不可达的地址。
# 1. 建一个 3 分区 topic
docker exec mq-b1 /opt/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 \
--create --topic b1.offsets.demo --partitions 3 --replication-factor 1
# 2. 写 30 条带 key 的消息
python3 - <<'EOF' | docker exec -i mq-b1 /opt/kafka/bin/kafka-console-producer.sh \
--bootstrap-server 127.0.0.1:9092 --topic b1.offsets.demo \
--property parse.key=true --property key.separator=:
for i in range(1, 31):
print(f"book-{i % 5}:loan-{i}")
EOF
# 3. 用组消费 12 条后退出（退出时会提交）
docker exec mq-b1 /opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server 127.0.0.1:9092 --topic b1.offsets.demo \
--group b1.offsets.group --from-beginning --max-messages 12
# 4. 看提交结果
docker exec mq-b1 /opt/kafka/bin/kafka-consumer-groups.sh \
--bootstrap-server 127.0.0.1:9092 --describe --group b1.offsets.group
第 3 步的本机输出里有 12 条 loan-N，并且末尾是：
Processed a total of 12 messages
第 4 步在本机得到的表是：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
b1.offsets.group b1.offsets.demo 0 0 6 6 - - -
b1.offsets.group b1.offsets.demo 1 12 12 0 - - -
b1.offsets.group b1.offsets.demo 2 0 12 12 - - -
消费的 12 条全部落在 p1，所以 p1 的 CURRENT-OFFSET 是 12、lag 是 0；p0 和 p2 从没被这个组读过，提交的是起始位置 0，lag 分别是 6 和 12。这条数据同时说明：位移是按分区记的，不是按组记一个总数。
本节判据
读完这一节，你应该能回答：
- 已提交位移存在哪个 topic、什么结构？ 答：
__consumer_offsets，key 是type=1的(group, topic, partition)三元组，value 里是 offset、leaderEpoch、metadata、commitTimestamp、topicId。 - 为什么一个组的记录都在同一个分区？ 答：分区号由组名哈希决定，本机两个组分别落在 22 和 18，和
abs(hashCode) % 50一致。 - position 和 committed offset 差在哪？ 答：position 在客户端内存里随
poll()前进；committed 只有提交才写服务端，是重启和再平衡后恢复的依据。 --describe的 LAG 用哪个值算？ 答：用已提交位移，LOG-END-OFFSET - CURRENT-OFFSET，不含已拉取未提交的部分。offsets.retention.minutes能不能在线改？ 答：本机实测不能，返回Cannot update these configs dynamically。
下一节把视角从「位移存在哪」推到「一群消费者怎么分这些分区」。经典再平衡协议的 JoinGroup、SyncGroup、Heartbeat 三个阶段，以及再平衡期间应用线程到底卡在哪一行代码，都会用两个消费者的实测时间线来说明。
阅读导航：上一节 5.3 客户端的元数据刷新 ｜ 下一节 6.2 经典再平衡协议
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。