kafka

03. Kafka 消费者组与偏移量管理

Kafka 消费者: Consumer Group 机制、Rebalance、自动/手动提交偏移量

Leeting Yan 4 分钟阅读 1976 字

Kafka 消费者是整个流处理链路的终点，也是大多数系统故障暴露的位置。消费者组（Consumer Group）的 Rebalance 机制、偏移量的提交策略、以及 Exactly-Once 消费的实现，是生产环境中最需要精细调优的部分。

1. 消费者组与分区分配

消费者组是 Kafka 实现高并发消费的核心机制。同一 Consumer Group 内的消费者共享 Topic 的订阅，每个 Partition 只能被组内一个消费者消费。

Topic: orders (5 partitions)

Consumer Group: order-processors (3 consumers)
   Consumer 1  ──→ Partition 0, Partition 1
   Consumer 2  ──→ Partition 2, Partition 3
   Consumer 3  ──→ Partition 4

───────────────────────────────────────
新增 Consumer 4 后触发的 Rebalance:
   Consumer 1  ──→ Partition 0
   Consumer 2  ──→ Partition 1
   Consumer 3  ──→ Partition 2
   Consumer 4  ──→ Partition 3, Partition 4

这个分配过程由 Group Coordinator 管理，每个 Consumer Group 会有一个 Broker 被选举为 Coordinator。

2. Rebalance 协议详解

Rebalance 是指消费者组内成员的 Partition 重新分配过程。理解 Rebalance 的触发条件和机制，是避免消费停顿的关键。

触发条件

条件说明影响
新消费者加入Consumer Group 扩容短暂停顿
消费者离开消费者进程退出或心跳超时短暂停顿
订阅变更消费者订阅的 Topic 列表变化短暂停顿
分区变化Topic 的分区数增加需要重新分配

分配策略

props.put("partition.assignment.strategy",
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
策略分配方式Rebalance 行为
Range（默认）按 Topic 范围分配全局 Rebalance，全量重新分配
RoundRobin全局轮询全局 Rebalance
Sticky尽量保持现有分配全局 Rebalance，但减少移动
CooperativeSticky（推荐）增量 Rebalance只移动必要的分区，消费不中断

CooperativeSticky 的增量 Rebalance 是 Kafka 2.4+ 引入的重大改进。传统 Rebalance 会暂停整个消费者组的所有消费，而 CooperativeSticky 通过两阶段协议，只"撤销（revoke）“必要的分区，然后"重新分配”，未被影响的分区可以继续消费。

3. 偏移量提交策略

偏移量（Offset）是消费者追踪读取进度的核心。Kafka 提供多种提交模式，适用于不同的可靠性需求：

自动提交（默认，最简单但不可靠）

props.put("enable.auto.commit", true);
props.put("auto.commit.interval.ms", 5000); // 每 5 秒自动提交

// 风险: 消息可能已被消费但尚未处理完，此时提交可能导致"消息丢失"
// 或: 消息已处理但提交失败，导致重复消费

手动同步提交（精确控制）

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

consumer.subscribe(Arrays.asList("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record); // 处理消息
    }
    try {
        consumer.commitSync(); // 处理完一批后精确提交
    } catch (CommitFailedException e) {
        // Rebalance 期间提交失败，需要重新消费这批次
        log.warn("Commit failed, messages will be re-consumed", e);
    }
}

手动异步提交（高吞吐场景）

// 异步提交不阻塞，适合对延迟敏感的场景
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        log.error("Async commit failed: {}", offsets, exception);
    }
});

按分区精确提交（最细粒度）

for (TopicPartition partition : records.partitions()) {
    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();

    // 只提交已处理的分区
    Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
    offsetMap.put(partition, new OffsetAndMetadata(lastOffset + 1));
    consumer.commitSync(offsetMap);
}

4. 心跳与会话超时

消费者通过心跳向 Group Coordinator 报告存活状态。心跳间隔和超时配置直接影响故障检测速度：

