Kafka 消费者组成员reblance机制

为什么需要reblance

由于消费者组订阅了topic，因topic partition数和消费者组成员个数不同而存在的分配机制。

什么情况下会reblance

1. Topic partition发生变化。
2. 订阅的Topic个数发生变化。
3. 消费者组成员个数发生变化。新增成员或已有成员离开。

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的？

1. 计算Group对应在Topic __consumer_offsets上的partition。

2. 根据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)。

1. 加入组(join group)

   这一步主要是该Group的所有成员向其Group Coordinator发送JoinGroup请求，请求加入消费者组。一旦所有成员都发送了JoinGroup请求，Coordinator就会从所有消费者组成员中选取一个作为leader，并把组成员信息和订阅信息也发给leader。

2. 组信息同步(sync group)

   这一步主要是leader分配消费方案。完成分配后，会把分配方案封装syncGroup请求中发送给Coordinator，其中非leader也会发送syncGroup请求给Coordinator，只是请求信息为空，Coordinator接收到syncGroup请求中的分配方案后，会把方案作为syncGroup的响应信息发送给各个成员。这样每个组成员都知道自己该消费那些分区了。

怎么避免无谓的reblance

由上可知能引起reblance无非下面三种情况：

1. Topic partition发生变化。
2. 订阅的Topic个数发生变化。
3. 消费者组成员个数发生变化。新增成员或已有成员离开。

其中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的有以下两种情况：

1. 超过session.timeout.ms没有及时发送心跳信息，导致组成员被踢出组。

2. 消费时间过长，超过max.poll.interval.ms还没有消费完本次poll的所有消息，导致 Consumer 主动发起 离开组 的请求。