Kafka 重平衡 全流程解析
前言
本文来自 极客时间 Kafka核心技术与实战
这段时间有看 极客时间的这个课程,
这里仅以分享的角度来做个笔记。
那么本文将涉及到以下几个知识点：
- 重平衡是什么？为什么要了解他？
- 发生重平衡的时机
- Kafka的心跳机制 与 Rebalance
- 消费者组状态切换
- 重平衡全流程解析
重平衡是什么？为什么要了解他？
-
重平衡是什么
Rebalance（重平衡 ）本质上是一种协议，
规定了一个Consumer Group下的所有 Consumer 如何达成一致，
来分配订阅Topic的每个分区。说简单点就是 给消费组每个消费者分配消费任务的过程。
-
为什么要了解他？
Rebalance是启动一个消费者组必经的过程，
当然这不是最主要的，最主要的是，在消费的过程中，
在某些情况下会导致这个过程再次发生，
带来的后果就是整个集群暂时性的瘫痪，
严重影响到Kafka的高可用
发生重平衡的时机
那么 Rebalance 会在什么时候发生呢？
-
订阅主题数发生变化
这种情况一般不会发生，
如果发生，那也是因为我们的业务调整才会，
所以这种基本要么不发生要么就是不可避免。 -
主题分区发生变化
这种情况发生会相对多一点，但是也有限，
在部署Kafka集群前，
我们就需要考虑到该集群的容量，
以便来确定好分区数。
虽然不一定一步到位，
但是调整的次数应该是极其有限的，
一般也可以选择在半夜低峰的时候进行调整，影响不大。 -
消费端的消费者组成员变化
基本上影响最大的就是这个原因了，
为什么这么说呢？
我们考虑下什么时候消费者组的成员会发生变化就能大概了解了。- 消费者处理消息超时，
即如果消费者处理消费的消息的时间超过了
Kafka集群配置的max.poll.interval.ms的值，
那么该消费者将会自动离组 - 心跳超时，
如果消费者在指定的session.timeout.ms时间内没有汇报心跳，
那么Kafka就会认为该消费已经dead了
- 消费者处理消息超时，
可以看出，消费端的消费者组成员变化一般都是由于异常引起的，
所以其产生的 Rebalance 也是最难控制的。
Kafka的心跳机制 与 Rebalance
Kafka的心跳机制 与 Rebalance 有什么关系呢？
事实上，重平衡过程是靠消费者端的心跳线程（Heartbeat Thread）通知到其他消费者实例的
每当消费者向其 coordinator 汇报心跳的时候，
如果这个时候 coordinator 决定开启 Rebalance ，
那么 coordinator 会将REBALANCE_IN_PROGRESS封装到心跳的响应中，
当消费者接受到这个REBALANCE_IN_PROGRESS，
他就知道需要开启新的一轮 Rebalance 了,
所以heartbeat.interval.ms除了是设置心跳的间隔时间，
其实也意味着 Rebalance 感知速度，
心跳越快，那么 Rebalance 就能更快的被各个消费者感知。
在 Kafka 0.10.1.0 版本之前，
发送心跳请求是在消费者主线程完成的，
也就是你写代码调用KafkaConsumer.poll方法的那个线程。
这样做有诸多弊病，最大的问题在于，消息处理逻辑也是在这个线程中完成的。
因此，一旦消息处理消耗了过长的时间，
心跳请求将无法及时发到协调者那里，
导致协调者“错误地”认为该消费者已“死”。
自 0.10.1.0 版本开始，
社区引入了一个单独的心跳线程来专门执行心跳请求发送，避免了这个问题。
消费者组状态切换
为什么要了解 消费者组状态 呢？
这里主要是为了方便讲解 Rebalance 流程，
所以你需要大概了解一下消费者组的状态切换，
如下图
消费者流转状态.jpg
其流转过程大概如下：
消费者状态流程过程.jpg
一个消费者组最开始是Empty状态，
当重平衡过程开启后，
它会被置于PreparingRebalance状态 等待成员加入，
成员都加入之后变更到CompletingRebalance状态等待分配方案，
当coordinator分配完个消费者消费的分区后，
最后就流转到Stable状态完成重平衡。
当有新成员加入或已有成员退出时，
消费者组的状态 从Stable直接跳到PreparingRebalance状态，
此时，所有现存成员就必须重新申请加入组。
当所有成员都退出组后，消费者组状态变更为Empty。
Kafka定期自动删除过期位移的条件就是，组要处于Empty状态。
因此，如果你的消费者组停掉了很长时间（超过7天），
那么Kafka很可能就把该组的位移数据删除了。
消费者端重平衡流程
在消费者端，重平衡分为两个步骤：
-
加入组。
当组内成员加入组时，它会向 coordinator 发送JoinGroup请求。
在该请求中，每个成员都要将自己订阅的主题上报，
这样协调者就能收集到所有成员的订阅信息。
一旦收集了全部成员的JoinGroup请求后，
Coordinator 会从这些成员中选择一个担任这个消费者组的领导者。通常情况下，第一个发送JoinGroup请求的成员自动成为领导者。
领导者消费者的任务是收集所有成员的订阅信息，
然后根据这些信息，制定具体的分区消费分配方案。
特别注意的是：这里说的是消费者领导者。选出领导者之后，
Coordinator 会把消费者组订阅信息封装进JoinGroup请求的 响应体中，
然后发给领导者，由领导者统一做出分配方案后，
进入到下一步：发送SyncGroup请求。
如下图就是 JoinGroup 的全过程[图片上传中...(25-消费者组重平衡全流程解析.jpg-d67470-1567669412412-0)]
JoinGroup 流程解析.jpg
-
领导者消费者（Leader Consumer）分配方案。
领导者向 Coordinator 发送SyncGroup请求，
将刚刚做出的分配方案发给协调者。
值得注意的是，其他成员也会向 Coordinator 发送SyncGroup请求，
只不过请求体中并没有实际的内容。
这一步的主要目的是让 Coordinator 接收分配方案，
然后统一以 SyncGroup 响应的方式分发给所有成员，
这样组内所有成员就都知道自己该消费哪些分区了。如下图：
SyncGroup全流程解析.jpg
消费者端重平衡流程 大概就这样了，下面我们再来看看：Broker端重平衡
Broker端重平衡
要剖析协调者端处理重平衡的全流程，
我们必须要分几个场景来讨论。
这几个场景分别是
- 新成员加入组
- 组成员主动离组
- 组成员崩溃离组
- 组成员提交位移。
接下来，我们一个一个来讨论。
-
新成员入组。
新成员入组是指组处于Stable状态后，有新成员加入。
如果是全新启动一个消费者组，Kafka是有一些自己的小优化的，流程上会有些许的不同。
我们这里讨论的是，组稳定了之后有新成员加入的情形。
当协调者收到新的JoinGroup请求后，
它会通过心跳请求响应的方式通知组内现有的所有成员，
强制它们开启新一轮的重平衡。
具体的过程和之前的客户端重平衡流程是一样的。
现在，我用一张时序图来说明协调者一端是如何处理新成员入组的。
新成员入组场景.jpg
-
组成员主动离组。
何谓主动离组？就是指消费者实例所在线程或进程调用close()方法主动通知协调者它要退出。
这个场景就涉及到了第三类请求：LeaveGroup请求。
协调者收到LeaveGroup请求后，依然会以心跳响应的方式通知其他成员，
因此我就不再赘述了，还是直接用一张图来说明。
组成员主动离组场景.jpg
-
组成员崩溃离组。
崩溃离组是指消费者实例出现严重故障，突然宕机导致的离组。
它和主动离组是有区别的，
因为后者是主动发起的离组，协调者能马上感知并处理。
但崩溃离组是被动的，协调者通常需要等待一段时间才能感知到，
这段时间一般是由消费者端参数session.timeout.ms控制的。
也就是说，Kafka一般不会超过session.timeout.ms就能感知到这个崩溃。
当然，后面处理崩溃离组的流程与之前是一样的，我们来看看下面这张图。
组成员崩溃离组场景.jpg
-
重平衡时协调者对组内成员提交位移的处理。
正常情况下，每个组内成员都会定期汇报位移给协调者。
当重平衡开启时，协调者会给予成员一段缓冲时间，
要求每个成员必须在这段时间内快速地上报自己的位移信息，
然后再开启正常的JoinGroup/SyncGroup请求发送。
还是老办法，我们使用一张图来说明。
组内成员提交位移场景.jpg
总结：
其实不论哪种方式，都是差不多的流程，这里放开举例，最主要的还是为了更加清晰，如果发生类似的问题，可以很快的从上面这些可能入手。
基本流程就是 Coordinator 感知到 消费者组的变化，
然后在心跳的过程中发送重平衡信号通知各个消费者离组，
然后消费者重新以 JoinGroup 方式加入 Coordinator，并选出Consumer Leader。
当所有消费者加入 Coordinator，
Consumer Leader会根据 Coordinator给予的分区信息给出分区方案。
Coordinator 将该方案以 SyncGroup 的方式将该方案执行下去，通知各个消费者，
这样就完成了一轮 重平衡了。
