未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

warning 适用边界与维护说明

本文适用于学习 Kafka 偏移量提交策略、观察 LAG，以及在测试消费者组中练习停组重置和回放；示例代码是局部教学片段，不是完整生产交付方案。正文已修正假设事故、HW 可读边界、偏移量保留、异步提交失败补交、巡检列号及事务回放边界等历史问题，不再将其列为未解决缺陷。当前仍需注意：内部 offset 记录格式未限定适用版本，不能直接作为跨版本解析规范；末段关于生产环境最常用策略和 Kafka 事务性能开销的概括未提供统计、基准或适用条件，不宜作为选型定论。提交与回放应以已成功处理的位置、失败恢复、业务幂等和明确的事务边界为前提。本次仅作全文结构审阅，未进行外部事实核验或代码、命令实测。

偏移量管理 - 消息的精确定位系统

info 前置知识

假设数据工程师小张刚接手一个消费者程序：消费完消息后没有正确提交偏移量，导致相同消息被重复处理。在这个示例中，假设用户收到了 10 份重复的促销短信，引发大量客服电话，并造成数万元广告费损失。这些人物、数量和损失是用于说明风险的假设，不是真实事故记录：偏移量虽小，责任重大。

偏移量的类型和含义

不同类型的偏移量

偏移量类型详解：

1. 日志末端偏移量 (Log End Offset - LEO)
   - 表示分区中下一条消息将被写入的位置
   - 由Kafka自动管理，递增分配

2. 消费者偏移量 (Consumer Offset)
   - 表示消费者下一次要消费的消息位置
   - 由消费者管理和提交

3. 高水位偏移量 (High Water Mark - HW)
   - 表示已复制消息可读范围的排他上界：消费者不能读取 offset >= HW 的消息；read_committed 模式下还受最后稳定偏移量（LSO）限制
   - 等于所有ISR副本中最小的LEO

4. 消费滞后 (Consumer Lag)
   - 计算公式: Lag = LEO - Consumer Offset
   - 表示消费者落后生产者的消息数量

偏移量在分区中的位置

分区数据结构示意：

Offset:     0    1    2    3    4    5    6    7    8    9
Messages: [msg0][msg1][msg2][msg3][msg4][msg5][msg6][msg7][msg8][msg9]
                                                                ↑
                                                               LEO = 10

Consumer Group A:             ↑
                  Consumer Offset = 4 (下次消费offset=4的消息)

Consumer Group B:                     ↑
                          Consumer Offset = 6

High Water Mark:                                              ↑
                                                         HW = 9

消费滞后计算：
- Group A Lag = LEO - Consumer Offset = 10 - 4 = 6条消息
- Group B Lag = LEO - Consumer Offset = 10 - 6 = 4条消息

__consumer_offsets 内部 Topic

Kafka 使用内置 Topic __consumer_offsets 存储所有消费者组的偏移量，取代了早期依赖 ZooKeeper 的方案（Kafka 0.9 之前）。

Topic 特性：

属性 值 说明
分区数 50（默认） 由 offsets.topic.num.partitions 控制
副本因子 3（默认） 由 offsets.topic.replication.factor 控制
清理策略 compact 按 Key 压缩，只保留每个消费者组+分区的最新 offset
写入者 Group Coordinator Consumer 通过 commitSync/commitAsync 发起提交请求，由协调器写入内部 Topic

分区路由逻辑：

某个消费者组的偏移量存储在哪个分区，由以下公式决定：

目标分区 = Kafka 对 groupId 哈希取非负值后 % __consumer_offsets 的实际分区数（默认配置为 50）

同一消费者组的所有分区偏移量都集中在 __consumer_offsets 的同一个分区，由该分区的 Leader Broker 管理。这个 Leader Broker 被称为该消费者组的 Group Coordinator。

记录格式：

  • Key：[GroupId, TopicName, PartitionId] 三元组
  • Value：[Offset, Metadata, CommitTimestamp, ExpirationTimestamp]

note compact 策略的意义

__consumer_offsets 使用日志压缩（log compaction）清理同一 Key 的旧记录，但这不表示偏移量永久保留。消费者组偏移量仍可能按 offsets.retention.minutes 及相应过期条件被协调器删除；删除标记也会由压缩机制清理。Topic 的 retention.ms 与消费者组偏移量的保留机制不能混为一谈。

偏移量提交策略

自动提交

配置简单，适合对消息丢失有一定容忍度的场景：

Properties props = new Properties();
// 开启自动提交
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// 自动提交检查间隔设为 5 秒；提交值为各分区下一条待消费的位置，不代表消息已经处理完成
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");

自动提交的风险：

