25 | 消费者组重平衡全流程解析

Jul 30, 2019

25 | 消费者组重平衡全流程解析-极客时间



下载APP





关闭

讲堂部落提薪训练营云原生训练营架构实战营企业版极客商城兑换中心 App下载浏览器插件

渠道合作

推荐作者

25 | 消费者组重平衡全流程解析

2019-07-30 胡夕来自北京

《Kafka核心技术与实战》

课程介绍



讲述：胡夕

时长13:57大小12.77M



你好，我是胡夕。今天我要和你分享的主题是：消费者组重平衡全流程解析。
之前我们聊到过消费者组的重平衡流程，它的作用是让组内所有的消费者实例就消费哪些主题分区达成一致。重平衡需要借助 Kafka Broker 端的 Coordinator 组件，在 Coordinator 的帮助下完成整个消费者组的分区重分配。今天我们就来详细说说这个流程。
先提示一下，我会以 Kafka 2.3 版本的源代码开启今天的讲述。在分享的过程中，对于旧版本的设计差异，我也会显式地说明。这样，即使你依然在使用比较旧的版本也不打紧，毕竟设计原理大体上是没有变化的。
触发与通知我们先来简单回顾一下重平衡的 3 个触发条件：
组成员数量发生变化。
订阅主题数量发生变化。
订阅主题的分区数发生变化。
就我个人的经验来看，在实际生产环境中，因命中第 1 个条件而引发的重平衡是最常见的。另外，消费者组中的消费者实例依次启动也属于第 1 种情况，也就是说，每次消费者组启动时，必然会触发重平衡过程。
这部分内容我在专栏第 15 讲中已经详细介绍过了，就不再赘述了。如果你不记得的话，可以先去复习一下。
今天，我真正想引出的是另一个话题：重平衡过程是如何通知到其他消费者实例的？答案就是，靠消费者端的心跳线程（Heartbeat Thread）。
Kafka Java 消费者需要定期地发送心跳请求（Heartbeat Request）到 Broker 端的协调者，以表明它还存活着。在 Kafka 0.10.1.0 版本之前，发送心跳请求是在消费者主线程完成的，也就是你写代码调用 KafkaConsumer.poll 方法的那个线程。
这样做有诸多弊病，最大的问题在于，消息处理逻辑也是在这个线程中完成的。因此，一旦消息处理消耗了过长的时间，心跳请求将无法及时发到协调者那里，导致协调者“错误地”认为该消费者已“死”。自 0.10.1.0 版本开始，社区引入了一个单独的心跳线程来专门执行心跳请求发送，避免了这个问题。
但这和重平衡又有什么关系呢？其实，重平衡的通知机制正是通过心跳线程来完成的。当协调者决定开启新一轮重平衡后，它会将“REBALANCE_IN_PROGRESS”封装进心跳请求的响应中，发还给消费者实例。当消费者实例发现心跳响应中包含了“REBALANCE_IN_PROGRESS”，就能立马知道重平衡又开始了，这就是重平衡的通知机制。
对了，很多人还搞不清楚消费者端参数 heartbeat.interval.ms 的真实用途，我来解释一下。从字面上看，它就是设置了心跳的间隔时间，但这个参数的真正作用是控制重平衡通知的频率。如果你想要消费者实例更迅速地得到通知，那么就可以给这个参数设置一个非常小的值，这样消费者就能更快地感知到重平衡已经开启了。
消费者组状态机重平衡一旦开启，Broker 端的协调者组件就要开始忙了，主要涉及到控制消费者组的状态流转。当前，Kafka 设计了一套消费者组状态机（State Machine），来帮助协调者完成整个重平衡流程。严格来说，这套状态机属于非常底层的设计，Kafka 官网上压根就没有提到过，但你最好还是了解一下，因为它能够帮助你搞懂消费者组的设计原理，比如消费者组的过期位移（Expired Offsets）删除等。
目前，Kafka 为消费者组定义了 5 种状态，它们分别是：Empty、Dead、PreparingRebalance、CompletingRebalance 和 Stable。那么，这 5 种状态的含义是什么呢？我们一起来看看下面这张表格。
了解了这些状态的含义之后，我们来看一张图片，它展示了状态机的各个状态流转。
我来解释一下消费者组启动时的状态流转过程。一个消费者组最开始是 Empty 状态，当重平衡过程开启后，它会被置于 PreparingRebalance 状态等待成员加入，之后变更到 CompletingRebalance 状态等待分配方案，最后流转到 Stable 状态完成重平衡。
当有新成员加入或已有成员退出时，消费者组的状态从 Stable 直接跳到 PreparingRebalance 状态，此时，所有现存成员就必须重新申请加入组。当所有成员都退出组后，消费者组状态变更为 Empty。Kafka 定期自动删除过期位移的条件就是，组要处于 Empty 状态。因此，如果你的消费者组停掉了很长时间（超过 7 天），那么 Kafka 很可能就把该组的位移数据删除了。我相信，你在 Kafka 的日志中一定经常看到下面这个输出：
Removed ✘✘✘ expired offsets in ✘✘✘ milliseconds.
这就是 Kafka 在尝试定期删除过期位移。现在你知道了，只有 Empty 状态下的组，才会执行过期位移删除的操作。
消费者端重平衡流程有了上面的内容作铺垫，我们就可以开始介绍重平衡流程了。重平衡的完整流程需要消费者端和协调者组件共同参与才能完成。我们先从消费者的视角来审视一下重平衡的流程。
在消费者端，重平衡分为两个步骤：分别是加入组和等待领导者消费者（Leader Consumer）分配方案。这两个步骤分别对应两类特定的请求：JoinGroup 请求和 SyncGroup 请求。
当组内成员加入组时，它会向协调者发送 JoinGroup 请求。在该请求中，每个成员都要将自己订阅的主题上报，这样协调者就能收集到所有成员的订阅信息。一旦收集了全部成员的 JoinGroup 请求后，协调者会从这些成员中选择一个担任这个消费者组的领导者。
通常情况下，第一个发送 JoinGroup 请求的成员自动成为领导者。你一定要注意区分这里的领导者和之前我们介绍的领导者副本，它们不是一个概念。这里的领导者是具体的消费者实例，它既不是副本，也不是协调者。领导者消费者的任务是收集所有成员的订阅信息，然后根据这些信息，制定具体的分区消费分配方案。
选出领导者之后，协调者会把消费者组订阅信息封装进 JoinGroup 请求的响应体中，然后发给领导者，由领导者统一做出分配方案后，进入到下一步：发送 SyncGroup 请求。
在这一步中，领导者向协调者发送 SyncGroup 请求，将刚刚做出的分配方案发给协调者。值得注意的是，其他成员也会向协调者发送 SyncGroup 请求，只不过请求体中并没有实际的内容。这一步的主要目的是让协调者接收分配方案，然后统一以 SyncGroup 响应的方式分发给所有成员，这样组内所有成员就都知道自己该消费哪些分区了。
接下来，我用一张图来形象地说明一下 JoinGroup 请求的处理过程。
就像前面说的，JoinGroup 请求的主要作用是将组成员订阅信息发送给领导者消费者，待领导者制定好分配方案后，重平衡流程进入到 SyncGroup 请求阶段。
下面这张图描述的是 SyncGroup 请求的处理流程。
SyncGroup 请求的主要目的，就是让协调者把领导者制定的分配方案下发给各个组内成员。当所有成员都成功接收到分配方案后，消费者组进入到 Stable 状态，即开始正常的消费工作。
讲完这里，消费者端的重平衡流程我已经介绍完了。接下来，我们从协调者端来看一下重平衡是怎么执行的。
Broker 端重平衡场景剖析要剖析协调者端处理重平衡的全流程，我们必须要分几个场景来讨论。这几个场景分别是新成员加入组、组成员主动离组、组成员崩溃离组、组成员提交位移。接下来，我们一个一个来讨论。
场景一：新成员入组。
新成员入组是指组处于 Stable 状态后，有新成员加入。如果是全新启动一个消费者组，Kafka 是有一些自己的小优化的，流程上会有些许的不同。我们这里讨论的是，组稳定了之后有新成员加入的情形。
当协调者收到新的 JoinGroup 请求后，它会通过心跳请求响应的方式通知组内现有的所有成员，强制它们开启新一轮的重平衡。具体的过程和之前的客户端重平衡流程是一样的。现在，我用一张时序图来说明协调者一端是如何处理新成员入组的。
场景二：组成员主动离组。
何谓主动离组？就是指消费者实例所在线程或进程调用 close() 方法主动通知协调者它要退出。这个场景就涉及到了第三类请求：LeaveGroup 请求。协调者收到 LeaveGroup 请求后，依然会以心跳响应的方式通知其他成员，因此我就不再赘述了，还是直接用一张图来说明。
场景三：组成员崩溃离组。
崩溃离组是指消费者实例出现严重故障，突然宕机导致的离组。它和主动离组是有区别的，因为后者是主动发起的离组，协调者能马上感知并处理。但崩溃离组是被动的，协调者通常需要等待一段时间才能感知到，这段时间一般是由消费者端参数 session.timeout.ms 控制的。也就是说，Kafka 一般不会超过 session.timeout.ms 就能感知到这个崩溃。当然，后面处理崩溃离组的流程与之前是一样的，我们来看看下面这张图。
场景四：重平衡时协调者对组内成员提交位移的处理。
正常情况下，每个组内成员都会定期汇报位移给协调者。当重平衡开启时，协调者会给予成员一段缓冲时间，要求每个成员必须在这段时间内快速地上报自己的位移信息，然后再开启正常的 JoinGroup/SyncGroup 请求发送。还是老办法，我们使用一张图来说明。
小结好了，消费者重平衡流程我已经全部讲完了。虽然全程我都是拿两个成员来举例子，但你可以很容易地扩展到多个成员的消费者组，毕竟它们的原理是相同的。我希望你能多看几遍今天的内容，彻底掌握 Kafka 的消费者重平衡流程。社区正在对目前的重平衡流程做较大程度的改动，如果你不了解这些基础的设计原理，后面想深入学习这部分内容的话，会十分困难。
开放讨论在整个重平衡过程中，组内所有消费者实例都会暂停消费，用 JVM GC 的术语来说就是，重平衡过程是一个 stop the world 操作。请思考一下，针对这个问题，我们该如何改进这个过程？我们是否能允许部分消费者在重平衡过程中继续消费，以提升消费者端的可用性以及吞吐量？
欢迎写下你的思考和答案，我们一起讨论。如果你觉得有所收获，也欢迎把文章分享给你的朋友。

