ZH
FrançaisDeutschTiếng ViệtEspañolEnglishPortuguês日本語简体中文한국어
首页 / 架构与设计 / Kafka

Kafka 中的 Consumer Group 与 Rebalance 在 Spring Boot 中的应用

Cris The Dev · 2026/10/13 · 9 分钟阅读 · 217

1. 概述

当运行多个相同服务的实例从 Kafka 读取时，你需要知道哪个消息由哪个实例处理。Kafka 通过 consumer group 和 rebalance 机制来解决这个问题。

本文解释了 Kafka 如何在组内分配分区，以及何时发生 rebalance。之后，你将配置 Spring Boot 使 rebalance 发生得更少，并确保不会丢失或重复数据。文末有一个使用 Embedded Kafka 的测试来验证分区分配方式。

2. Consumer Group 如何工作

consumer group 是具有相同 group.id 的一组 consumer。Kafka 确保：

  • 主题的每个分区在任何时刻都分配给组内恰好一个 consumer。
  • 一个 consumer 可以接收多个分区。
  • 不同的组独立读取。每个组都有各自的 offset。

假设 orders 主题有 4 个分区：

Group "billing" 中有 2 个 consumer：
  consumer-A -> orders-0, orders-1
  consumer-B -> orders-2, orders-3

Group "billing" 中有 5 个 consumer：
  consumer-A -> orders-0
  consumer-B -> orders-1
  consumer-C -> orders-2
  consumer-D -> orders-3
  consumer-E -> （闲置，没有分区）

重要的后果是分区数限制了一个组的最大并行度。添加超过分区数的 consumer 不会加快处理速度。

协调由一个 broker 担任，称为 group coordinator。Consumer 定期向 coordinator 发送 heartbeat 来表示它仍然活着。

2.1. 依赖

添加 Spring for Apache Kafka。Spring Boot 3.x 已经管理了版本：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

2.2. 一个简单的 Listener

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);

    @KafkaListener(topics = "orders", groupId = "billing", concurrency = "2")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("partition={} offset={} order={}",
                record.partition(), record.offset(), record.value());
    }
}
  • groupId = "billing" 将 consumer 放入 billing 组。
  • concurrency = "2" 在同一实例中创建 2 个 consumer（2 个线程）。如果运行 2 个服务实例，该组将有 4 个 consumer，每个 consumer 获得 1 个分区。

启动时的日志显示分配结果：

billing: partitions assigned: [orders-0, orders-1]
billing: partitions assigned: [orders-2, orders-3]

3. Rebalance 是什么以及何时发生

Rebalance 是组在其成员之间重新分配分区的过程。它在以下情况下发生：

  1. 新 consumer 加入，例如你扩展了 pod 或部署了新实例。
  2. Consumer 正常离开，例如通过调用 close() 进行关闭。
  3. Consumer 被认为已死亡：coordinator 在 session.timeout.ms 内没有收到 heartbeat。
  4. Consumer 处理过慢：两次 poll() 调用之间的间隔超过 max.poll.interval.ms。Consumer 自动离开组。
  5. 主题变化：增加分区数，或有新主题与已订阅的模式匹配。

问题是在旧的 eager 协议中，所有 consumer 必须先归还所有分区，然后才能重新分配。整个组在此期间停止处理，这种现象通常称为 “stop-the-world”。对于大型组或滚动部署，rebalance 可能会反复发生。

4. 重要的超时参数

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: billing
      enable-auto-commit: false
      max-poll-records: 100
      properties:
        session.timeout.ms: 45000
        heartbeat.interval.ms: 3000
        max.poll.interval.ms: 300000
参数 含义
session.timeout.ms 没有 heartbeat 的最大时间，超过该时间后 consumer 将被移除
heartbeat.interval.ms Heartbeat 发送周期，通常 ≤ 1/3 的 session.timeout.ms
max.poll.interval.ms 两次 poll() 调用之间的最大间隔
max-poll-records 每次 poll() 的最大记录数

Heartbeat 从单独的后台线程发送。因此，即使主线程繁忙，consumer 仍然在 session.timeout.ms 内保持 “活着” 状态。但是 max.poll.interval.ms 只测量处理线程。如果一个批次处理时间过长，consumer 仍然会被踢出组。

日志中的典型错误看起来像这样：

