10.5.3 消费者一致性与提交策略（offset提交、再均衡影响）

消费者一致性与提交策略是确保消息“处理一次”语义的关键环节。Kafka 以分区为消费并行度单位，消费者通过 offset 标记已处理位置。提交过早会导致消息丢失，提交过晚会导致重复消费，因此需要结合业务幂等性与处理耗时选择合适的提交策略。

原理草图：offset 提交与再均衡影响

Offset 提交方式与适用场景（示例+命令解释）
- 自动提交：简单但不可靠，适合不敏感场景。

# /opt/kafka/config/consumer.properties
enable.auto.commit=true
auto.commit.interval.ms=5000

同步提交：处理完成后提交，阻塞等待结果。
异步提交：性能好但可能丢失提交回执，需回调兜底。

Java 消费者示例（手动提交，含命令解释）

// File: ConsumerManualCommit.java
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;

public class ConsumerManualCommit {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-consumer");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100");

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

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> r : records) {
                // 业务处理：确保幂等（例如写库时用业务主键去重）
                System.out.printf("partition=%d offset=%d value=%s%n",
                        r.partition(), r.offset(), r.value());
            }
            // 处理完成后提交，确保已处理数据不丢失
            consumer.commitSync();
        }
    }
}

运行命令与说明

# 1) 编译
javac -cp kafka-clients-3.6.1.jar ConsumerManualCommit.java

# 2) 运行
java -cp .:kafka-clients-3.6.1.jar ConsumerManualCommit

# 说明：
# - commitSync 在处理完成后提交，避免“处理未完成却提交”
# - max.poll.records 控制单次拉取数量，避免处理过长触发再均衡

再均衡（Rebalance）对一致性的影响与处理
- 触发条件：消费者实例增减、订阅变更、心跳超时、会话超时。
- 影响：未提交 offset 可能被新消费者重复处理；提交过早会跳过未处理消息。
- 处理建议：在 rebalance 前提交已处理 offset。

带 Rebalance 回调的示例（关键点解释）

consumer.subscribe(Collections.singletonList("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 再均衡前提交，减少重复
        consumer.commitSync();
    }
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 可在此恢复本地状态或日志
        System.out.println("Assigned: " + partitions);
    }
});

配置建议（关键参数与解释）

# /opt/kafka/config/consumer.properties
enable.auto.commit=false        # 手动提交
max.poll.records=200             # 单次拉取条数，降低处理超时风险
max.poll.interval.ms=300000      # 最大处理间隔，需大于最长处理时间
session.timeout.ms=10000         # 会话超时，心跳丢失将触发再均衡
request.timeout.ms=30000         # 请求超时，需配合 session.timeout.ms

排错指南（含命令）

# 1) 查看消费组是否频繁再均衡
/opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server 127.0.0.1:9092 \
  --describe --group order-consumer

# 预期效果：
# - LAG 持续增大：处理慢或提交失败
# - MEMBER_ID 频繁变化：可能频繁再均衡

# 2) 查看客户端日志中是否有 Rebalance 或 CommitFailedException
# 常见原因：处理太慢、max.poll.interval.ms 过小、心跳超时

练习
1. 将 enable.auto.commit 改为 true，观察重启消费者后是否出现重复消费或丢失。
2. 把 max.poll.interval.ms 设置为 10000，并在处理逻辑中 Thread.sleep(15000)，观察是否触发再均衡。
3. 使用 kafka-consumer-groups.sh --describe 查看 lag 变化，记录不同提交策略下的延迟对比。

小结
- 手动提交 + 业务幂等是多数生产场景的最佳实践。
- 再均衡会放大“未提交”与“过早提交”的问题，应通过回调与参数配置控制。