Kafka Rebalance机制分析

以你之姓@ 2021-06-10 20:38 274阅读 0赞

### **什么是Rebalance** ###

Rebalance 本质上是一种协议，规定了一个 Consumer Group 下的所有 consumer 如何达成一致，来分配订阅 Topic 的每个分区。

例如：某 Group 下有 20 个 consumer 实例，它订阅了一个具有 100 个 partition 的 Topic 。正常情况下，kafka 会为每个 Consumer 平均的分配 5 个分区。这个分配的过程就是 Rebalance。

### **触发 Rebalance 的时机** ###

Rebalance 的触发条件有3个。

*  组成员个数发生变化。例如有新的 consumer 实例加入该消费组或者离开组。
 *  订阅的 Topic 个数发生变化。
 *  订阅 Topic 的分区数发生变化。

Rebalance 发生时，Group 下所有 consumer 实例都会协调在一起共同参与，kafka 能够保证尽量达到最公平的分配。但是 Rebalance 过程对 consumer group 会造成比较严重的影响。在 Rebalance 的过程中 consumer group 下的所有消费者实例都会停止工作，等待 Rebalance 过程完成。

### **Rebalance 过程分析** ###

Rebalance 过程分为两步：Join 和 Sync。

1.  Join 顾名思义就是加入组。这一步中，所有成员都向coordinator发送JoinGroup请求，请求加入消费组。一旦所有成员都发送了JoinGroup请求，coordinator会从中选择一个consumer担任leader的角色，并把组成员信息以及订阅信息发给leader——注意leader和coordinator不是一个概念。leader负责消费分配方案的制定。

[![img][]][img 1]

1.  Sync，这一步leader开始分配消费方案，即哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案封装进SyncGroup请求中发给coordinator，非leader也会发SyncGroup请求，只是内容为空。coordinator接收到分配方案之后会把方案塞进SyncGroup的response中发给各个consumer。这样组内的所有成员就都知道自己应该消费哪些分区了。

[![img][img 2]][img_img 2]

### **Rebalance 场景分析** ###

**新成员加入组**

[![img][img 3]][img_img 3]

**组成员“崩溃”**

**组成员崩溃和组成员主动离开是两个不同的场景。**因为在崩溃时成员并不会主动地告知coordinator此事，coordinator有可能需要一个完整的session.timeout周期(心跳周期)才能检测到这种崩溃，这必然会造成consumer的滞后。可以说离开组是主动地发起rebalance；而崩溃则是被动地发起rebalance。

[![img][img 4]][img_img 4]

**组成员主动离开组**

[![img][img 5]][img_img 5]

**提交位移**

[![img][img 6]][img_img 6]

### **如何避免不必要的rebalance** ###

要避免 Rebalance，还是要从 Rebalance 发生的时机入手。我们在前面说过，Rebalance 发生的时机有三个：

*  组成员数量发生变化
 *  订阅主题数量发生变化
 *  订阅主题的分区数发生变化

后两个我们大可以人为的避免，发生rebalance最常见的原因是消费组成员的变化。

消费者成员正常的添加和停掉导致rebalance，这种情况无法避免，但是时在某些情况下，Consumer 实例会被 Coordinator 错误地认为 “已停止” 从而被“踢出”Group。从而导致rebalance。

当 Consumer Group 完成 Rebalance 之后，每个 Consumer 实例都会定期地向 Coordinator 发送心跳请求，表明它还存活着。如果某个 Consumer 实例不能及时地发送这些心跳请求，Coordinator 就会认为该 Consumer 已经 “死” 了，从而将其从 Group 中移除，然后开启新一轮 Rebalance。这个时间可以通过Consumer 端的参数 session.timeout.ms进行配置。默认值是 10 秒。

除了这个参数，Consumer 还提供了一个控制发送心跳请求频率的参数，就是 heartbeat.interval.ms。这个值设置得越小，Consumer 实例发送心跳请求的频率就越高。频繁地发送心跳请求会额外消耗带宽资源，但好处是能够更加快速地知晓当前是否开启 Rebalance，因为，目前 Coordinator 通知各个 Consumer 实例开启 Rebalance 的方法，就是将 REBALANCE\_NEEDED 标志封装进心跳请求的响应体中。

