5.4.4 协调者处理“同步组请求”

水深无声 2022-11-19 06:27 128阅读 0赞

#### 5.4.4 协调者处理“同步组请求” ####

消费组状态进入“等待同步”后，协调者会发送“加入组响应”给每个消费者。如果是普通消费者收到“加入组响应”，会立即发送“同步组请求”。而如果是主消费者收到，它会先执行完分区分配工作，然后才发送“同步组请求”。正常情况下，普通消费者会先于主消费者发送“同步组请求”。协调者处理普通消费者的“同步组请求”，只会设置对应消费者成员元数据的回调方法，因为主消费者还没有发送“同步组请求”，就还没有到返回“同步组响应”的时机，所以只能将回调方法暂存到元数据中。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

协调者处理消费者发送的“同步组请求”时，还会从“准备再平衡”“稳定”两种状态进入。先来看消费组状态为“准备再平衡”的处理步骤。

(1)第一个消费者加入组，消费组状态为“等待同步”，第一个消费者同时作为主消费者还在执行
    分区分配工作。
    (2)第二个消费者作为新的消费者加入组，将消费组状态改为“准备再平衡”
    (3)第一个？肖费者执行完分区分配工作，会发送“同步组请求”给协调者。
    (4)协调者处理第一个消费者的“同步组请求”，由于消费组状态是“准备再平衡”，它会返回“正在再平衡”的错误码给第一个消费者。
    (5)第一个消费者收到错误的“同步组响应”，会重新发送“加入组请求".

上面的示例中，第二个消费者加入组将消费组状态改为“准备再平衡”(步骤。))。但第一个消费者还认为消费组状态是“等待同步”，它作为主消费者只顾执行分区分配任务(步骤(1))。第一个消费者执行分配工作后，发送“同步组请求”给协调者后(步骤。))，才意识到向己之前的工作其实都向做了，因为现在有一个新的消费者加入了。即使协调者允许接受第一个消费者的“同步组请求”，在这之后，消费组成员不再只有第一个消费者，还是需要再执行一次再平衡操作的。与其那样，协调者还不如直接拒绝第一个消费者的“同步组请求”(步骤(4))，让它重新发送“加入组请求”(步骤(5)),这样消费者和协调者的交互步骤就不会过于烦琐、冗余。

再来看协调者处理“同步组请求"，从“稳定”状态进入的场景。有3个消费者，第一个是主消费者，第二个和第三个都是普通消费者。它们都加入组后，消费组状态为“等待同步”。下面的步骤是协调者处理这3个消费者发送“同步组请求”的不同执行顺序和流程。

(1)第二个消费者收到“加入组响应”后立即发送“同步组请求”，由于第一个消费者还没发送“同步组请求”，协调者处理第二个消费者的“同步组请求”会设置第二个消费者成员元数据的回调方法。
    (2)第一个消费者执行完分区分配，发送“同步组请求”。协调者处理第一个消费者的“同步组请求”，也会设置第一个消费者成员元数据的回调方法。
    (3)协调者返回“同步组响应”给元数据中有回调方法的消费者，即第一个消费者和l第二个消费者。因为第三个消费者还没发送“同步组请求”，所以它的元数据回调方法为空，协调者不会返回响应给它。
    (4)协调者发送完“同步组响应”后，更改消费组状态为“稳定”。
    (5)第三个消费者发送“同步组请求”，由于消费组状态为“稳定”，协调者直接返回“同步组响应”给它。

前面我们分析了协调者处理“加入组请求”和“同步组请求”的流程，以及相关的消费组状态机转换。除此之外，消费者也可能会离开消费组。比如消费者应用程序被手动停掉，虽然消费者不会马上从消费组元数据中移除，但是协调者会对注册在消费组中的所有消费者进行监控，一旦消费者没有反应，就会将其从消费组中移除。如果消费组中所有消费者成员都离开了，协调者也会把消费组删除掉。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/b3601b6569e74fdc9f0551ad5bab16e6.png