【Kafka系列】Kafka的再均衡机制_kafka rebalnce

什么是再均衡（Re-Blanced）机制？

​​再均衡是Kafka中用于重新分配消费者组（consumer group）中消费者（consumer）和分区（partition）之间关系的机制。当消费者组内的消费者数量变化，或者消费者订阅的主题发生变化（新增或删除分区），都可能触发再平衡。Kafka提供了三种再平衡策略：Round Robin（轮询），Range（范围）和Sticky（粘性）。

1. Round Robin（轮询）： 这种策略会以轮询的方式将所有分区依次分配给消费者，确保每个消费者都能均匀地获得分区。

2. Range（范围）： Range策略首先计算每个消费者可以消费的分区个数，然后按照顺序将指定个数范围的分区分配给各个消费者。这有助于均衡分配消费压力。

3. Sticky（粘性）： Sticky是较新版本中新增的策略，旨在解决Round Robin和Range策略可能导致某些消费者负载过重的问题。Sticky策略在保持均衡的基础上，尽可能保持未宕机的消费者仍然消费它们之前负责的分区，以减少不必要的再平衡。

再平衡会在以下情况发生时触发：

1. 新增或删除消费者：当消费者组中新增或删除消费者时，需要重新分配分区。

2. 消费者订阅主题发生变化：例如，使用正则表达式订阅的主题，新增符合条件的主题时会触发再平衡。

3. 主题新增分区：如果消费者订阅的主题发生新增分区的情况，新增的分区需要被分配给消费者。

选择合适的再平衡策略取决于实际需求和场景，例如，轮询适用于简单的均衡场景，而范围和粘性策略适用于更复杂的场景，可以更好地保持负载均衡和避免不必要的再平衡。
本文主要会通过几个示例来对上面讲解的三种分区重分配策略的基本实现原理进行讲解。

​Range​ Assignor（范围分区策略）

• 配置

partition.assignment.strategy=org.apache.kafka.clients.consumer.RangeAssignor
1

Range Assignor 是Kafka中的再均衡衡策略之一，它以单个topic为一个维度来计算分区分配给消费者。这个策略的核心思想是按照字母顺序对消费者和主题的分区进行排序，并确保每个消费者平均获得尽可能多的主题分区。

具体的分配过程如下：

1. 对消费组中的所有消费者按字母排序。

2. 对topic的分区按照分区号排序。

3. 计算每个消费者最少应分配的分区数，以及需要多分配一个分区的消费者数量。

4. 前几个消费者分配多一个分区，余下的按照最少分配数分配。

​
上图示例说明了这一过程，其中有一个主题（topic），四个分区（partition）​partition-0、​partition-1、​partition-2、​partition-3，以及三个消费者（consumer）​consumer-0、​consumer-1、​consumer-2。假设 N 为分区数，M 为消费者数量：

• A = N/M 为每个消费者最少的分区数，即A = 4/3 = 1。

• B = N%M 为需要多分配分区的消费者数量，即 B = 4%3 = 1。

最终的分配结果是：

• consumer-0：partition-0、partition-1
• consumer-1：partition-2
• consumer-2：partition-3

虽然Range策略在单个主题下表现均衡，但在处理多个主题时，可能导致消费者排序靠前的负载比排序靠后的负载多很多。这是Range策略的一个潜在弊端。在涉及多个主题的情况下，可以考虑其他再平衡策略以满足不同的负载均衡需求。

• 核心源码