// 心跳配置
props.put("heartbeat.interval.ms", 3000);    // 每 3 秒发送一次心跳
props.put("session.timeout.ms", 10000);       // 10 秒内未收到心跳视为离线

// 注意: session.timeout.ms 必须 ≥ 3 * heartbeat.interval.ms

// 轮询超时（max.poll.interval.ms > session.timeout.ms）
props.put("max.poll.interval.ms", 300000);    // 5 分钟内必须调用 poll()
// 如果消息处理耗时过长，需增大此值，否则会被踢出消费者组

处理耗时消息的策略：

// 方案 1: 增大 max.poll.interval.ms
props.put("max.poll.interval.ms", 600000); // 10 分钟

// 方案 2: 将耗时任务异步化， poll() 保持活跃
ExecutorService executor = Executors.newFixedThreadPool(10);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        executor.submit(() -> process(record));
    }
    consumer.pause(consumer.assignment());   // 暂停获取新消息
    // 等待批次处理完成...
    consumer.resume(consumer.assignment());  // 恢复消费
}

// 方案 3: 使用独立线程发送心跳（Kafka 2.3+ 后不需要，已内置）

5. Exactly-Once 消费

在 Kafka 中实现 Exactly-Once 消费，需要配合事务生产者 + 事务消费者 + 外部系统的事务支持：

// Kafka → Kafka 场景的 Exactly-Once
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka1:9092");
consumerProps.put("group.id", "etl-processors");
consumerProps.put("isolation.level", "read_committed"); // 只读取已提交的事务消息
consumerProps.put("enable.auto.commit", false);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

producer.initTransactions();
consumer.subscribe(Arrays.asList("input-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    if (records.isEmpty()) continue;

    producer.beginTransaction();
    for (ConsumerRecord<String, String> record : records) {
        String transformed = transform(record.value());
        producer.send(new ProducerRecord<>("output-topic", transformed));
    }

    // 将消费偏移量作为事务的一部分提交
    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
    for (TopicPartition partition : records.partitions()) {
        long lastOffset = records.records(partition).getLast().offset();
        offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
    }
    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
    producer.commitTransaction();
}

关键要点：

  • 消费者设置 isolation.level=read_committed，过滤掉未提交的事务消息
  • sendOffsetsToTransaction 将消费偏移量和消息发送绑定在同一个事务中
  • 如果处理失败，整个事务回滚，消息不会被标记为已消费

6. 消费性能优化

# 批量拉取（减少网络往返）
fetch.min.bytes=50000          # 每次拉取至少 50KB
fetch.max.wait.ms=500          # 最多等待 500ms
max.poll.records=500           # 每次 poll 返回最多 500 条

# 消费并发
max.poll.interval.ms=300000    # 处理超时 5 分钟
session.timeout.ms=30000       # 会话超时 30 秒
heartbeat.interval.ms=10000    # 心跳间隔 10 秒

消费组 Rebalance 深度解析

Rebalance 是消费者组协调的核心机制，但频繁 rebalance 会导致消费停顿：

分配策略算法适用场景缺点
Range按分区范围分配分区数均匀的 topic分区数不均时倾斜
RoundRobin轮询分配同构消费者无状态，重启后全变
Sticky尽量保持原有分配大多数场景实现复杂
Cooperative增量再平衡（Kafka 2.4+）避免全局停顿需要客户端支持

推荐 CooperativeStickyAssignor：Kafka 2.4+ 默认策略，避免传统 eager rebalance 的「全组停止消费」问题，仅对有变化的分区进行迁移。

Consumer Lag 监控：

# 查看消费延迟
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group --describe

# 输出关键指标：
# CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# 1200000         1200500         500  consumer-1

Lag 持续上升通常是消费者处理能力不足的信号，应监控并告警。

延伸阅读

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「kafka」更多文章

  1. 集群升级与滚动重启实践
  2. Kafka 应用测试策略：Testcontainers 与集成测试
  3. 压缩算法选型：lz4、zstd、snappy 与 gzip