Kafka Consumer 深入

📖 前置阅读：本文假设读者已掌握 SpringBoot Kafka 的基本消费操作（@KafkaListener）。如果还不熟悉，建议先阅读 SpringBoot Kafka 全操作指南。

一、⚡ 问题切入：消费者重启后，怎么知道上次读到哪了？

RabbitMQ 的答案是"消息消费后就删了，不需要记位置"。RocketMQ 的答案是"Broker 帮你记 offset"。Kafka 的答案是——消费者自己记，记在一个叫 __consumer_offsets 的内部 Topic 里：

消费者在 Partition-2 上消费到 offset=1500
        ↓ 提交 offset
__consumer_offsets Topic:
    Key:   (order-consumer-group, order-topic, 2)
    Value: offset=1500

消费者重启 ↓
        ↓ 读取 offset
从 offset=1501 继续消费

这个设计是 Kafka 和 RabbitMQ/RocketMQ 最核心的消费端差异——Kafka 的消费者对自己的消费进度负全责。如果消费者忘记提交 offset，重启后就会从上次提交的位置重新消费，产生重复消息。

二、Offset 提交机制

2.1 自动提交 vs 手动提交

Kafka 提供了两种 Offset 提交方式：

flowchart TD
classDef startEnd fill:#701a4c,stroke:#e11d48,stroke-width:2px,color:#fce7f3,font-weight:bold;
classDef condition fill:#2a1147,stroke:#a855f7,stroke-width:1.5px,color:#ede9fe,font-weight:bold;
classDef process fill:#1e1e24,stroke:#6b7280,stroke-width:1.5px,color:#e5e7eb;
classDef highlight fill:#450a0a,stroke:#dc2626,stroke-width:1.5px,color:#fecaca,font-weight:bold;

    POLL([poll 拉取消息]) --> PROCESS[处理消息]
    PROCESS --> MODE{提交模式}
    MODE -- "自动提交" --> AUTO["每隔 auto.commit.interval.ms\n自动提交最后一次 poll 的 offset"]
    MODE -- "手动提交" --> MANUAL["业务处理成功后\n显式调用 ack.acknowledge()"]

    AUTO --> RISK1["风险：消息还没处理完\n但 offset 已提交\n→ 进程挂了丢消息"]
    MANUAL --> RISK2["风险：消息已处理完\n但 offset 没提交\n→ 重启后重复消费"]

    class POLL startEnd;
    class MODE condition;
    class AUTO,MANUAL highlight;
    class RISK1,RISK2 process;

手动提交比自动提交更安全——至少你知道什么时候提交了。消息重复消费可以用幂等解决，但消息丢失无法恢复。

2.2 Spring Kafka 的七种提交模式

spring:
  kafka:
    listener:
      # 关闭自动提交——使用 Listener 级别的提交控制
      ack-mode: manual

// manual: 调用 acknowledge()，在下一轮 poll 时提交
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void handle(OrderMessage msg, Acknowledgment ack) {
    processOrder(msg);
    ack.acknowledge();  // 标记为已处理，下一轮 poll 时提交
}

// manual_immediate: 调用 acknowledge()，立即提交
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void handle(OrderMessage msg, Acknowledgment ack) {
    processOrder(msg);
    ack.acknowledge();  // 立即提交——不等下一轮 poll
}

⚠️ 新手提示：manual 和 manual_immediate 的核心区别——manual 是延迟提交（下一轮 poll 时提交），减少网络 I/O 次数；manual_immediate 是立即提交，可靠性更高但每次处理完都发生一次网络请求。单条消息处理时间长的选 manual_immediate，高吞吐量场景选 manual。

2.3 __consumer_offsets —— Kafka 的内部 Topic

所有 Offset 提交最终都写入 __consumer_offsets 这个内部 Topic：

__consumer_offsets (默认 50 个 Partition)

消息示例：
Key:   [ConsumerGroup: "order-group", Topic: "order-topic", Partition: 2]
Value: Offset=1500, Timestamp=2024-01-15T10:30:00

提交操作的本质：
    Producer (消费者侧) 向 __consumer_offsets 发送一条消息
    消息内容是"我在这个 Partition 的消费进度是 X"

可以用命令行查看某个 ConsumerGroup 的消费进度：

docker exec -it kafka \
  /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group order-consumer-group \
  --describe

# 输出示例：
# GROUP                  TOPIC         PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-consumer-group   order-topic   0          1500            1505            5
# order-consumer-group   order-topic   1          3200            3200            0
# order-consumer-group   order-topic   2          800             820             20

关键列：

三、Rebalance —— Partition 重新分配

3.1 什么触发 Rebalance