时间轴：
T=0s   poll() 返回 offset 0~99 的消息，开始处理
T=3s   处理到 offset 49，程序崩溃
T=5s   (如果没崩溃) 自动提交 offset=100
T=5s   (实际崩溃了，假设此前已提交offset=0且无后续成功提交) 下次启动从 offset=0 重新消费，offset 0~49 重复处理，offset 50~99 首次处理

自动提交由消费者调用 poll() 时检查并触发，并非独立后台线程按时提交。频繁调用 poll() 有助于及时触发检查，但提交完成时间仍受协调器状态、网络和失败重试影响，auto.commit.interval.ms 不保证提交在该间隔内成功完成。

手动同步提交

Properties props = new Properties();
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Collections.singletonList("hello-kafka"));

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

        for (ConsumerRecord<String, String> record : records) {
            // 处理消息（写数据库、调接口等）
            processMessage(record);
        }

        // 处理完整批消息后，同步提交各分区下一条待消费的位置（最后处理的 offset + 1）
        // commitSync() 会阻塞，直到 Broker 确认提交成功或抛异常
        if (!records.isEmpty()) {
            try {
                consumer.commitSync();
            } catch (CommitFailedException e) {
                // 通常因为 Consumer 被踢出 Group（session.timeout 超时）
                log.error("Offset 提交失败，消息可能重复处理", e);
            }
        }
    }
}

手动异步提交

// 处理完一批消息后异步提交，不阻塞 poll 循环
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        // 异步提交失败不会自动重试，需要记录日志或告警
        // 禁用自动提交时，close() 不会自动补交失败的 offset；应用可在后续批次或正常退出前显式执行 commitSync()，并处理失败
        log.warn("异步提交失败，offsets={}", offsets, exception);
    } else {
        log.debug("异步提交成功，offsets={}", offsets);
    }
});

精细化控制：按分区提交

当需要逐条确认或按分区独立控制提交进度时，可以指定提交的精确 offset：

// 每处理完 100 条消息提交一次，而不是等整批结束
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
int count = 0;

for (ConsumerRecord<String, String> record : records) {
    processMessage(record);
    count++;

    // 记录当前处理到的位置（提交值 = 当前 offset + 1，表示下次从这里开始）
    offsetsToCommit.put(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1)
    );

    // 每 100 条提交一次
    if (count % 100 == 0) {
        consumer.commitSync(offsetsToCommit);
        offsetsToCommit.clear();
    }
}

// 提交剩余部分
if (!offsetsToCommit.isEmpty()) {
    consumer.commitSync(offsetsToCommit);
}

自动提交 vs 手动提交对比

维度 自动提交 手动同步提交 手动异步提交
代码复杂度 低 中 中
吞吐量影响 定期提交仍有请求开销，通常不逐批同步等待确认 每批等待 Broker 确认，有延迟 不同步等待提交确认，但仍有请求及回调开销，实际吞吐取决于处理与配置
消费语义 取决于处理与 poll 的顺序，不能无条件保证 at-least-once 处理成功后提交可支持 at-least-once 处理成功后提交可支持 at-least-once，需处理提交失败
丢消息风险 提交位置超过已完成处理的位置时，崩溃后可能跳过未处理消息 仅提交已成功处理的位置可避免因提前提交而跳过消息，但不是端到端不丢失保证 同样必须只提交已成功处理的位置；提交失败通常增加重复处理风险，后续补交必须由应用安排
重复消费风险 有（崩溃时未提交已处理的消息） 有（崩溃后重启会重复最后一批） 有
适用场景 日志、监控等可容忍重复的场景 数据库写入、金融交易 高吞吐、幂等处理场景

使用命令行管理偏移量

kafka-consumer-groups.sh 是管理消费者组偏移量最常用的命令行工具。

查看消费者组状态

# 列出集群中所有消费者组
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --list

# 查看某个消费者组的详细偏移量信息（核心命令）
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group hello-consumer-group \
    --describe

输出示例：

GROUP                TOPIC        PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID       HOST         CLIENT-ID
hello-consumer-group hello-kafka  0          45              50              5    consumer-1-xxx    /10.0.0.1    consumer-1
hello-consumer-group hello-kafka  1          38              40              2    consumer-1-xxx    /10.0.0.1    consumer-1
hello-consumer-group hello-kafka  2          52              52              0    consumer-2-xxx    /10.0.0.2    consumer-2
  • CURRENT-OFFSET：消费者已提交的 offset（下次从这里消费）
  • LOG-END-OFFSET：分区的最新 offset（LEO）
  • LAG：积压消息数 = LEO - CURRENT-OFFSET，监控的核心指标
# 查看所有消费者组的汇总 LAG（适合巡检）
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --describe --all-groups 2>/dev/null | \
    awk 'NF>=6 && $1!="GROUP" {print $1, $2, $3, $6}' | \
    column -t

Offset 重置

warning 重置 offset 会影响消费进度，必须在消费者停止后执行

