Offset
定义与作用
Offset 是 Kafka 中消息在其所属 Partition 中的唯一递增序号。每个 Partition 中的第一条消息 Offset 为 0，后续每条消息的 Offset 依次递增。Offset 之于 Partition，如同数组索引之于数组——它是消费者定位自己"读到哪了"的唯一坐标。
在分布式消费场景中，Offset 解决了消费者状态追踪这个核心问题：多个消费者如何在不相互通信的情况下，各自知道该从哪条消息继续消费。
核心原理
Offset 的三层含义
| Offset 类型 | 含义 | 维护者 |
|---|---|---|
| Log End Offset (LEO) | 该 Partition 中下一条消息将被写入的 Offset | Broker |
| Committed Offset | Consumer 已确认消费完成、提交到 Kafka 的 Offset | Consumer → __consumer_offsets |
| Current Position | Consumer 当前读取位置（内存中），下次 poll() 的起点 | Consumer |
消费者 Offset 提交流程
auto.offset.reset 的策略
当 Consumer 启动且没有已提交的 Offset 时（或提交的 Offset 已过期），此参数决定从哪开始消费：
| 值 | 行为 | 适用场景 |
|---|---|---|
latest（默认） | 从 Partition 尾部开始，只消费新消息 | 关注实时数据 |
earliest | 从 Partition 头部开始，消费所有历史消息 | 新消费者需要全量数据 |
none | 找不到已提交 Offset 时抛出异常 | 严格要求必须有历史状态 |
完整示例
示例一：手动管理 Offset 实现精确控制
场景：一个批处理任务，需要每处理 100 条消息手动提交一次 Offset，防止中途崩溃导致大量重复处理。
操作前环境：Topic offset-demo 中有 500 条消息（Offset 0-499）。
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "batch-processor");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false"); // 关闭自动提交
props.put("auto.offset.reset", "earliest"); // 如果没有已提交 Offset，从头开始
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("offset-demo"));
int count = 0;
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
processRecord(record); // 处理消息
count++;
}
if (count >= 100) {
consumer.commitSync(); // 每 100 条提交一次
System.out.println("Committed at count=" + count);
count = 0;
}
}
操作前后对比：
| 场景 | Offset 提交策略 | 崩溃后行为 |
|---|---|---|
自动提交 (auto.commit=true, 5s) | 每 5 秒提交一次 | 可能丢失已处理但未提交的消息（<5s 内的数据） |
| 手动提交（每 100 条） | 精确控制在 100 条边界 | 最多重复处理 0-99 条 |
示例二：命令行查看和重置 Offset
场景：消费者组 my-group 的 Offset 因 Bug 被错误提交，需要回退到 1 小时前的位置重新消费。
操作前环境：Topic orders 有持续的消息写入，消费者组 my-group 已消费到最新位置。
步骤：
# 1. 查看消费者组当前的 Offset 状态
bin/kafka-consumer-groups.sh \
--bootstrap-server localhost:9092 \
--group my-group \
--describe
# GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
# my-group orders 0 150234 150240 6
# my-group orders 1 148901 148910 9
# 2. 获取 1 小时前的 Offset
# --time 参数接受 Unix 时间戳（毫秒）
TIMESTAMP_1H_AGO=$(date -d '1 hour ago' +%s000)
bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
--topic orders --time $TIMESTAMP_1H_AGO \
--bootstrap-server localhost:9092
# orders:0:142000
# orders:1:140500
# 3. 重置消费者组的 Offset（必须先停止消费者组的所有成员）
bin/kafka-consumer-groups.sh \
--bootstrap-server localhost:9092 \
--group my-group \
--topic orders \
--reset-offsets \
--to-datetime $(date -d '1 hour ago' +%Y-%m-%dT%H:%M:%S.000) \
--execute
# GROUP TOPIC PARTITION NEW-OFFSET
# my-group orders 0 142000
# my-group orders 1 140500
操作前后对比：
| 时间点 | Partition 0 Offset | Partition 1 Offset |
|---|---|---|
| 重置前 | 150,234 | 148,901 |
| 重置后（1小时前） | 142,000 | 140,500 |
| 重启消费者后将重新消费 | 8,234 条 | 8,401 条 |
易错场景
易错 1：自动提交导致"消息丢失"的假象
场景：enable.auto.commit=true，Consumer poll() 拿到一批消息后在业务线程中处理，但处理过程中 auto.commit.interval.ms 到期，自动提交了当前所有已拉取的 Offset（包括尚未处理完的消息）。此时 Consumer 崩溃，重启后这些消息不会再被拉取。
教训：需要"至少处理一次"保证时，使用 enable.auto.commit=false 并在处理完成后手动提交。
易错 2：auto.offset.reset=earliest 与新 Group ID
场景：每次启动都给 Consumer 一个新的 group.id，且 auto.offset.reset=earliest。
后果：每次重启都会从头消费所有历史消息，造成大量重复处理。
教训：Consumer Group ID 应该是稳定的业务标识，不应每次启动都改变。
面试高频考点
Q：Kafka 的 Offset 管理与传统消息队列的 ACK 机制有何不同？为什么 Kafka 选择 Offset 模型？
A：传统 MQ（如 RabbitMQ）由 Broker 跟踪每条消息的消费状态（"已投递/已确认"），每条消息需要独立的元数据。这在小规模消息场景可行，但在百万 QPS 场景下成为瓶颈。
Kafka 的设计更简单：每个 Partition 的消费进度只需要一个数字（Offset）。因为一个 Partition 在同一个消费组内只被一个 Consumer 消费，该 Consumer 的位置就是该 Partition 在该消费组内的 Offset。
优势：
- 元数据极小（1 个数字 vs 每条消息一个状态）
- Consumer 可以回退（rewind）重新消费历史数据
- Offset 的读写是 O(1) 操作