Kafka中的再均衡

在《Kafka消费者的使用和原理》中已经提到过“再均衡”的概念，咱们先回顾下，一个主题能够有多个分区，而订阅该主题的消费组中能够有多个消费者。每个分区只能被消费组中的一个消费者消费，可认为每一个分区的消费权只属于消费组中的一个消费者。可是世界是变化的，例如消费者会宕机，还有新的消费者会加入，而为了应对这些变化，让分区所属权的分配合理，这都须要对分区所属权进行调整，也就是所谓的“再均衡”。本文将对再均衡的相关知识进行详细叙述。

触发时机

首先，咱们须要了解什么状况下会触发再均衡，在前文已经提到了消费者数量的变化，是须要再均衡的。在使用Kafka时，除了消费者数量可能会变化，分区数量也一样可能变化，咱们能够人为的对分区数量进行修改，可是Kafka只容许增长分区，因此咱们只能把分区数量调大，不能调小，不然会收到InvalidPartitionException异常。关于为何不能减小分区，可参考下面的回答：

按Kafka现有的代码逻辑，此功能是彻底能够实现的，不过也会使得代码的复杂度急剧增大。实现此功能须要考虑的因素不少，好比删除掉的分区中的消息该做何处理？若是随着分区一块儿消失则消息的可靠性得不到保障；若是须要保留则又须要考虑如何保留。直接存储到现有分区的尾部，消息的时间戳就不会递增，如此对于Spark、Flink这类须要消息时间戳（事件时间）的组件将会受到影响；若是分散插入到现有的分区中，那么在消息量很大的时候，内部的数据复制会占用很大的资源，并且在复制期间，此主题的可用性又如何获得保障？与此同时，顺序性问题、事务性问题、以及分区和副本的状态机切换问题都是不得不面对的。反观这个功能的收益点倒是很低，若是真的须要实现此类的功能，彻底能够从新建立一个分区数较小的主题，而后将现有主题中的消息按照既定的逻辑复制过去便可。

简单来讲，就是作这个功能须要考虑不少因素，这样会把代码弄的很复杂，而收益却很低，并且存在替代方案来实现该效果，建立一个分区数小的主题，再把当前主题迁移过去。

除了消费者、分区数量的变化，还有一种状况，也须要进行再均衡。当消费者订阅主题时使用的是正则表达式，例如“test.*”，表示订阅全部以test开头的主题，当有新的以test开头的主题被建立时，则须要经过再均衡将该主题的分区分配给消费者。

再均衡的三种触发时机，咱们已经清楚了，下面咱们看下再均衡是如何实现的。

协调者

再均衡，将分区所属权分配给消费者。所以须要和全部消费者通讯，这就须要引进一个协调者的概念，由协调者为消费组服务，为消费者们作好协调工做。在Kafka中，每一台Broker上都有一个协调者组件，负责组成员管理、再均衡和提交位移管理等工做。若是有N台Broker，那就有N个协调者组件，而一个消费组只需一个协调者进行服务，那该由哪一个Broker为其服务？肯定Broker须要两步：

1. 计算分区号
   partition = Math.abs(groupID.hashCode() % offsetsTopicPartitionCount)

根据groupID的哈希值，取余offsetsTopicPartitionCount（内部主题consumer_offsets的分区数，默认50）的绝对值，其意思就是把消费组哈希散列到内部主题__consumer_offsets的一个分区上。肯定协调者为何要和内部主题扯上关系。这就跟协调者的做用有关了。协调者不只是负责组成员管理和再均衡，在协调者中还须要负责处理消费者的偏移量提交，而偏移量提交则正是提交到consumer_offsets的一个分区上。因此这里须要取余offsetsTopicPartitionCount来肯定偏移量提交的分区。

1. 找出分区Leader副本所在的Broker

肯定了分区就简单了，分区Leader副本所在的Broker上的协调者，就是咱们要找的。

这个算法一般用于帮助定位问题。当一个消费组出现问题时，咱们能够先肯定协调者的Broker，而后查看Broker端的日志来定位问题。

交互方式

协调者，咱们肯定了。那协调者和消费者之间是如何交互的？协调者如何掌握消费者的状态，又如何通知再均衡。这里使用了心跳机制。在消费者端有一个专门的心跳线程负责以heartbeat.interval.ms的间隔频率发送心跳给协调者，告诉协调者本身还活着。同时协调者会返回一个响应。而当须要开始再均衡时，协调者则会在响应中加入REBALANCE_IN_PROGRESS，当消费者收到响应时，便能知道再均衡要开始了。

因为再平衡的开始依赖于心跳的响应，因此heartbeat.interval.ms除了决定心跳的频率，也决定了再均衡的通知频率。
如今咱们再从新看下，触发再均衡的时机，前面说到有三种状况触发再均衡，分别是消费者数量的增长或减小、分区数的增长和新建立主题，其中消费者数量增长、分区数增长和新建立主题，这都是必须是人为操做，算是计划内的再均衡。而消费者数量减小则除了是人为操做，也可能由于其余缘由致使，属于计划以外的再均衡，这是咱们须要关心的，毕竟再均衡的开销仍是很大的，全部消费者都会中止工做，因此咱们应尽可能避免没必要要的再均衡。下面咱们看下影响消费者数量减小的参数有哪些：

1. session.timeout.ms：Broker端参数，消费者的存活时间，默认10秒，若是在这段时间内，协调者没收到任何心跳，则认为该消费者已崩溃离组；
2. heartbeat.interval.ms：消费者端参数，发送心跳的频率，默认3秒；
3. max.poll.interval.ms：消费者端参数，两次调用poll的最大时间间隔，默认5分钟，若是5分钟内没法消费完，则会主动离组。

能够看出session.timeout.ms和heartbeat.interval.ms是相关的，这里给出一个建议参考的公式：

session.timeout.ms ≥ 3 * heartbeat.interval.ms

为尽可能避免由于偶发的网络缘由，心跳没法到达协调者，在超时以前，应至少能发送3轮心跳。再给出一个经验值的设置：session.timeout.ms=6s，heartbeat.interval.ms=2s。

max.poll.interval.ms的设置，则主要和下游处理时间有关，例以下游处理时间须要6分钟，那按默认值是不合理的，消费者会频繁主动离组。因此须要把值设置的比下游处理时间大一点，避免没必要要的再均衡。

这一小节主要讲了协调者如何通知消费者开始再均衡，以及如何设置参数避免没必要要的再均衡，下面咱们看下再均衡的流程是怎么样的。

流程

1. 当消费者收到协调者的再均衡开始通知时，须要当即提交偏移量；
2. 消费者在收到提交偏移量成功的响应后，再发送JoinGroup请求，从新申请加入组，请求中会含有订阅的主题信息；
3. 当协调者收到第一个JoinGroup请求时，会把发出请求的消费者指定为Leader消费者，同时等待rebalance.timeout.ms，在收集其余消费者的JoinGroup请求中的订阅信息后，将订阅信息放在JoinGroup响应中发送给Leader消费者，并告知他成为了Leader，同时也会发送成功入组的JoinGroup响应给其余消费者；
4. Leader消费者收到JoinGroup响应后，根据消费者的订阅信息制定分配方案，把方案放在SyncGroup请求中，发送给协调者。普通消费者在收到响应后，则直接发送SyncGroup请求，等待Leader的分配方案；
5. 协调者收到分配方案后，再经过SyncGroup响应把分配方案发给全部消费组。
6. 当全部消费者收到分配方案后，就意味着再均衡的结束，能够正常开始消费工做了。