📚 课程目录第4章 Kafka消费者API实战

4. Kafka消费者API实战：消费者与消费组概念、订阅Topic、偏移量提交（自动与手动）、再均衡监听器

好，咱们进入第四章。这一章我打算带大家亲手操作Kafka消费者API。说实话，很多做实时ETL的同学，最后数据对不上、丢数据、重复消费，根子都出在消费者这端。你想想看，生产端写得再漂亮，消费端一塌糊涂，整个管道就废了。

这一章我们重点啃四个东西：消费者与消费组的概念、订阅Topic的方式、偏移量提交（自动vs手动），以及再均衡监听器。我会把我在生产环境踩过的坑一并讲出来。

核心观点：消费者API不是简单的“拉数据”，而是一场关于一致性、分区所有权和故障恢复的博弈。理解消费组，你就理解了Kafka的并行度上限。

4.1 消费者与消费组：一张图说清楚

先看一张我手绘的SVG图，把消费组和分区的关系讲透。

这张图我想表达什么？说白了，消费组是一个逻辑订阅者。组内的每个消费者瓜分Topic的所有分区，但一个分区只能分配给一个消费者。我见过不少新手，开了10个消费者，Topic只有3个分区，结果7个消费者一直闲着——浪费资源不说，还以为是Kafka出问题了。

我的经验：消费组内消费者数量不要超过分区数。多出来的消费者不会帮你提升吞吐，反而增加再均衡开销。我个人习惯是消费者数 = 分区数，或者略少。

4.2 订阅Topic：两种方式，一个坑

订阅Topic有两种写法：subscribe() 和 assign()。前者是自动分配分区，后者是手动指定分区。我分别讲。

4.2.1 subscribe() —— 自动分配

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

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic", "another-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", 
            record.offset(), record.key(), record.value());
    }
}

这段代码很常见。注意subscribe()会触发组协调协议，消费者加入组后，由Group Coordinator分配分区。你不需要关心哪个分区给谁，Kafka帮你搞定。

注意：subscribe() 必须配合 group.id 使用。不设置group.id会抛异常。我曾经在测试环境忘了配，结果一直报错，排查了半天才发现。

4.2.2 assign() —— 手动指定

TopicPartition partition0 = new TopicPartition("my-topic", 0);
TopicPartition partition1 = new TopicPartition("my-topic", 1);
consumer.assign(Arrays.asList(partition0, partition1));

手动指定分区时，不需要消费组。你直接告诉消费者：你就消费这两个分区。这种方式适合做有状态的处理，比如你从某个特定分区恢复数据。但缺点也很明显——没有故障转移，消费者挂了就挂了。

特性	subscribe()	assign()
分区分配	自动（Group Coordinator）	手动指定
消费组	必须	不需要
再均衡	支持	不支持
故障转移	自动	无
适用场景	大部分实时ETL	精确恢复、测试

我个人建议：生产环境用subscribe()，除非你有非常特殊的需求。assign()虽然灵活，但你要自己处理所有异常情况，太累了。

4.3 偏移量提交：自动 vs 手动

偏移量（offset）是Kafka消费者的命根子。提交偏移量就是告诉Kafka：“我已经消费到这条消息了，下次从这里开始。” 提交方式分两种：自动和手动。

4.3.1 自动提交

props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");

默认就是自动提交，每5秒提交一次。看起来省事，但有个大坑：重复消费。你想想看，如果poll了一批数据，处理到一半程序挂了，但偏移量还没提交。重启后，Kafka会从上一次提交的位置重新拉取，这批数据就重复处理了。

我的建议：自动提交只适合对数据重复不敏感的场景，比如日志采集、监控指标。做实时ETL，尤其是涉及金额、库存的，千万别用自动提交。

4.3.2 手动提交

手动提交分两种：commitSync() 和 commitAsync()。

commitSync() —— 同步提交

