Kafka Consumer Reblance

kafka consumer 通过偏移量来记录消息的消费进度,当consumer poll一次消息时,

consumer内部维护了一个指针,能够探测到下一条要消费的数据,当reblance的时候,

才会去GroupMetadata消费者组的元数据里拿最近一次提交的offset当初始offset。

通过消费者组的机制,根据负载策略分配各消费者的消费分区以完成消费负载,

默认情况下采用平均分配

当通过subscribe方法订阅某些主题时,此时该消费者还未真正加入到订阅组,只有当

 consumeer#poll 方法被调用后,并且会向 broker 定时发送心跳包,如果 broker 在

 session.timeout.ms 时间内未收到心跳包,则 broker 会任务该消费者已宕机,会将其剔除,

并触发消费端的分区重平衡。

消费者也有活体锁情况,就是消费者正常与broker发送心跳,然后并没有消费进展,要避免

这个消费者占着分区不消费的情况,可以通过配置max.poll.interval.ms参数触发consumer的

pollTimeoutExpired(long now),原理就是通过减小我们poll的频率,导致当前时间-最近一次poll

时间大于max.poll.interval.ms,从而使整个消费者退出消费者组

Reblance 触发条件

触发reblance一般有三种情况,组成员数量发生变化、订阅主题数量发生变化、订阅主题

的分区数发生变化,最常见的是第一种情况,由于消费者的心跳检测判断该消费者该退出

亦或消费者重启后的加入,导致reblance

Reblance 通知机制

consumer的reblance是由broker的Coordinator协调器来完成整个消费者组的重平衡的,

通过一个独立的心跳线程检测各消费者的状态,当Reblance触发,Coordinator想开始

一轮新的reblance,通过在心跳消息里封装“REBALANCE_IN_PROGRESS”消息响应

给各消费者,从而开始重平衡,heartbeat.interval.ms是设置心跳间隔时间的参数,同时

更多的是用来控制重平衡通知的频率,想更快的让消费者组响应重平衡就可以调小这个参数

消费者组状态机

通过消费者组状态机,协调流转整个reblance流程

Empty:组内没有成员,但是消费者组可能存在已提交的位移数据且这些位移还未过期

Dead: 组内无成员且已被Coordinator移除了

PreparingReblance: 准备开始重平衡,所有成员必须重新请求加入消费者组

CompletingReblance: 所有成员已经加入,正在等待分配方案

Stable: 重平衡完成


状态机流转过程

一个消费者组最开始是 Empty 状态,当重平衡过程开启后,它会被置于 PreparingRebalance

 状态等待成员加入,之后变更到 CompletingRebalance 状态等待分配方案,最后流转到 

Stable 状态完成重平衡

Consumer Reblance 流程

从上面的流程可以看出,重平衡第一步就是所有的消费者重新加入消费者组,这个过程

有一个超时时间,如果有成员在超时时间之内,无法完成加入组操作,它就会被排除在

这轮 Rebalance 之外,第二步是在消费者中选出一个leader执行重平衡策略

首先消费者加入的时候,需要向Coordinator汇报自己的所有订阅信息,收集完信息之后,

会将第一个加入的消费者成为leader,然后Coordinator将收集到的订阅信息发给leader,

leader根据各成员订阅的主题以及各主题的分区数,然后根据分配策略决定每个消费者

对应主题该消费哪个分区,然后发给Coordinator,然后Coordinator响应给各个消费者

完成分配工作。

©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容