// partitionsPerTopic表示topic和分区关系，key是topic，value是分区数量
// subscriptions表示订阅关系，key是消费者，value是订阅的topic
@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, Subscription> subscriptions) {
    // 得到topic和订阅的消费者集合信息，例如{t0:[c0, c1], t1:[C0, C1]}
    Map<String, List<String>> consumersPerTopic = consumersPerTopic(subscriptions);
    // 保存topic分区和订阅该topic的消费者关系结果map
    Map<String, List<TopicPartition>> assignment = new HashMap<>();
    for (String memberId : subscriptions.keySet())
        // memberId就是消费者client.id+uuid(kafka在client.id上追加的)
        assignment.put(memberId, new ArrayList<TopicPartition>());
 
    // 遍历每个topic和消费者集合信息组成的map（由这个遍历可知，range策略分配结果在各个topic之间互不影响）
    for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {
        // topic名称
        String topic = topicEntry.getKey();
        // topic的消费者集合信息
        List<String> consumersForTopic = topicEntry.getValue();
 
        // 当前topic的分区数量
        Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
        // 如果当天topic没有分区，那么继续遍历下一个topic
        if (numPartitionsForTopic == null)
            continue;
 
        // 消费者集合根据字典排序
        Collections.sort(consumersForTopic);
        // 每个topic分区数量除以消费者数量，得出每个消费者分配到的分区数量
        int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
        // 无法整除的剩余分区数量
        int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();
        // 根据topic名称和分区数量，得到分区集合信息
        List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);
        // 遍历订阅当前topic的消费者集合
        for (int i = 0, n = consumersForTopic.size(); i < n; i++) {
            // 分配到的分区的开始位置
            int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition);
            // 分配到的分区数量（整除分配到的分区数量，加上1个无法整除分配到的分区--如果有资格分配到这个分区的话。判断是否有资格分配到这个分区：如果整除后余数为m，那么排序后的消费者集合中前m个消费者都能分配到一个额外的分区）
            int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);
            // 给消费者分配分区
            assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length));
        }
    }
    return assignment;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

​Round Robin Assignor（轮询分区策略）

​RoundRobinAssignor是一种用于分区分配的策略，它针对所有的主题分区。其工作原理是将所有分区和所有消费者列举出来，进行排序，然后通过轮询的方式将每个分区分配给每个消费者。如果某个消费者没有订阅特定主题，那么就跳过该消费者，继续分配给下一个消费者。

上图展示了轮询分区的分配过程：

1. 有三个消费者consumer-0、consumer-1、consumer-2，以及三个主题topic-A、topic-B、topic-C。

2. 通过轮询分区策略，对所有partition和所有consumer进行排序。

3. 分配的顺序是：topic-A-p0、topic-A-p1、topic-A-p2，依此类推，直到topic-C-p2。

4. 最终的分配结果是：

• Consumer-0: topic-A-p0、topic-A-p1

• Consumer-1: topic-A-p2

• Consumer-2: topic-B-p0、topic-B-p1、topic-B-p2、topic-C-p0、topic-C-p1、topic-C-p2

上图所示为一个不一致订阅的例子：

1. Consumer-0订阅Topic-A、Topic-B。

2. Consumer-1订阅Topic-B、Topic-C。

分配过程：

• 第一轮：Consumer-0得到topic-A-p0，Consumer-1得到topic-B-p0。

• 第二轮：Consumer-0得到topic-A-p1，Consumer-1得到topic-B-p1。

• 第三轮：Consumer-0得到topic-A-p2，Consumer-1得到topic-B-p2。

• 第四、五、六轮：Consumer-0得到topic-B-p0、topic-B-p1、topic-B-p2，Consumer-1得到topic-C-p0、topic-C-p1、topic-C-p2。

可以看到，由于消费者的订阅关系不一致，导致Consumer-1多消费了3个分区。因此，在Consumer Group中最好保持一致的订阅关系，以获得更均匀的分区分配。​

这种策略的优点在于简单且公平，每个消费者都有机会获取到每个主题的分区。然而，在实际应用中，如果消费者的订阅关系不一致，可能导致分配不够均匀。因此，在使用RoundRobinAssignor时，最好确保Consumer Group中的所有消费者具有一致的订阅关系，以获得更好的分区均衡效果。

• 核心代码

// partitionsPerTopic表示topic和分区关系，key是topic，value是分区数量
// subscriptions表示订阅关系，key是消费者，value是订阅的topic信息
@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, Subscription> subscriptions) {
    Map<String, List<TopicPartition>> assignment = new HashMap<>();
    for (String memberId : subscriptions.keySet())
        assignment.put(memberId, new ArrayList<TopicPartition>());
 
    // 将消费者集合先按照字典排序，再构造成一个环形迭代器
    CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet()));
    // 以topic名称排序（SortedSet<String> topics = new TreeSet<>();TreeSet保存topic名称从而实现排序），遍历topic下的分区，得到全部分区（分区主要信息包括topic名称和分区编号）
    for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) {
        final String topic = partition.topic();
        // assigner.peek()得到最后一次遍历的消费者。如果遍历的当前分区所属topic不在最后一次遍历的消费者订阅的topic范围内，那么从环形迭代器中轮询选择下一个消费者，直到选择的消费者订阅的topic集合包含当前topic。
        while (!subscriptions.get(assigner.peek()).topics().contains(topic))
            assigner.next();
        // 给消费者分配分区，并轮询到下一个消费者
        assignment.get(assigner.next()).add(partition);
    }
    return assignment;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

