为什么需要reblance
由于消费者组订阅了topic,因topic partition数和消费者组成员个数不同而存在的分配机制。
什么情况下会reblance
- Topic partition发生变化。
- 订阅的Topic个数发生变化。
- 消费者组成员个数发生变化。新增成员或已有成员离开。
reblance
的协调者
reblance
过程需要Group Coordinator
的参与。
Group Coordinator
是一个服务,每个Broker
启动的时候都会启动一个该服务。其作用是存储Group的Meta信息,并负责存储其订阅的Topic的partition对应offset信息。
partition的offset信息的存储方式在Kafka不同版本中是不一样的:
在0.9版本以前是存储在ZK中的,存放路径是
consumers/{group}/offsets/{topic}/{partition}
,其中ZK不适合频繁的写操作。在以后的版本中将Partition的Offset信息记录到Kafka内置Topic中,Topic为
__consumer_offsets
上面描述了Group Coordinator
的作用,那新消费者组创建的时候是如何选择自己的Group Coordinator
的?
计算Group对应在Topic
__consumer_offsets
上的partition。-
根据Partition找到该Partition的leader所对应的Broker,该Broker上的
Group Coordinator
就是该Group的Coordinator。
Group在
Topic
__consumer_offsets
上的对应的partition的的计算算法是:// groupId是消费者组的Id // groupMetadataTopicPartitionCount:是__consumer_offsets 的分区数,默认为50 partitionId=Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount)
如何reblance
reblance
发生时,Group下的所有成员都会协调在一起共同参与,kafka能够保证最大公平的分配。但是在reblance过程中,Group下的所有成员实例都会停止消费,直到reblance完成。
reblance主要分为两个操作,加入组(join group)和组信息同步(sync group)。
-
加入组(join group)
这一步主要是该Group的所有成员向其
Group Coordinator
发送JoinGroup
请求,请求加入消费者组。一旦所有成员都发送了JoinGroup
请求,Coordinator
就会从所有消费者组成员中选取一个作为leader,并把组成员信息和订阅信息也发给leader。 -
组信息同步(sync group)
这一步主要是leader分配消费方案。完成分配后,会把分配方案封装
syncGroup
请求中发送给Coordinator
,其中非leader也会发送syncGroup
请求给Coordinator
,只是请求信息为空,Coordinator
接收到syncGroup
请求中的分配方案后,会把方案作为syncGroup
的响应信息发送给各个成员。这样每个组成员都知道自己该消费那些分区了。
怎么避免无谓的reblance
由上可知能引起reblance无非下面三种情况:
- Topic partition发生变化。
- 订阅的Topic个数发生变化。
- 消费者组成员个数发生变化。新增成员或已有成员离开。
其中1和2我们可以人为或约定规范的方式来减少reblance的情况发生,但是3是引起reblance的最常见原因。
除了消费者成员正常的添加和停止之外,还有些情况下Coordinator
会错误的认为消费者组成员已停止而将其踢出组
以致发生reblance。
在描述会发生上述误reblance之前,先解释下consumer端的几个参数:
key | 描述 |
---|---|
session.timeout.ms | 用来控制最大多长时间向coordinator 发送自己活着的心跳不会被认为超时,默认值1o秒 |
heartbeat.interval.ms | 用来控制发送心跳请求频率,值越小,发送心跳频率会越高 |
max.poll.interval.ms | 限定了 Consumer 端应用程序两次调用 poll 方法的最大时间间隔。默认值是 5分钟 |
通过上述三个参数可知,引起误reblance的有以下两种情况:
超过
session.timeout.ms
没有及时发送心跳信息,导致组成员被踢出组。消费时间过长,超过
max.poll.interval.ms
还没有消费完本次poll的所有消息,导致 Consumer 主动发起离开组
的请求。