Member consumer-billing-1 sending LeaveGroup request to coordinator due to
consumer poll timeout has expired. This means the time between subsequent
calls to poll() was longer than the configured max.poll.interval.ms ...
CommitFailedException: Commit cannot be completed since the group has already rebalanced

需要满足的公式：

max-poll-records × 处理 1 条记录的时间  <  max.poll.interval.ms

注意： 不要通过将 max.poll.interval.ms 增加到几小时来“修复”。真正挂起的 consumer 会长时间持有分区而没有人处理。减少 max-poll-records，或将繁重任务移出 consume 线程。

5. 减少 Rebalance 的影响

5.1. Cooperative Rebalancing

使用 CooperativeStickyAssignor，只有真正需要转移的分区才会被收回。其他 consumer 在 rebalance 进行时继续处理自己的分区。

spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

示例：组中有 A（orders-0..3），然后 B 加入。

Eager:       A 归还全部 4 个分区 -> 重新分配：A {0,1}, B {2,3}
Cooperative: A 只归还 {2,3}，继续处理 {0,1} -> B 获得 {2,3}

注意： 不能在一次部署中直接将运行中的组从 eager 切换到 cooperative。需要进行两次滚动部署。第一次声明两个 assignor，例如 CooperativeStickyAssignor,RangeAssignor。第二次才移除 RangeAssignor。

从 Kafka 4.0 起，新的 rebalance 协议（KIP-848）已 GA。你可以通过 group.protocol=consumer 来启用它。此时分配由 broker 执行，rebalance 默认具有增量特性，客户端的 partition.assignment.strategy 和 session.timeout.ms 参数不再使用。如果你的集群支持此功能，这是首先应该考虑的选项。

5.2. 静态成员

每次 pod 重启时，consumer 获得一个新的 member.id，这会触发两次 rebalance（离开然后重新加入）。Static membership 为 consumer 分配一个固定的标识符：

spring:
  kafka:
    consumer:
      properties:
        group.instance.id: billing-${HOSTNAME}
        session.timeout.ms: 60000

如果 consumer 在 session.timeout.ms 内回到，coordinator 会直接将相同的分区分配回来，无需 rebalance。这种方式非常适合 Kubernetes StatefulSet，其中 HOSTNAME 是稳定的（billing-0、billing-1 等）。

注意： group.instance.id 在组内必须唯一。当 concurrency > 1 时，Spring Kafka 会自动为每个 consumer 子实例添加后缀 -n。如果两个实例使用相同的 ID，后进入的实例将“踢出”前一个实例（FencedInstanceIdException）。不要对具有随机 pod 名称的 Deployment 使用 static membership，因为你不会获得任何好处。

6. Rebalance 时的 Offset 处理

rebalance 的最大风险是重复处理。Consumer A 已处理完 record 但尚未提交，分区转移给 B，B 从最后提交的 offset 重新读取。

Spring Kafka 默认在每个批次后提交（AckMode.BATCH），并在分区被收回前自动提交待处理的 offset。如果需要干预，例如刷新缓冲区或记录日志，可以使用 ConsumerAwareRebalanceListener：

@Component
public class OrderRebalanceListener implements ConsumerAwareRebalanceListener {

    private static final Logger log = LoggerFactory.getLogger(OrderRebalanceListener.class);

    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
                                                Collection<TopicPartition> partitions) {
        log.info("即将失去分区 {}，刷新内存中的数据", partitions);
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
                                               Collection<TopicPartition> partitions) {
        log.info("已提交并归还分区 {}", partitions);
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
                                     Collection<TopicPartition> partitions) {
        log.info("收到分区 {}", partitions);
    }

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer,
                                 Collection<TopicPartition> partitions) {
        log.warn("丧失分区 {}，未能及时提交", partitions);
    }
}
  • onPartitionsRevokedBeforeCommit 在容器提交 offset 前运行。这是完成未完成工作的地方。
  • onPartitionsRevokedAfterCommit 在 offset 已提交后运行。
  • onPartitionsLost 在 consumer 被踢出组时调用，例如超出 max.poll.interval.ms。此时分区可能已属于另一个 consumer，所以不要提交。

向容器工厂注册 listener：

@Configuration
public class KafkaConfig {

    @Bean
    ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> consumerFactory,
            OrderRebalanceListener rebalanceListener) {

        var factory = new ConcurrentKafkaListenerContainerFactory<Object, Object>();
        configurer.configure(factory, consumerFactory);
        factory.getContainerProperties().setConsumerRebalanceListener(rebalanceListener);
        return factory;
    }
}