​ Sticky（粘性分区策略）

什么是粘性分区？

​Sticky分区策略的原理在于分配的结果具有"粘性"，即在进行新一轮分配之前，考虑上一次分配的结果，尽量减少分配的变动，以节省开销。这种策略在Kafka 0.11.x版本中引入，旨在实现以下目标：

1. 均衡性： 粘性分区策略首先会尽量均衡地分配分区给消费者。每个消费者按顺序获取一个或多个分区，以确保负载尽可能平均分布。

2. 保持稳定： 一旦分区被分配给某个消费者，该分区尽量保持在同一消费者上。这就是"Sticky"的含义，它试图将分区“粘”在已经处理它的消费者上。

3. 减小重新平衡频率： 粘性分区策略通过尽量减小重新平衡的频率来提高消费者组的稳定性。重新平衡可能由消费者加入或退出、心跳超时等触发，但策略会努力保持已分配的分区不变，只分配新增的分区。

4. 适应消费者变动： 在同一消费者组内，如果某个消费者出现问题或者新的消费者加入，粘性分区策略尽量保持原有分配的分区不发生变化，从而减小整体系统的不稳定性。

Sticky分区策略的核心思想是通过在分配时考虑上一次的分配结果，最大程度地保持分区的稳定性，减少重新平衡的开销，提高整个消费者组的可用性和性能。然而，在某些场景下，需要权衡稳定性和负载均衡，具体选择何种策略取决于系统的特定需求。

粘性分区基本原理

由于粘性分区的实现原理比较复杂，对于开发者来说去死抠细节没有意义，这里简单总结下粘性分区的原理，​Sticky分区分配策略的原理可以总结如下：

1. 初始化分配： 在初始状态下，Kafka将分区均匀分配给消费者。每个消费者按照顺序获取一个或多个分区，以确保负载尽可能平均地分布。

2. 粘性分区分配： 一旦分区被分配给某个消费者，该分区将尽量保持分配给同一消费者。这种"粘性"的特性是指策略尝试将分区保持在已经处理它的消费者上，以降低重新平衡的频率。

3. 分区重新平衡： 当有新的消费者加入消费者组、有消费者退出，或者某个分区失去联系时，系统需要执行分区再分配。然而，Sticky策略会尽量减小重新平衡的频率。它不会立即重新分配所有分区，而是尽量保持已分配的分区不变，并仅分配新增的分区。

4. 重新平衡触发： 重新平衡可以由消费者加入或退出消费者组、消费者心跳超时，以及某个分区失去联系等情况触发。

5. 维持粘性： Sticky策略致力于保持分区分配的稳定性，减少分区再分配的次数，从而减小整个消费者组的不稳定性。这有助于提高系统的可用性和性能。

Sticky分区策略的主要优点在于减少分区再分配的频率，降低系统的不稳定性，以及减轻重新平衡的成本。然而，在某些情况下，可能需要更频繁的重新平衡以确保负载分配的公平性。因此，在选择分区分配策略时，需要根据具体的使用情况和需求进行权衡。

​总结

​本文介绍了Kafka的再均衡机制，即在消费者组内的消费者数量变化或订阅的主题发生变化时，如何重新分配消费者与分区的关系。Kafka提供了三种再平衡策略：Round Robin（轮询）、Range（范围）和Sticky（粘性）。

1. Round Robin（轮询）： 以轮询的方式将所有分区依次分配给消费者，确保每个消费者都能均匀地获得分区。

2. Range（范围）： 先计算每个消费者可以消费的分区个数，然后按顺序将指定个数范围的分区分配给各个消费者，以均衡分配消费压力。

3. Sticky（粘性）： 在保持均衡的基础上，尽可能保持未宕机的消费者继续消费其之前负责的分区，减少不必要的再平衡，提高消费者组的稳定性。

再平衡可能在新增或删除消费者、消费者订阅主题变化、主题新增分区等情况下触发。选择合适的再平衡策略取决于实际需求和场景。文章通过Range Assignor、Round Robin Assignor和Sticky策略的实现原理进行了详细介绍, 在实际应用中，需要根据系统需求权衡分区分配策略的选择。