在同一个Consumer Group中，同一个Topic的不同分区会分配给不同的消费者进行消费，给消费者分配分区的操作是在哪里进行的呢？分区又是如何进行分配的呢？先介绍下Rebalance操作的原理。
方案一：
Kafka老版本的解决方案是通过Zookeeper的Watch实现。每个Consumer Group在Zookeeper下维护了一个“/consumers/[group_id]/ids”路径，在此路径下使用临时节点记录属于此Consumer Group的消费者Id，由Consumer启动时创建。还有两个与ids节点同级的节点，分别是：owners节点，记录了分区与消费者的对应关系；offsets节点，记录了此Consumer Group在某个分区上的消费位置。
每个Broker,Topic以及分区在Zookeeper中也都对应一个路径，如下：
-
/brokers/ids/broker_id:记录了host,port以及分配在此broker上的topic的分区列表。
-
/brokers/topics/[topic_name]:记录了每个partition的Leader,ISR等信息。
-
/brokers/topics/[topic_name]/partitions/[partition_num]/state:记录了当前Leader,选举epoch等信息。
consumer group在zookeeper路径.jpg
每个Consumer都分别在“/consumers/[group_id]/ids”和“/brokers/ids”路径上注册一个Watcher。当“/consumers/[group_id]/ids”路径的子节点发生变化时，表示Consumer Group中的消费者出现了变化；当“/brokers/ids”路径的子节点发生变化时，表示Broker出现了增减。这样，通过Watcher，每个消费者就可以监控consumer group和Kafka集群的状态了。但是严重的依赖zookeeper集群，会出现两个比较严重的问题：
-
羊群效应(Herd Effect)：一个被Watch的zookeeper节点变化，导致大量的Watcher通知需要被发送给客户端，这将导致在通知期间其他操作延迟。一般出现这种情况的主要原因是没有找到客户端真正的关注点，是滥用Watcher的一种场景。任何Broker或Consumer加入或退出，都会向其余所有的Consumer发送Watcher通知触发Rebalance，就会出现“羊群效应”。
-
脑裂(Split Brain)：
每个Consumer都是通过Zookeeper中保存的这些元数据判断consumer group的状态，Broker的状态以及Rebalance结果的。由于Zookeeper只保证“最终一致性”，不保证“Simultaneously Consistent Crosss-Client Views”,不同Consumer在同一个时刻可能连接到Zookeeper集群中不同的服务器，看到的元数据就可能不一样，这就会造成不正确的Rebalance尝试。
方案二：
由于上述两个原因，Kafka后续版本进行了改进，也对消费者进行了重新设计。设计思想是：将全部的Consumer Group分成多个子集，每个Consumer Group子集在服务器对应一个GroupCoordinator对其进行管理。GroupCoordinator是KafkaServer中用于管理Consumer Group的组件。消费者不再依赖Zookeeper，而只有GroupCoordinator在Zookeeper上添加Watcher。消费者在加入或退出Consumer Group时会修改Zookeeper中保存的元数据，此时会触发GroupCoordinator设置的Watcher，通知GroupCoordinator开始Rebalance操作。简述过程：
1） 当前消费者准备加入某Consumer Group或GroupCoordinator发生故障转移时，消费者不知道GroupCoordinator的网络位置，消费者会向Kafka集群中任一Broker发送ConsumerMetadataRequest，此请求中包含了其Consumer Group的GroupId,收到请求的Broker会返回ConsumerMetadataResponse作为响应，其中就包括了管理此Consumer Group的GroupCoordinator相关信息。
2） 消费者根据ConsumerMetadataResponse中的GroupCoordinator信息，连接到GroupCoordinator并周期性的发送HeartbeatRequest。发送HeartbeatRequest的主要作用是通知GroupCoordinator这个消费者是在线的，GroupCoordinator会认为长时间未发送HeartbeatRequest的消费者已经下线，这样会触发新的Rebalance。
3）如果HeartbeatResponse中带有IllegalGeneration异常，说明GroupCoordinator发起了Rebalance操作，此时消费者发送JoinGroupRequest给GroupCoordinator，JoinGroupRequest的主要目的是为了通知GroupCoordinator会根据收到的JoinGroupRequest和Zookeeper中的元数据完成对此Consumer Group的分区分配。
4）GroupCoordinator会在分配完成后，将分配结果写入Zookeeper保存，并通过JoinGroupResponse返回给消费者。消费者根据JoinGroupResponse中分配的分区开始消费数据。
5）消费者成为Consumer Group的成员后，会周期性发送HeartbeatRequest。如果HeartbeatResponse包含IllegalGeneration异常，则执行第三步，如果找不到对应的GroupCoordinator（HeartbeatResponse包含NotCoordinatorForGroup），则周期性的执行步骤1，直到成功。
这个方案虽然解决了“羊群效应”和“脑裂”，还是有两个问题：
- 分区分配的操作是在服务端的GroupCoordinator完成的，这就要求服务端实现Partition的分配策略。当要使用新的Partition分配策略时，就必须修改服务端的代码或配置，之后重启服务，比较麻烦。
- 不同的Rebalance策略有不同的验证需求。当需要自定义分配策略和验证需求时就会比较麻烦。
方案三：
为了解决上述问题，Kafka0.9进行了重新的设计，将分区分配的工作放到了消费者处理，但Consumer Group管理的工作依然由GroupCoordinator处理。这就让不同的模块关注不同的业务，实现了业务的拆分和解耦，这种思想在设计时很重要。
重新设计后的协议在上一个版本的协议上进行了修改，将JoinGroupRequest的处理过程分为两个阶段，分别是Join Group阶段和Synchronizing Group State阶段。
新版本的变化有什么？
当消费者查找到管理当前Consumer Group的GroupCoordinator后，就会进入Join Group阶段，Consumer首先向GroupCoordinator发送JoinGroupRequest请求，其中包含消费者的相关信息；服务端的GroupCoordinator收到JoinGroupRequest后会暂存消息，收集到全部消费者之后，根据JoinGroupRequest中的信息来确定Consumer Group中可用的消费者，从中选取一个消费者成为Group Leader，还会选取使用的分区分配策略，最后将这些信息封装成JoinGroupResponse返回给消费者。
虽然每个消费者都会收到JoinGroupResponse，但是只有Group Leader收到的JoinGroupResponse中封装了所有消费者的信息。当消费者确定自己是Group Leader后，会根据消费者的信息已经选定的分区分配策略进行分区分配。
在Synchronizing Group State阶段，每个消费者会发送SyncGroupRequest到GroupCoordinator，但是只有Group Leader的SyncGroupRequest请求包含了分区的分配结果，GroupCoordinator根据Group Leader的分区分配结果，形成SyncGroupResponse返回给所有的Consumer。消费者收到SyncGroupResponse后进行解析，即可获取分配给自身的分区。
网友评论