分享给需要的人，Ta购买本课程，你将得20元

生成海报并分享

赞 13

提建议

24 | 请求是怎么被处理的？

26 | 你一定不能错过的Kafka控制器

 写留言

精选留言(110)

LRocccccc
置顶
2019-08-02
是我理解错了么？为什么场景一二三的图中的SyncGroup请求，都是等待Leader分配方案，leader不应该提供方案么？
作者回复: 嗯嗯，确实是。只是目前Kafka要求所有consumer都发送SyncGroup请求给Coordinator，因为分配方案只能通过SyncGroupResponse的方式获取。图中只是想表示这是一种机制，没有太区分consumer leader和其他consumer
共 7 条评论
14
rm -rf 😊ི
置顶
2019-07-31
老师，在Broker 端重平衡场景剖析这个第一个图里面，既然协调者说了成员2是这个组的leader，为啥成员2的SyncGroup请求会是”等待leader分配“？这是笔误吗？后面几幅图好像也这样。。。
作者回复: 只是想表明这是统一的一种机制。。。源代码中肯定没有这样的话。。。
共 3 条评论
3
DFighting
2019-09-09
重平衡能不能参照JVM中的Minor gc和Major gc，将重平衡分为两步，在资源的角度讲集群进行分区，这里的资源可以理解为分区，因为后两种变化都是涉及到分区——新主题或已有主题的分区数量变化，对于现有的三种重平衡情况分别做如下处理： 1、新成员入区，在当前区内进行重平衡，不要影响其他的分区 2、资源分区中需要消费的分区队列数量发生的变化，也只是涉及到当前分区的重平衡。这样设计的话就需要处理一个资源分区太空闲和太繁忙时的问题，我觉得可以参考m树的节点分裂和合并，这么做比m树更简单，因为它没有层级关系，只是资源分区的整合和划分而已，实现的时候还能兼顾到网络的局部特性，当然这只是初步想法，没有详细设计和验证，不知道有没有什么地方没有考虑周全，望老师能指点一二。
展开
作者回复: 嗯嗯，非常赞的思路。现在社区正在对rebalance进行改革中有很多思想和你也有重合之处。
共 3 条评论
44
Frank
2019-08-06
这节课，干货很多，我现在有个疑问，重平衡时需要从消费者实例中选择一个leader，让leader来发起重平衡方案，那为啥不直接让协调者组件来处理呢？
作者回复: 客户端自己确定分配方案有很多好处。比如可以独立演进和上线，不依赖于服务器端
共 7 条评论
35
ban
2019-08-01
不会超过 session.timeout.ms 就能感知老师，请问下，消费者已经崩溃了，不会发送心跳，协调者这时候怎么做到能到session.timeout.ms感知的。
作者回复: 每次consumer发送心跳时会顺带发送session timeout时间，这样Coordinator收到后会根据这个session timeout时间计算下次deadline时间，如果过了deadline还没有收到直接fail掉该consumer
共 4 条评论
31
Geek_0819
2020-01-12
老师，有个问题文中说joingroup时等待所有消费者上报订阅信息，协调者通过什么判断所有消费者都已经上报了，或者说怎么知道有多少消费者客户端。如果上报信息后，消费者客户端崩溃了，这能等待下次心跳才能知道吗？
作者回复: join group时也是有一个总的超时时间的（取所有member最大的rebalance超时时间），靠这个作为判断是否进入到下一阶段的阈值。
共 7 条评论
18
菜鸟和谐号
2019-08-19
不同的消费者消费不同的topic主题的领导者都是一个吗，我记得好像不同topic主题的协调者都不是一个啊，老师求解答
作者回复: 一个组对应一个Coordinator，与topic主题无关
17
极极
2020-02-20
老师，最后一张图，再平衡前，要求在规定时间内提交位移，这个规定时间如何设置？默认值是什么？
作者回复: rebalance timeout，默认是max.poll.intervals的值
共 5 条评论
15
wgcris
2019-12-21
老师，您好，请教个问题，最近使用consumer消费，发现consumer会出现commit offset failed，coordinator is not available 的错误，导致consumer卡住，无法消费。根据错误信息，是由于找不到groupcoordinator导致的，但coordinator不可以是在什么情况下发生的？一个猜测是由于consumer端发送心跳超时导致groupcoordinator认为该consumer死了，将该consumer剔除该group，导致该consumer不可用，一直卡住，不知道这种解释是否正确
展开
作者回复: commit失败先看看是不是消息处理慢导致的吧。比如增加max.poll.interval.ms的值或降低max.poll.records的值试试看。Client端报出Coordinator不可用不一定表示Coordinator真的不可用
14
注定非凡
2019-11-08
1 重平衡的通知 A ：重平衡过程通过消息者端的心跳线程（Heartbeat Thread）通知到其他消费者实例。 B ：Kafka Java消费者需要定期地发送心跳请求到Broker端的协调者，以表明它还存活着。（1）在kafka 0.10.1.0版本之前，发送心跳请求是在消费者主线程完成的，也就是代码中调用KafkaConsumer.poll方法的那个线程。这样做，消息处理逻辑也是在这个线程中完成的，因此，一旦消息处理消耗了过长的时间，心跳请求将无法及时发到协调者那里，导致协调者错判消费者已死。（2）在此版本后，kafka社区引入了单独的心跳线程来专门执行心跳请求发送，避免这个问题。 C ：重平衡的通知机制是通过心跳线程来完成的，当协调者决定开启新一轮重平衡后，他会将“REBALANCE_IN_PROGRESS”封装进心跳请求的响应中，发还给消费者实例。当消费者实例发现心跳响应中包含了”REBALANCE_IN_PROGRESS”，就能立即知道重平衡开始了。 D ：消费者端的参数 heartbeat.interval.ms的真实用途是控制重平衡通知的频率。 2 消费者组状态机 Kafka设计了一套消费者组状态机（State Machine），帮助协调者完成整个重平衡流程。 A ：kafka消费者组状态（1）Empty：组内没有任何成员，但消费者组可能存在已提交的位移数据，而且这些位移尚未过期。（2）Dead：组内没有任何成员，但组的元数据信息已经在协调者端被移除。协调者保存着当前向它注册过的所有组信息，所谓元数据就是类似于这些注册信息。（3）PreparingRebalance：消费者组准备开启重平衡，此时所有成员都要重新请求加消费者组（4）CompletingRebalance：消费者组下所有成员已经加入，各个成员正在等待分配方案。（5）stable：消费者组的稳定状态。该状态表明重平衡已经完成，组内成员能够正常消费数据了。 B ：Kafka定期自动删除过期位移的条件就是，组要处于Empty状态。如果消费者组停了很长时间（超过7天），那么Kafka很可能就把该组的位移数据删除了。 3 消费者端重平衡流程 A ：重平衡的完整流程需要消费者端和协调者组件共同参与才能完成。 B ：在消费者端，重平衡分为两个步骤：（1）加入组，对应请求：JoinGroup请求（2）等待领导者消费者分配方案：SyncGroup请求 C ：当组内成员加入组时，他会向协调者发送JoinGroup请求。在该请求中，每个成员都要将自己订阅的主题上报，这样协调者就能收集到所有成员的订阅信息。一旦收集了全部成员的JoinGroup请求后，协调者会从这些成员中选择一个担任这个消费者组的领导者。 D ：通常情况下，第一个发送JoinGroup 请求的成员自动成为领导者。这里的领导者是具体的消费者实例，它既不是副本，也不是协调者。领导者消费者的任务是收集所有成员的订阅信息，然后根据这些信息，制定具体的分区消费分配方案。 E ：选出领导者之后，协调者会把消费者组订阅信息封装进JoinGroup请求的响应中，然后发给领导者，由领导统一做出分配方案后，进入下一步：发送SyncGroup请求。 F ：领导者向协调者发送SyncGroup请求，将刚刚做出的分配方案发给协调者。值得注意的是，其他成员也会向协调者发送SyncGroup请求，只是请求体中并没有实际内容。这一步的目的是让协调者接收分配方案，然后统一以SyncGroup 响应的方式发给所有成员，这样组内成员就都知道自己该消费哪些分区了。 4 Broker端重平衡场景剖析 A ：新成员入组当协调者收到新的JoinGroup请求后，它会通过心跳请求响应的方式通知组内现有的所有成员，强制他们开启新一轮的重平衡。 B ：组成员主动离组消费者实例所在线程或进程调用close()方法主动通知协调者他要退出。这个场景涉及第三类请求：LeaveGroup请求。协调者收到LeaveGroup请求后，依然会以心跳响应的方式通知其他成员。 C ：组成员崩溃离组崩溃离组是指消费者实例出现严重故障，突然宕机导致的离组。崩溃离组是被动的，协调者通常需要等待一段时间才能感知，这段时间一般是由消费者端参数session.timeout.ms控制的。 D ：重平衡时协调者对组内成员提交位移的处理正常情况下，每个组内成员都会定期汇报位移给协调者。当重平衡开启时，协调者会给予成员一段缓冲时间，要求每个成员必须在这段时间内快速地上报自己的位移信息，然后在开启正常JoinGroup/SyncGroup请求发送。
展开
14
明翼
2019-07-30
老师有两个问题请教下： 1）组状态在empty的时候，删除位移信息，这个时间间隔（文中7天）是否可以配置那，还是和普通的默认topic的消息存活时间一样吗？ 2）这个设计我有点迷糊，都有协调者了为什么不让协调者统一做订阅分配那，让领导者做不是更麻烦吗？
作者回复: 1. 可以配置offsets.retention.minutes 2. 新版本consumer的一个改进就是把分区分配策略从server端移到consumer端来做。Client端代码演进的速度和容易程度要远胜于服务器端，算是一个优势吧
14
绿箭侠
2019-10-21
老师，看前面也有人问为啥不把订阅分配方案移到协调者上统一去做？您说Client端代码演进速度和容易程度远胜于服务器端，是一个优势。这里还是没明白，为什么Client端代码演进速度和容易程度更好？！！
作者回复: 这只是其中的一个可能的原因。client端代码更新的难度要远小于broker端。如果是broker代码更新，你需要rolling upgrade所有集群中的broker，在生产环境中并不一定有这样的时间窗口
10
Li Shunduo
2019-07-30
请问当重平衡开启时，协调者会给予提交位移的缓冲时间是多少？如果超过了会拒绝提交的位移吗？
作者回复: 没有具体的限制。反正如果consumer提交的位移请求到broker端时整个group已经从Preparing进化到Completing了，那么就晚了，broker会拒绝这个提交请求
共 2 条评论
9
快跑
2019-08-01
老师你好，订阅主题数量发生变化是指什么？怎么才能触发这个场景的发生？
作者回复: consumer.subscribe(Pattern.compile("test.*"))就可以。这样当你新建了一个test开头的topic，订阅信息就发生变化了，需要重新rebalance
共 3 条评论
8
对与错
2020-07-29
请问订阅主题发送变化之后触发reblance的流程是什么样子的?
作者回复: consumer会定期更新元数据，如果发现了新增的订阅分区，会主动触发rebalance
7
ATSS码农哥
2020-04-20
老师我有一个问题您说consumer group里面的consumer leader会提供分配方案所以这个分配方案是根据coordinator通过joinGroup收集到的所有<topic, partition> 去做一个统一的分配是么？在结束分配以后每个consumer都有可能负责和之前完全不一样的分区？
作者回复: 是的，rebalance结束之后非常可能被分配到和之前完全不一样的分区。社区后面加入了一个StickyAssignor，力图最大程度地保证之前分区分配的粘性，可以试试。具体方法是设置partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor
7
陈国林
2020-01-19
老师好，我说下自己的愚见。为了 Rebalance 过程避免 STW，是否可以开启另外一个线程用于 ”预分配“，预分配的算法最大化的保证 consumer 实例消费的状态不变，确保 Rebalance 过程只需要变更部分 consumer 实例。预分配成功后，再走真正的 Rebalance 流程，不知是否可行
作者回复: 短时间内我不确定这个方案是否可行，大体上看是一个很好的想法。如果可以细化的话，不妨提一个KIP：）
7
兔2🐰🍃
2019-10-24
作者回复: 每次consumer发送心跳时会顺带发送session timeout时间，这样Coordinator收到后会根据这个session timeout时间计算下次deadline时间，如果过了deadline还没有收到直接fail掉该consumer 老师说“Kafka 一般不会超过 session.timeout.ms 就能感知到这个崩溃”，感觉上面回复中的 session timeout 要小于这个消费者端参数，那么心跳请求里的session timeout 是什么计算的？
展开
作者回复: 这块需要更正一次，不是在心跳时携带的session timeout，这个时间在加入组的时候就已经发给Coordinator了。
7
roderickyu
2019-08-26
老师您好，收到重平衡通知后，如果某个consumer提交位移超时了，那么会造成重复消费吧？是不是只能在应用层去重？
作者回复: 有可能重复。目前用业务去重更保险些
7
Mr.Brooks
2020-05-20
消费者组创建的过程中，协调者何时判断所有消费者已加入，还是说它在收到第一个joingroup请求后等一段时间？
作者回复: 嗯，是的。你的理解是对的~
共 2 条评论
6