Rebalance 是 Kafka 的 ConsumerGroup 内部协调机制——当组内实例变化时，Partition 重新分配给各个消费者实例。

3.2 Rebalance 的过程

flowchart TD
classDef startEnd fill:#701a4c,stroke:#e11d48,stroke-width:2px,color:#fce7f3,font-weight:bold;
classDef condition fill:#2a1147,stroke:#a855f7,stroke-width:1.5px,color:#ede9fe,font-weight:bold;
classDef process fill:#1e1e24,stroke:#6b7280,stroke-width:1.5px,color:#e5e7eb;
classDef data fill:#052e16,stroke:#16a34a,stroke-width:1.5px,color:#bbf7d0,font-weight:bold;
classDef highlight fill:#450a0a,stroke:#dc2626,stroke-width:1.5px,color:#fecaca,font-weight:bold;

    TRIGGER([触发条件]) --> COORD[Group Coordinator Broker\n通知所有消费者 Rebalance]
    COORD --> REVOKE["① 所有消费者\n放弃当前分配的 Partition"]
    REVOKE --> LEADER["② Group Leader\n制定新的分配方案"]
    LEADER --> ASSIGN["③ 分配方案发回 Coordinator\nCoordinator 分发给所有消费者"]
    ASSIGN --> RESUME[④ 消费者从新的\noffset 开始消费]

    class TRIGGER startEnd;
    class COORD,LEADER highlight;
    class REVOKE,ASSIGN,RESUME process;

Rebalance 期间：整个 ConsumerGroup 停止消费。从 Revoke 到 Resume 之间，消息一直在 Partition 上堆积但不消费——这就是 Consumer Lag 突然飙升的原因。

3.3 四种分区分配策略

Kafka 支持四种分配策略，决定"哪个消费者分到哪些 Partition"：

// 演示 Range 策略的不均匀问题
// Topic-A 有 3 个 Partition，2 个消费者
Range 策略：
    消费者-1 → Partition-0, Partition-1  (2 个)
    消费者-2 → Partition-2               (1 个) ← 不均匀！

// 同样的场景，RoundRobin 策略：
RoundRobin 策略：
    消费者-1 → Partition-0, Partition-2  (2 个)
    消费者-2 → Partition-1               (1 个) ← 仍然不均匀

CooperativeSticky 的优势——增量 Rebalance：

传统的 Eager Rebalance（Range / RoundRobin / Sticky）：
    所有消费者 → 释放所有 Partition → 重新分配 → 继续消费
    ✗ 整个 ConsumerGroup 停止消费！

CooperativeSticky（增量 Rebalance）：
    只释放需要移动的 Partition → 其他 Partition 继续消费 → 分配新 Partition
    ✓ 只有部分 Partition 暂停消费

# 推荐配置——使用 CooperativeSticky
spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy:
          org.apache.kafka.clients.consumer.CooperativeStickyAssignor

3.4 Rebalance 与重复消费

Rebalance 是重复消费的第一来源：

消费者 A 正在处理 Partition-0 的 offset=100~150
    ↓ 处理到 offset=120 时，Rebalance！
Partition-0 被分配给消费者 B
    ↓
消费者 B 从上次提交的 offset=100 开始消费
    ↓
offset=100~120 的消息 → 被消费者 A 和消费者 B 都处理了 → 重复！

解决方案：

1. Consumer 做幂等——用业务 Key 去重（orderId + action 作为去重键）
2. 调大超时时间——减少不必要的 Rebalance
3. 使用 CooperativeSticky——减少 Rebalance 影响范围

// 幂等消费——不管 Rebalance 怎么重分配，同一条消息不会被处理两次
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void handle(ConsumerRecord<String, OrderMessage> record, Acknowledgment ack) {
    String idempotentKey = "kafka:consumed:" + record.topic()
            + ":" + record.partition() + ":" + record.offset();

    // Redis SETNX 原子判重
    Boolean firstTime = redisTemplate.opsForValue()
            .setIfAbsent(idempotentKey, "1", Duration.ofHours(24));

    if (Boolean.FALSE.equals(firstTime)) {
        log.warn("重复消息，跳过: offset={}", record.offset());
        ack.acknowledge();  // 重复消息也要提交——否则下次还拉回来
        return;
    }

    try {
        processOrder(record.value());
        ack.acknowledge();
    } catch (Exception e) {
        // 处理失败 → 删除幂等标记，允许重试时重新处理
        redisTemplate.delete(idempotentKey);
        // 不调用 acknowledge() → 消息会被重新投递
    }
}

3.5 Rebalance 相关的关键超时参数

max.poll.interval.ms 是最容易出问题的参数——如果消费者处理一批消息的时间超过了这个值（默认 5 分钟），Kafka 认为消费者卡住了，触发 Rebalance：

