消费者Rebalance机制

优质博文：IT-BLOG-CN

一、消费者Rebalance机制

在Apache Kafka中，消费者组
Consumer Group会在以下几种情况下发生重新平衡Rebalance：
【1】消费者加入或离开消费者组： 当一个新的消费者加入消费者组或一个现有的消费者离开消费者组时，Kafka会触发重新平衡，以重新分配分区给消费者。
【2】消费者崩溃或失去连接： 如果Kafka检测到某个消费者崩溃或失去连接（例如，由于网络问题或消费者进程被终止），它会触发重新平衡。
【3】主题的分区数量发生变化： 如果一个主题的分区数量增加或减少，Kafka会触发重新平衡，以确保新的分区被分配给消费者组中的消费者。
【4】消费者组协调器变更： 消费者组协调器是负责管理消费者组的一个Kafka Broker。如果消费者组协调器发生变更（例如，协调器所在的Broker崩溃），也会触发重新平衡。
【5】消费者组成员发送心跳失败： 消费者需要定期向消费者组协调器发送心跳heartbeat以表明它们仍然活跃。如果心跳失败，协调器会认为该消费者已经失去连接，从而触发重新平衡。

rebalance只针对subscribe这种不指定分区消费的情况，如果通过assign这种消费方式指定了分区，kafka不会进行rebanlance。

Kafka在高峰期重平衡rebalancing会导致消费者组的停顿，影响系统的性能和稳定性。为了避免在高峰期发生重平衡，可以采取以下几种策略：
【1】优化分区分配策略： 使用RangeAssignor或StickyAssignor等分区分配策略来减少重平衡的频率和影响。

RangeAssignor 是Kafka默认的分区分配策略之一，它将分区按范围分配给消费者。

我们通过一个具体的例子来说明RangeAssignor如何分配分区。

假设我们有一个Kafka主题my-topic，它有6个分区P0, P1, P2, P3, P4, P5，并且我们有3个消费者C1, C2, C3在一个消费者组中。

初始分配：假设初始分配如下：

C1: P0, P1
C2: P2, P3
C3: P4, P5

消费者组成员变化：现在假设C2离开了消费者组，那么RangeAssignor会重新分配分区，以确保分区尽量按顺序和均匀地分配给剩余的消费者。新的分配可能如下：

C1: P0, P1, P2
C3: P3, P4, P5

在这个过程中，RangeAssignor将分区按顺序重新分配给剩余的消费者，确保每个消费者分配到的分区尽量连续。

新消费者加入：现在假设有一个新消费者C4加入了消费者组，RangeAssignor会再次按顺序和均匀地分配分区。新的分配可能如下：

C1: P0, P1
C3: P2, P3
C4: P4, P5

在这个过程中，RangeAssignor将分区重新分配，以确保每个消费者分配到的分区尽量连续和均匀。

通过这个例子，我们可以看到RangeAssignor的分配策略：
1、将分区按顺序分配给消费者。
2、当消费者组成员变化时，重新分配分区，以确保分区尽量按顺序和均匀地分配给所有消费者。
3、分区分配尽量保持连续性。
这种策略的好处是分区分配简单且稳定，减少了分区在消费者组成员变化时的重新分配范围，从而减少了重平衡的频率和影响。

以下是配置RangeAssignor的代码示例：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Properties;

public class RangeAssignorExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "example-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 设置分区分配策略为 RangeAssignor
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RangeAssignor");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅主题
        consumer.subscribe(List.of("example-topic"));

        // 消费消息的逻辑
        // ...
    }
}

StickyAssignor 是Kafka 2.4及以上版本引入的一种分区分配策略，它的目标是尽量保持分区分配的稳定性，减少重平衡的频率。

我们通过一个具体的例子来说明StickyAssignor如何分配分区。

假设我们有一个Kafka主题my-topic，它有6个分区P0, P1, P2, P3, P4, P5，并且我们有3个消费者C1, C2, C3在一个消费者组中。

初始分配：假设初始分配如下：

C1: P0, P1
C2: P2, P3
C3: P4, P5

消费者组成员变化：现在假设C2离开了消费者组，那么StickyAssignor会尽量保持现有的分区分配不变，并重新分配C2的分区。新的分配可能如下：

C1: P0, P1, P2
C3: P3, P4, P5

在这个过程中，StickyAssignor尽量保持C1和C3的分区分配不变，只是将C2的分区重新分配给其他消费者。

新消费者加入：现在假设有一个新消费者C4加入了消费者组，StickyAssignor会尝试保持现有的分区分配不变，并将分区尽量均匀地分配给所有消费者。新的分配可能如下：

C1: P0, P1
C3: P4, P5
C4: P2, P3

在这个过程中，StickyAssignor保持了C1和C3的分区不变，并将C2的分区重新分配给C4。

通过这个例子，我们可以看到StickyAssignor的分配策略：
1、尽量保持现有的分区分配不变。
2、当消费者组成员变化时，尽量最小化分区在消费者之间的移动。
3、尽量保持分区分配的平衡性。
这种策略的好处是减少了重平衡带来的影响，提高了分区分配的稳定性，减少了因分区移动带来的数据重新加载和处理的开销。

