4. 消息消费原理

Kafka 消费端的设计直接决定了消息堆积的感知、定位与恢复效率。本章深入剖析 Consumer 的核心机制：拉取模型、poll 循环、位移提交（自动/手动）以及再均衡（Rebalance）。理解这些原理是诊断堆积问题的前提。

4.1 拉取模型：Consumer 主动 Pull

Kafka 采用 Consumer 主动拉取（Pull） 模式，而非 Broker 推送（Push）。Consumer 根据自身处理能力，通过 poll() 方法从 Broker 拉取消息。

4.2 poll 循环：消费的核心引擎

Consumer 通过一个 无限循环 调用 poll(Duration timeout) 来获取消息。每次 poll 返回一个 ConsumerRecords 集合，包含多个分区的消息批次。

典型 poll 循环结构如下：

while (true) {
    // 1. 拉取消息，超时时间 1 秒
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

    // 2. 处理消息批次
    for (ConsumerRecord<String, String> record : records) {
        process(record);  // 业务处理逻辑
    }

    // 3. 同步提交位移（手动模式）
    consumer.commitSync();
}

poll 循环的关键参数

4.3 位移提交：自动 vs 手动

位移（Offset）是 Consumer 在分区内的消费位置标记。提交位移决定了下次 poll 的起始位置，直接影响消费语义（至少一次、最多一次、精确一次）。

4.3.1 自动提交（enable.auto.commit = true）

• 提交时机： poll 循环内部，每次 poll 返回后自动提交上一批消息的位移。
• 提交间隔： 由 auto.commit.interval.ms 控制（默认 5000ms）。
• 风险： 如果 Consumer 在处理消息后、自动提交前崩溃，重启后会从上次提交的位移开始消费，导致重复消费。

4.3.2 手动提交（enable.auto.commit = false）

开发者显式控制提交时机，分为同步和异步两种方式：

// 手动异步提交 + 关闭前同步兜底
try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            process(record);
        }
        // 异步提交，不阻塞
        consumer.commitAsync((offsets, exception) -> {
            if (exception != null) {
                log.error("异步提交失败: {}", offsets, exception);
            }
        });
    }
} finally {
    // 关闭前同步提交，确保位移不丢失
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

4.4 再均衡（Rebalance）

Rebalance 是 Kafka 实现消费组高可用和分区分配动态调整的核心机制。当 Consumer 加入、离开或分区数变化时，触发 Rebalance，重新分配分区给组内所有 Consumer。

4.4.1 Rebalance 触发条件

• Consumer 加入/离开： 新 Consumer 加入组，或现有 Consumer 超时（session.timeout.ms）被踢出。
• 分区数变更： 管理员增加主题分区数。
• 订阅主题变更： Consumer 使用正则订阅时，匹配的主题列表发生变化。
• Consumer 调用 unsubscribe()： 主动取消订阅。

4.4.2 Rebalance 的代价

4.5 再均衡监听器：优雅处理位移

通过实现 ConsumerRebalanceListener 接口，可以在 Rebalance 前后执行自定义逻辑，避免位移丢失和重复消费。

consumer.subscribe(Collections.singletonList("topic-demo"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Rebalance 前：提交当前已处理但未提交的位移
        System.out.println("分区被回收，提交位移: " + partitions);
        consumer.commitSync(currentOffsets);  // 手动维护的位移 Map
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Rebalance 后：可以重新定位或初始化
        System.out.println("新分配分区: " + partitions);
        // 例如：从最早位移开始消费
        // consumer.seekToBeginning(partitions);
    }
});

4.6 本章小结

• 拉取模型： Consumer 主动 Pull，自主控制速率，是堆积诊断的起点。
• poll 循环： 核心引擎，参数 max.poll.records、max.poll.interval.ms 直接影响消费吞吐和稳定性。
• 位移提交： 自动提交简单但有重复风险；手动提交（异步+同步兜底）是生产推荐方案。
• Rebalance： 消费组高可用的代价是 STW 和重复消费，频繁 Rebalance 是堆积的常见诱因。
• 监听器： 利用 ConsumerRebalanceListener 在 Rebalance 前后保护位移，减少堆积影响。