# 如果单条消息处理耗时 2 秒，max.poll.records=500
# 最坏情况：500 条 × 2 秒 = 1000 秒 ≈ 16 分钟 > 5 分钟 → Rebalance！
# 解决方案：
spring:
  kafka:
    consumer:
      max-poll-records: 50           # 每次只拉 50 条
      properties:
        max.poll.interval.ms: 600000  # 拉到 10 分钟

四、多线程消费模型

4.1 Kafka 的消费线程特点

Kafka 和 RabbitMQ/RocketMQ 有一个重要的消费端差异：Kafka 的单个 Consumer 实例不是线程安全的——不能在多个线程中共享同一个 KafkaConsumer 实例。

但可以用以下方式实现多线程消费：

4.2 方式一：concurrency 配置（推荐）

// 同一个 @KafkaListener 起 3 个 KafkaConsumer 线程
// 每个线程独立 poll，独立提交 offset
@KafkaListener(
    topics = "order-topic",
    groupId = "order-group",
    concurrency = "3"   // ← 起 3 个 KafkaConsumer 实例
)
public void handle(OrderMessage msg, Acknowledgment ack) {
    processOrder(msg);
    ack.acknowledge();
}

# 或者在配置文件中统一设置
spring:
  kafka:
    listener:
      concurrency: 3

concurrency 的含义：

Topic: order-topic (6 个 Partition)
ConsumerGroup: order-group, concurrency=3

KafkaConsumer-1 → Partition-0, Partition-1  (独立的线程 + TCP 连接)
KafkaConsumer-2 → Partition-2, Partition-3
KafkaConsumer-3 → Partition-4, Partition-5

每个线程独立 poll、独立提交 offset——互不影响

⚠️ 新手提示：concurrency 不要超过 Topic 的 Partition 总数。6 个 Partition + concurrency=10 = 有 4 个 KafkaConsumer 永远闲着——Kafka 规定一个 Partition 只能被同一个 ConsumerGroup 内的一个消费者消费。

4.3 方式二：消息内异步处理（小心 Offset 丢失）

@KafkaListener(topics = "order-topic", groupId = "order-group")
public void handle(OrderMessage msg, Acknowledgment ack) {
    // 收到消息后立即提交 offset——然后用线程池异步处理
    ack.acknowledge();

    // 异步处理——注意：如果这里抛异常，消息已经"丢了"
    CompletableFuture.runAsync(() -> {
        try {
            processOrder(msg);
        } catch (Exception e) {
            log.error("异步处理失败: orderId={}", msg.getOrderId(), e);
            // 消息的 offset 已提交——无法重试了！
        }
    });
}

这种方式有风险——offset 已提交，如果异步处理失败，消息就丢了。只适用于"丢了也没关系"的非关键业务。

4.4 方式三：池化处理 + 批量提交

更好的做法是——先处理整批，处理成功后统一提交：

@KafkaListener(topics = "order-topic", groupId = "order-group")
public void handleBatch(List<OrderMessage> messages, Acknowledgment ack) {
    // 用线程池并发处理一批消息
    List<CompletableFuture<Void>> futures = messages.stream()
        .map(msg -> CompletableFuture.runAsync(() -> processOrder(msg)))
        .toList();

    // 等待整批处理完成
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
        .join();  // 阻塞等待——整批处理完

    // 全部成功才提交
    ack.acknowledge();
}

三种多线程方式对比：

五、完整 Consumer 配置速查

🎯 总结

1. Offset 是消费者自己的责任：存在 __consumer_offsets 内部 Topic，由消费者提交。自动提交有丢消息风险，手动提交有重复消费风险——手动提交 + 幂等去重是生产标配。

2. Rebalance 是重复的第一来源：实例增减时 Partition 重新分配，正在处理但未提交的消息会被新消费者重新拉取。解决方案是幂等 + CooperativeSticky 增量 Rebalance。

3. max.poll.interval.ms 最容易被忽略：单批消息处理时间超过这个值就会触发 Rebalance。处理慢时调大这个值或调小 max.poll.records。

4. 多线程消费首选 concurrency：每个线程独立的 KafkaConsumer 实例——线程安全、可靠性最高。concurrency 不要超过 Partition 数。

5. CooperativeSticky > Sticky > RoundRobin > Range：Kafka 2.4+ 默认用 CooperativeSticky——最均匀且支持增量 Rebalance。

📖 下一步阅读：消费端的全部控制力都搞清楚了。接下来进入 Kafka 的流处理引擎——Kafka Streams DSL、精确一次语义、Log Compaction。继续阅读 Kafka Streams 与高级特性。