Consumer Offset 提交

定义与作用

Offset 提交是 Consumer 将"我已经消费到哪个位置了"持久化到 Kafka 内部 Topic __consumer_offsets 的过程。它是 Consumer 实现故障恢复的关键机制——Consumer 崩溃后重新启动，从已提交的 Offset 继续消费，而非从头开始或丢掉进度。

核心原理

Offset 提交的完整流程

自动提交 vs 手动提交

提交方式	代码	特点
自动提交	无需编码	简单但可能丢数据或重复
同步提交	`consumer.commitSync()`	可靠但阻塞消费
异步提交	`consumer.commitAsync(callback)`	高性能但失败不重试
精确分区提交	`consumer.commitSync(offsets)`	可只提交特定分区

完整示例

示例一：处理完每条消息后同步提交（最高可靠性）

场景：金融交易处理，必须保证每条消息处理完才提交 Offset。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "payment-processor");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

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

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            processPayment(record);  // 处理支付
            // 处理成功 → 立即同步提交
            consumer.commitSync();
        }
    }
}

操作前后对比：

场景	自动提交（间隔 5s）	手动逐条提交
处理 100 条后崩溃	丢失 ~50 条（未提交的）	最多丢失 1 条（当前正在处理的）
吞吐量	~10K/s	~5K/s（每次提交 1 次网络往返）

示例二：批量处理后提交（平衡可靠性与性能）

场景：日志收集，可容忍少量重复，但需要高吞吐。

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

    int count = 0;
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            processLog(record);
            count++;

            // 每 1000 条提交一次
            if (count % 1000 == 0) {
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        System.err.println("异步提交失败: " + exception.getMessage());
                    }
                });
            }
        }
    }
}

操作前后对比：

提交频率	吞吐量	崩溃后最大重复量
每条 1 次	~5K/s	0-1 条
每 100 条 1 次	~50K/s	0-100 条
每 1000 条 1 次	~80K/s	0-1000 条

示例三：关闭时安全提交（最佳实践）

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

    try {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                processOrder(record);
            }
            // 正常运行期间异步提交
            consumer.commitAsync();
        }
    } catch (WakeupException e) {
        // 收到关闭信号，忽略
    } finally {
        // ⚠️ 关键：关闭前同步提交一次，保证当前进度被持久化
        try {
            consumer.commitSync();
        } finally {
            consumer.close();
        }
    }
}

易错场景

易错 1：自动提交时处理失败但 Offset 已提交

场景：enable.auto.commit=true，消息被 poll 后在业务线程中处理抛出异常，但自动提交触发了。

后果：消息已"确认"但实际未处理成功，下次消费从下一条开始——该消息丢失。

正确做法：任何需要"至少处理一次"的场景，关闭自动提交，在处理成功后手动提交。

易错 2：提交的 Offset 比实际处理的位置大

// 错误：poll 后立即提交，但消息还没处理
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
consumer.commitSync();  // ← 提交了这批消息的末尾 Offset
// 如果接下来处理时崩溃 → 这批消息全部丢失
for (ConsumerRecord<String, String> record : records) {
    processRecord(record);
}

面试高频考点

Q：commitSync 和 commitAsync 如何选择？

A：

场景	推荐
要求每条消息都被确认	`commitSync`（逐条或批量）
高吞吐，可容忍少量重复	`commitAsync`（批量）
两者兼顾	正常运行 `commitAsync`，异常/关闭时 `commitSync`

实际最佳实践：正常运行期间使用 commitAsync（批量 + 回调记录失败），在 Consumer 关闭的 finally 块中使用 commitSync 作为"最后一道防线"。