configurer.configure(...) 保留 application.yml 中的所有配置。下一行只是添加 rebalance listener。

注意： 无论配置多好，Kafka 仅对普通 consumer 提供 at-least-once 保证。使你的处理幂等，例如在调用 PaymentService 之前检查 orderId 是否已支付。不要假设 rebalance 永远不会导致重复。

7. 测试分区分配

以下测试使用 @EmbeddedKafka 来验证同一组中的两个 consumer 均匀分配 4 个分区：

@EmbeddedKafka(partitions = 4, topics = "orders")
class ConsumerGroupAssignmentTest {

    @Test
    void twoConsumersInSameGroupShareAllPartitions(EmbeddedKafkaBroker broker) {
        Map<String, Object> props = KafkaTestUtils.consumerProps("billing", "false", broker);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        try (var consumerA = new KafkaConsumer<String, String>(props);
             var consumerB = new KafkaConsumer<String, String>(props)) {

            consumerA.subscribe(List.of("orders"));
            consumerB.subscribe(List.of("orders"));

            long deadline = System.currentTimeMillis() + 30_000;
            while (System.currentTimeMillis() < deadline
                    && !(consumerA.assignment().size() == 2 && consumerB.assignment().size() == 2)) {
                consumerA.poll(Duration.ofMillis(200));
                consumerB.poll(Duration.ofMillis(200));
            }

            Set<TopicPartition> all = new HashSet<>(consumerA.assignment());
            all.addAll(consumerB.assignment());

            assertThat(consumerA.assignment()).hasSize(2);
            assertThat(consumerB.assignment()).hasSize(2)
                    .doesNotContainAnyElementsOf(consumerA.assignment());
            assertThat(all).hasSize(4);
        }
    }
}

说明：

  • @EmbeddedKafka 启动内存中的 broker，创建具有 4 个分区的 orders 主题。Broker 直接作为测试方法的参数注入。
  • 两个 KafkaConsumer 共享 group.id = "billing"。
  • 循环在两个 consumer 上交替调用 poll()，因为 consumer 只在 poll() 内参与 rebalance。如果只有 consumer A 进行 poll，rebalance 将无法完成。两个 consumer 运行在同一线程上但是两个独立对象，所以不违反 “一个 consumer 只能被一个线程使用” 的规则。
  • 断言检查每个 consumer 有 2 个分区、彼此不重叠，且总共覆盖了 4 个分区。

这也是测试不同 partition.assignment.strategy 的好方法。添加第三个 consumer，关闭一个 consumer，然后观察 assignment() 的变化。

8. 常见错误

  • Consumer 数量超过分区数：多余的 consumer 将闲置。增加 concurrency 不会加快系统速度。
  • Listener 中的同步处理过长：超过 max.poll.interval.ms 会导致 rebalance 循环。Consumer 被踢出，回到，接收相同批次，然后再次被踢出。
  • 在 onPartitionsLost 中手动提交：导致 CommitFailedException，或更糟的是覆盖新 consumer 的 offset。
  • 在两个不同的服务之间共享 group.id：两个服务将“分享”消息而不是各自接收完整消息。每个独立服务应该有自己的组。
  • 在大型组上使用 eager 协议进行滚动部署：每次 pod 重启都会导致两次 stop-the-world。使用 cooperative、static membership 或 Kafka 4.x 的新协议。

9. 总结

Consumer group 允许你在水平方向上扩展 Kafka 读取，其上限是分区数。Rebalance 是这种灵活性的代价。当成员加入、离开、死亡或处理过慢时，它就会发生。

为了减少 rebalance 的危害，你应该做以下事情：

  • 合理配置 max.poll.interval.ms 和 max-poll-records。
  • 使用 CooperativeStickyAssignor，或在 Kafka 4.x 上使用 group.protocol=consumer。
  • 在基础设施允许的情况下启用 static membership。
  • 通过 ConsumerAwareRebalanceListener 处理分区生命周期。

最后，始终设计幂等处理，因为 at-least-once 是你需要接受的默认保证。

kafkaspring-kafkaconsumer-grouprebalancespring-boot
作者的应用Reflex Arena11 brain-speed games, live 1v1 duels, monthly seasonsiOS · Android

同主题文章