除了以上两个参数，Consumer 端还有一个参数，用于控制 Consumer 实际消费能力对 Rebalance 的影响，即 max.poll.interval.ms 参数。它限定了 Consumer 端应用程序两次调用 poll 方法的最大时间间隔。它的默认值是 5 分钟，表示你的 Consumer 程序如果在 5 分钟之内无法消费完 poll 方法返回的消息，那么 Consumer 会主动发起 “离开组” 的请求，Coordinator 也会开启新一轮 Rebalance。

通过上面的分析，我们可以看一下那些rebalance是可以避免的：

**第一类非必要 Rebalance 是因为未能及时发送心跳，导致 Consumer 被 “踢出”Group 而引发的**。这种情况下我们可以设置 **session.timeout.ms 和 heartbeat.interval.ms** 的值，来尽量避免rebalance的出现。（**以下的配置是在网上找到的最佳实践，暂时还没测试过**）

*  设置 session.timeout.ms = 6s。
 *  设置 heartbeat.interval.ms = 2s。
 *  要保证 Consumer 实例在被判定为 “dead” 之前，能够发送至少 3 轮的心跳请求，即 session.timeout.ms >= 3 \* heartbeat.interval.ms。

将 session.timeout.ms 设置成 6s 主要是为了让 Coordinator 能够更快地定位已经挂掉的 Consumer，早日把它们踢出 Group。

**第二类非必要 Rebalance 是 Consumer 消费时间过长导致的**。此时，**max.poll.interval.ms** 参数值的设置显得尤为关键。如果要避免非预期的 Rebalance，你最好将该参数值设置得大一点，比你的下游最大处理时间稍长一点。

总之，要为业务处理逻辑留下充足的时间。这样，Consumer 就不会因为处理这些消息的时间太长而引发 Rebalance 。

### **相关概念** ###

**coordinator**

Group Coordinator是一个服务，每个Broker在启动的时候都会启动一个该服务。Group Coordinator的作用是用来存储Group的相关Meta信息，并将对应Partition的Offset信息记录到Kafka内置Topic(\_\_consumer\_offsets)中。Kafka在0.9之前是基于Zookeeper来存储Partition的Offset信息(consumers/\{group\}/offsets/\{topic\}/\{partition\})，因为ZK并不适用于频繁的写操作，所以在0.9之后通过内置Topic的方式来记录对应Partition的Offset。

每个Group都会选择一个Coordinator来完成自己组内各Partition的Offset信息，选择的规则如下：

*  1，计算Group对应在\_\_consumer\_offsets上的Partition
 *  2，根据对应的Partition寻找该Partition的leader所对应的Broker，该Broker上的Group Coordinator即就是该Group的Coordinator

Partition计算规则：

`partition-Id(__consumer_offsets) = Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount)`

其中groupMetadataTopicPartitionCount对应offsets.topic.num.partitions参数值，默认值是50个分区

### **一次Rebalance所耗时间** ###

**测试环境**

1个Topic，10个partition，3个consumer

在本地环境进行测试

**测试结果**

经过几轮测试发现每次rebalance所消耗的时间大概在 **80ms~100ms**平均耗时在**87ms**左右。

[img]: /images/20210527/53863bb8b06542fb8deffeacc3bcf478.png
[img 1]: https://upload-images.jianshu.io/upload_images/3951014-30b27a129a67480a.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/870/format/webp
[img 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zOTUxMDE0LTYzZjhmY2ZiZjg5NGUxN2UucG5n?x-oss-process=image/format,png
[img_img 2]: /images/20210527/4d2d1b9e6d85424497d06abc375f0ed0.png
[img 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zOTUxMDE0LTUyMTI3NWUwNDA3MThmZTMucG5n?x-oss-process=image/format,png
[img_img 3]: /images/20210527/3ec78fdd3a0b40bb872c0edf413a3dac.png
[img 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zOTUxMDE0LWEzOTY1ZWIyNmEyZjIzNTkucG5n?x-oss-process=image/format,png
[img_img 4]: /images/20210527/ae32226d0cda44d6af63ba411dce753e.png
[img 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zOTUxMDE0LTgyZWFkODIxZDQxMjI0ZTYucG5n?x-oss-process=image/format,png
[img_img 5]: /images/20210527/1a0a6fe24b4949d9abddb6d65e0187e4.png
[img 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zOTUxMDE0LTkxZWQyNTE3NWEyYjhkOTcucG5n?x-oss-process=image/format,png
[img_img 6]: /images/20210527/2cd062ab537f4fd68c80ac2ae41b8000.png