3.2.1 分区的所有权

爱被打了一巴掌 2022-11-13 05:22 116阅读 0赞

### 3.2 消费者再平衡操作 ###

消费者连接器的核心处理逻辑是ZKRebalanceListener的再平衡操作，它起了承上启下的作用，上一节初始化消费者连接器只是“创建了队列和消息流”，再平衡操作会“为消费者重新分配分区”。只有为消费者分配了分区，拉取线程才会开始拉取分区的消息。

因为分区要被重新分配，分区的所有者都会发生变化，所以在还没有重新分配分区之前，所有消费者都要停止已有的拉取钱程。同时，分区分配给消费者都会在ZK中记录所有者信息，所以也要`先删除ZK上的节点数据。只有和分区相关的ZK所有者、拉取线程都释放了，才可以开始分配分区`。

如果说在重新分配分区前没有释放这些信息，再平衡后就可能造成同一个分区被多个消费者所有的情况。比如分区P1原先归消费者1所有，如果没有释放拉取钱程和ZK节点，再平衡后分区Pl被分配给消费者2了，这样消费者l和消费者2就共享了分区P1，而这显然不符合Kafka中关于“一个分区只能被分配给一个消费者”的限制条件。ZKRebalancelistener.rebalance()执行再平衡操作的步骤如下。

(1)关闭数据拉取线程，清空队列和消息流，提交偏移量。
    (2)释放分区的所有权，删除ZK中分区和消费者的所有者关系。
    (3)将所有分区重新分配给每个消费者，每个消费者都会分到不同的分区。
    (4)将分区对应的消费者所有者关系写入ZK，记录分区的所有权信息。
    (5)重新启动消费者的拉取线程管理器，管理每个分区的拉取线程。

相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]  
拉取线程和分区所有权的关闭和开启顺序为：停止拉取钱程→释放分区的所有权→添加分区的所有权→启动拉取线程。如果先释放分区的所有权，就会出错，因为拉取钱程依赖于分区。同样，如果先启动拉取线程，然后才添加分区的所有权，也会导致拉取线程因没有分区而无法工作。

--------------------

#### 3.2.1 分区的所有权 ####

分区的所有权记录在ZK的／consumers/\[group\_id\]／owner／\[topic\]/\[partition\_id\]→consumer＿thread\_id节点，表示“主题一分区”会被指定的消费者线程所消费，或者说分区被分配给消费者、消费者拥有了分区。要释放分区的所有权，只需要删除分区对应的ZK节点；要重建分区的所有权，数据源中除了包含分区，还要有消费者线程编号。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
释放或添加分区所有权的方法中，参数localTopicRegistry和ParttitionAssignment这两个数据集是通过其他步骤产生的。ZK中不仅记录了消费者和分区的所有权映射关系，而且记录了消费组的消费者列表、主题的分区列表，这些信息为消费者分配分区提供了数据来源。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/46aa53fb1d144a849b51d20a67f2465f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210325203446322.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0,size_16,color_FFFFFF,t_70