props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        // 处理业务逻辑
        process(record);
    }
    try {
        consumer.commitSync();
    } catch (CommitFailedException e) {
        // 提交失败处理
        log.error("commit failed", e);
    }
}

同步提交会阻塞，直到Kafka确认提交成功。优点是可靠，缺点是慢。如果处理一批数据就要等一次提交，吞吐量会下降。

commitAsync() —— 异步提交

consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        log.error("Commit failed for offsets {}", offsets, exception);
    }
});

异步提交不阻塞，性能好。但有个问题：如果提交失败，回调里你只能记录日志，没法重试。为什么？因为异步提交的顺序无法保证，重试可能导致后提交的偏移量覆盖先提交的，造成数据丢失。

最佳实践：我一般用 commitAsync + 定期commitSync 的组合。正常处理时异步提交，程序关闭前或发生再均衡时，同步提交一次，确保偏移量落盘。

try {
    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            process(record);
        }
        consumer.commitAsync();
    }
} finally {
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

4.4 再均衡监听器：别让数据丢了

再均衡（Rebalance）是消费组内分区所有权重新分配的过程。什么时候触发？消费者加入、离开、崩溃，或者分区数变化。再均衡期间，整个消费组会停止消费，直到分配完成。

再均衡监听器让你在分区被撤销或分配时执行自定义逻辑。Kafka提供了ConsumerRebalanceListener接口，有两个方法：

onPartitionsRevoked：分区被撤销前调用。适合提交偏移量、清理状态。
onPartitionsAssigned：分区被分配后调用。适合加载状态、初始化资源。

consumer.subscribe(Arrays.asList("my-topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 提交偏移量，防止重复消费
        consumer.commitSync();
        // 清理本地状态
        for (TopicPartition partition : partitions) {
            stateStore.remove(partition);
        }
        System.out.println("Partitions revoked: " + partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 加载状态
        for (TopicPartition partition : partitions) {
            stateStore.load(partition);
        }
        System.out.println("Partitions assigned: " + partitions);
    }
});

我曾经踩过的坑：有一次做实时ETL，消费组内一个消费者OOM挂了，触发再均衡。因为没有实现监听器，偏移量没提交，其他消费者接手分区后，从更早的偏移量开始消费，导致大量数据重复处理，下游的数据库被写爆了。从那以后，我每个生产环境的消费者都配了再均衡监听器，在revoked里提交偏移量。

4.5 完整实战：一个可靠的消费者

把上面所有知识点串起来，写一个生产级别的消费者：

public class ReliableConsumer {
    private static final Logger log = LoggerFactory.getLogger(ReliableConsumer.class);
    private volatile boolean running = true;

    public void consume() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "etl-group");
        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("max.poll.records", "500");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList("input-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                log.info("Rebalance: partitions revoked: {}", partitions);
                consumer.commitSync();
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                log.info("Rebalance: partitions assigned: {}", partitions);
            }
        });

        try {
            while (running) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    // 业务处理
                    processRecord(record);
                }
                // 异步提交，性能优先
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        log.error("Async commit failed for offsets: {}", offsets, exception);
                    }
                });
            }
        } catch (Exception e) {
            log.error("Consumer error", e);
        } finally {
            try {
                // 关闭前同步提交，确保偏移量
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }

    private void processRecord(ConsumerRecord<String, String> record) {
        // 模拟ETL处理
        log.debug("Processing record: offset={}, value={}", record.offset(), record.value());
    }

    public void shutdown() {
        running = false;
    }
}

这个消费者做到了：

手动提交偏移量，避免自动提交的重复消费问题
异步提交为主，关闭前同步提交兜底
再均衡监听器在分区撤销时提交偏移量
优雅关闭，确保资源释放

总结一下：消费者API的核心就三件事——订阅方式选对、偏移量提交策略想清楚、再均衡监听器别漏掉。这三件事做好了，你的实时ETL管道就稳了一大半。