重置操作不会等消费者自动停止，如果消费者还在运行，重置结果会被消费者的正常提交覆盖。必须先停止消费者，再执行重置，最后重启消费者。

重置到最早位置（重新消费所有消息）：

kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group hello-consumer-group \
    --topic hello-kafka \
    --reset-offsets \
    --to-earliest \
    --execute

重置到最新位置（跳过所有积压消息）：

kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group hello-consumer-group \
    --topic hello-kafka \
    --reset-offsets \
    --to-latest \
    --execute

重置到指定 offset（精确定位）：

# 将 hello-kafka 的所有分区 offset 重置到 100
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group hello-consumer-group \
    --topic hello-kafka \
    --reset-offsets \
    --to-offset 100 \
    --execute

# 只重置特定分区（分区 0 重置到 offset=50）
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group hello-consumer-group \
    --topic hello-kafka:0 \
    --reset-offsets \
    --to-offset 50 \
    --execute

重置到指定时间点（按时间回溯，非常实用）：

# 将 offset 重置到 2024-01-15 08:00:00 时刻之后的第一条消息
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group hello-consumer-group \
    --topic hello-kafka \
    --reset-offsets \
    --to-datetime 2024-01-15T08:00:00.000 \
    --execute

先预览，再执行（--dry-run 不实际修改）：

# 加 --dry-run 只显示重置结果，不真正执行
kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --group hello-consumer-group \
    --topic hello-kafka \
    --reset-offsets \
    --to-earliest \
    --dry-run

代码中的 Seek 操作

除了命令行，也可以在消费者代码中直接定位 offset，比命令行更灵活：

consumer.subscribe(Collections.singletonList("hello-kafka"),
    new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // Rebalance 前提交当前 offset
            consumer.commitSync();
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // Rebalance 完成后，为每个分配到的分区设置起始 offset
            for (TopicPartition partition : partitions) {
                // 每次分区分配都从开头重新消费，后续 Rebalance 也会再次回放；仅适用于明确需要重放的场景。seek 修改本地消费位置，不等同于提交组 offset
                consumer.seekToBeginning(Collections.singleton(partition));
                // 或：从分区末尾开始（跳过所有积压）
                // consumer.seekToEnd(Collections.singleton(partition));
                // 或：定位到特定 offset
                // consumer.seek(partition, 100L);
            }
        }
    }
);

消费语义总结

偏移量的管理方式是消费语义的重要组成部分；实际交付保证还取决于处理是否成功、故障恢复、事务边界和业务幂等。三种语义的典型实现方式如下：

语义 含义 实现方式 适用场景
At-most-once（最多一次） 消息可能丢失，不会重复 先提交 offset，再处理消息 可以接受丢失的日志、指标采集
At-least-once（至少一次） 消息不会丢失，可能重复 先处理消息，再提交 offset 大多数业务场景，业务层需幂等
Exactly-once（恰好一次） 在规定的事务边界内，已提交的处理结果恰好生效一次 将输出记录与消费 offset 纳入同一 Kafka 事务，下游使用 read_committed；幂等 Producer 本身不足以保证消费处理恰好一次 适用于 Kafka 内的事务性读—处理—写流程，不直接覆盖外部系统副作用

生产环境最常用的是 At-least-once + 业务幂等，而不是 Exactly-once。原因：Exactly-once 需要开启 Kafka 事务，性能开销大，且要求消费者和生产者都在同一套 Kafka 事务范围内，跨系统（如写 MySQL）无法覆盖。

掌握检查

  • [ ] 理解 LEO、HW、Consumer Offset、Consumer Lag 四个偏移量的含义和计算方法
  • [ ] 了解 __consumer_offsets Topic 的存储机制（compact 策略、分区路由公式）
  • [ ] 掌握自动提交和手动提交（同步/异步）的区别，能根据场景选择合适策略
  • [ ] 能使用 kafka-consumer-groups.sh --describe 查看消费滞后，并用 --reset-offsets 重置 offset
  • [ ] 理解三种消费语义（At-most-once / At-least-once / Exactly-once）的含义和实现代价

实践练习

基础题：启动一个消费者组消费 hello-kafka，让它积累一些 LAG，然后用命令行查看当前每个分区的 CURRENT-OFFSET 和 LAG。接着停止消费者，将 offset 重置到 earliest，重启消费者验证是否从头开始消费。

进阶题：修改消费者代码，实现"每 50 条消息提交一次 offset，而不是每批结束后提交"的逻辑。并在提交之前故意抛出一个异常，观察重启后从哪个 offset 开始重新消费，验证 at-least-once 语义。

前置知识：副本机制与一致性 下一步学习：性能调优实战 - 学习如何通过配置调整提升 Kafka 吞吐量和降低延迟 相关文档：消费者深入剖析 | 主题与分区设计 | 监控与故障排查