5.3.1 “准备再平衡”

系统管理员 2022-11-18 05:46 90阅读 0赞

### 5.3 延迟的加入组操作 ###

> 协调者处理不同消费者的“加入组请求”，由于不能立即返回“加入组响应”给每个消费者，它会创建一个“延迟操作”，表示协调者会延迟发送“加入组响应”给消费者。但协调者不会为每个消费者的“加入组请求”都创建一个“延迟操作”，而是仅当消费组状态从“稳定”转变为“准备再平衡”，才创建一个“延迟操作”对象。

为了保证只创建一个“延迟操作”，只有消费组的状态为“稳定”时才可以创建“延迟操作”，并且在创建“延迟操作”的同时，更新消费组状态为“准备再平衡”。这样协调者在处理下一个消费者的“加入组请求”时，因为消费组状态已经更新为“准备再平衡”，就不会创建“延迟操作”对象了。下面两个步骤示例了协调者如何处理两个消费者的“加入组请求”。

(1)初始时消费组状态为“稳定”，第一个消费者加入消费组。因为消费组状态为“稳定”，所以协调者允许消费者执行再平衡操作。协调者更改消费组的状态为“准备再平衡”，并创建一个延迟的操作对象。
    (2)消费组状态为“准备再平衡”，第二个消费者加入消费组。因为消费组状态为“准备再平衡”，所以协调者不允许消费者执行再平衡操作。消费组状态仍然不变，协调者也石会创建延迟的操作对象。

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

#### 5.3.1 “准备再平衡” ####

> 协调者处理消费者的“加入组请求”，如果消费者设置的成员编号未知，协调者会为这个消费者指定一个新的成员编号，然后创建消费者成员元数据，并加入到消费组元数据中。如果消费者成员编号是已知的，说明消费组元数据中已经存在对应的消费者成员元数据，只需要更新已有的成员元数据。

协调者为消费者分配的成员编号，会作为“加入组响应结果”返回给消费者。消费者发送“同步组请求”时必须指定这个新分配的成员编号，这样协调者才能知道成员编号对应的消费者元数据。实际上，后面如果消费者要重新加入消费组，再次发送“加入组请求”时也需要指定成员编号。通常来  
说，协调者为消费者分配了一个成员编号，协调者的消费组元数据就会一直记录这个消费者的信息。成员编号的主要作用就是用来标识一个消费者，消费者后续的任何请求动作都应该带有分配给它的成员编号。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

协调者处理请求时，doJoinGroup（)方法和doSyncGroup（)方法都会对“消费组元数据”对象进行代码块的同步加锁。目的是保证协调者在处理不同请求时，操作“消费组元数据”对象是线程安全的。

注意：消费组元数据管理“消费者成员元数据”的列表，为了保证列表操作（增加、删除、修改)的线程安全，需妥在“消费组元数据”对象上进行代码块的同步。另外，doJoi.nGroup（)方法依次调用了addMel’lberAndRebalance（)、l’laybePrepareRebalance（)方法。l’laybePrepareRebalance()方法也加了代码块同步，这里的加锁同步因为是在同一个线程内，所以是一个可重入锁。

添加和更新元数据都会执行l’laybePrepareRebalance（)方法。但只有消费组状态为“稳定”或者“等待同步”，才’允许调用prepareRebalance（)方法。PrepareRebalance（)方法会先将消费组状态更新为“准备再平衡”，然后开始执行“准备再平衡”操作。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

注意：旧API基于ZK，每个消费者都会执行再平衡操作。新API采用协调者，只有调用PrepareRebalance（)方法的那个消费者才会启动“再平衡操作”。但消费者启动“再平衡操作”，并不意味着它就是主消费者。比如消费组中已经有多个消费者经过一次再平衡后稳定了，新加入一个未知编号的消费者，它会发起“再平衡操作”，但是负责执行分区分配工作的主消费者，可能还是上一次再平衡完成后的那个，而不是刚刚加入的这个消费者。

消费组的状态从“稳、定”进入“准备再平衡”，表示准备开始再平衡操作。一次再平衡操作只会由一个消费者发起，并只会创建一个延迟的操作对象。延迟操作还和延迟缓存有关，因为延迟操作如果不能及时完成，就应该放入延迟缓存。当然，延迟缓存不是普通的数据缓存，它还要提供检查延迟操作能否完成的方法，并且保证在指定时间内未完成时，必须强制完成超时的延迟操作。

注意：服务端创建的延迟操作对象和消费者客户端创建的延迟任务（心跳任务、偏移量提交任务)很类似，都指定了一个超时时间。但前者表示延迟操作如采一直没有完成，在起时的时候会被强制完成，在这个时间之前，外部事件可以先检查并尝试能否提前完成。而客户端的延迟任务只能等到超时的时候才会被客户端的轮询触发，在超时时间之前不允许执行。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/5bd2353c79144365976929f44ff20bd6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/7cbd7b507f5c41718c4c89aaf8fd3f85.png