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响应给各个消费者
完成分配工作。