以下是配置StickyAssignor的代码示例：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Properties;

public class StickyAssignorExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "example-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 设置分区分配策略为 StickyAssignor
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.StickyAssignor");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅主题
        consumer.subscribe(List.of("example-topic"));

        // 消费消息的逻辑
        // ...
    }
}

或者在配置中进行指定

group.id=my-consumer-group
partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor

【2】增加session.timeout.ms和heartbeat.interval.ms：增加session.timeout.ms和heartbeat.interval.ms的值，这样可以减少消费者因为心跳超时而被认为失效，从而触发重平衡。

1、session.timeout.ms是消费者与Kafka broker之间的会话超时时间。如果在这个时间内Kafka broker没有收到某个消费者的心跳，broker就会认为该消费者已经失效，并触发重平衡。
2、heartbeat.interval.ms是消费者发送心跳给Kafka broker的时间间隔。心跳是消费者向broker表示自己仍然活跃的方式。

session.timeout.ms=30000
heartbeat.interval.ms=3000

3、heartbeat.interval.ms的值通常要远小于session.timeout.ms的值。这样可以确保在会话超时之前，消费者有多次机会发送心跳。一般建议session.timeout.ms至少是heartbeat.interval.ms的10倍，以确保有足够的时间进行多次心跳尝试。

【3】合理配置消费者组：确保消费者组中的消费者数量稳定，避免频繁地增加或减少消费者。尽量在低峰期进行消费者的添加或移除操作。

【4】优化消费者性能：提高消费者的处理能力，确保消费者能够及时处理消息，避免因为处理延迟导致的重平衡。使用异步处理或批量处理来提高消费者的吞吐量。

【5】监控和报警：实时监控Kafka集群和消费者组的状态，设置报警机制，当检测到重平衡风险时，及时采取措施。

【6】使用静态成员Static Membership：Kafka 2.3及以上版本支持静态成员功能，可以通过配置group.instance.id来减少重平衡的频率。

group.instance.id是Kafka 2.4.0引入的一个配置项，用于为每个消费者实例指定一个唯一的标识符。当消费者组中的消费者具有唯一的group.instance.id时，Kafka可以更智能地处理消费者组成员的变化，从而减少不必要的重平衡。

静态成员：通过配置group.instance.id，消费者实例变成了“静态成员”，即使它们暂时断开连接，Kafka也会保留它们的成员身份。这与传统的动态成员（没有group.instance.id）不同，动态成员在断开连接后会被移除，从而触发重平衡。

group.id=my-consumer-group
group.instance.id=consumer-instance-1

【7】调整rebalance.timeout.ms：增加rebalance.timeout.ms的值，确保消费者有足够的时间完成重平衡过程，避免因超时导致的频繁重平衡。

消费者Rebalance分区分配策略

主要包含四种relalance策略：RangeAssignor（范围分配策略），RoundRobinAssignor（轮询分配策略），StickyAssignor（粘性分配策略），CooperativeStickyAssignor（协作粘性分配策略），之前已经讲过两个，这里聊聊剩下的两个

`RoundRobinAssignor`（轮询分配策略）

RoundRobinAssignor采用轮询的方式将分区分配给消费者。它会将所有分区和消费者按照字典顺序排序，然后依次将每个分区分配给下一个消费者，直到所有分区都被分配完毕。

`CooperativeStickyAssignor`（协作粘性分配策略）

CooperativeStickyAssignor是StickyAssignor的改进版本，它引入了协作重平衡的概念，使得重平衡过程更加平滑，减少了重平衡期间的停顿时间。

二、Rebalance 过程

第一阶段：选择"组协调器"
组协调器GroupCoordinator：每个consumer group都会选择一个broker作为自己的组协调器coordinator，负责监控这个消费组里的所有消费者的心跳，以及判断是否宕机，然后开启消费者rebalance。

consumer group中的每个consumer启动时会向kafka集群中的某个节点发送FindCoordinatorRequest请求来查找对应的组协调器GroupCoordinator，并跟其建立网络连接。

组协调器选择方式：consumer消费的offset要提交到__consumer_offsets的哪个分区，这个分区leader对应的broker就是这个consumer group的coordinator

第二阶段：加入消费组JOIN GROUP
在成功找到消费组所对应的GroupCoordinator之后就进入加入消费组的阶段，在此阶段的消费者会向GroupCoordinator发送JoinGroupRequest请求，并处理响应。然后GroupCoordinator从一个consumer group中选择第一个加入group的consumer作为leader(消费组协调器)，把consumer group情况发送给这个leader，接着这个leader会负责制定分区方案。

第三阶段：SYNC GROUP
consumer leader通过给GroupCoordinator发送SyncGroupRequest，接着GroupCoordinator就把分区方案下发给各个consumer，他们会根据指定分区的leader broker进行网络连接以及消息消费。