5.4.3 协调者处理“加入组请求”

超、凢脫俗 2023-01-12 15:52 100阅读 0赞

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

#### 5.4.3 协调者处理“加入组请求” ####

> 按照5.4.l节第三种的示例，每个消费者加入消费组都有足够的时间触发一次再平衡操作。在没有完成再平衡操作之前，不会有新的消费者加入消费组。消费组的状态会按照“稳定→准备再平衡→等待同步→稳定”的顺序一直循环下去。虽然消费组的状态变化很有规律，但协调者处理不同消费者的“加入组请求”时执行流程是不同的。下面再举3个不同的示例，前两个示例有两个消费者，我们从正常的状态转换开始，逐步引入异常的状态转换。最后为了模拟另一种异常情况，第气个示例有3个消费者，对应的是5.4.1节的第二种场景。

注意：5.4.1节从程序角度“消费者先同时加入组，再依次一个个加入”来分析。而本节则从正常的状态开始，再依次引入不同的状态转换流程。两种分析角度不同，不过列举的示例大体上是相同的。

1.  有规律的消费组状态转换  
    示例一：两个消费者依次加入消费组。第一个消费者首先从“稳定状态”进入，然后消费组状态分别更改为“准备再平衡”“等待同步”，后面两个状态的更改并不在doJoi.nGroup()方法巾。“稳定状态”改为“准备再平衡”是在addMel’lberAndRebalance()方法中完成的，“准备再平衡”改为“等待同步”是在延迟操作的onCol’lpleteJoi…n()方法中完成的。假设“等待同步”之后，第一个消费组发送“同步组请求”并收到“同步组响应”，消费组状态最终回到‘稳定”状态。

第二个消费者发送“加入组请求”时，消费组的初始状态也是“稳定”状态。但实际上协调者调用了两次doJoi…nGroup()方法，一次是处理第二个消费者的“加入组请求”，另一次是处理第一个消费者重新发送的“加入组请求飞第二个消费者从“稳定状态”进入后，会更改消费组状态为“准备再平衡”，并且创建一个延迟的操作，但是延迟操作不满足完成的条件。第－个消费者重新发送“加入组请求”则是从“准备再平衡状态”进入，并更改消费组状态、为“等待同步”，然后完成延迟的操作。

如图5-14所示，协调者的doJoi…nGroup()方法在第一次再平衡操作时，处理第一个消费者的“加入组请求”从“稳定状态”进入。第二次再平衡操作时，处理第二个消费者的“加入组请求”从“稳、定状态”进入，处理第一个消费者重新发送的“加入组请求”，则从“准备再平衡”状态进入(图中灰色背最部分为协调者处理加入组请求的人- 状态)。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

1.  无规律的消费组状态转换  
    示例二：如果第一个消费者收到“加入组响应”，消费组状态为“等待同步”。这时第二个消费者加入消费组，协调者处理第二个消费者的“加入组请求”则从“等待同步”状态进入。如图5-15所示，协调者一共只发生一次再平衡操作。第一个消费者发送了两次“加入组请求”，第一次是从“稳定状态”进入，第二次是从“准备再平衡”状态进入的。

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

示例三：如图5-16所示，和示例二类似，但在第一个消费者重新发送“加入组请求”之前，第三个消费者先发送了“加入组请求”。协调者处理第三个消费者的人- 状态是“准备再平衡”，但并不会改变消费组的状态，这种情况比较特别，但也是允许的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

上面的后两个示例中，协调者的doJoi.nGroup()方法分别从消费组的3种状态进入请求处理流程。实际上，协调者处理“加入组请求”和“同步组请求”都会有3种状态的入- 。下面的代码简化了“消费者是否已经在消费组中”的判断，添加或更新成员用addOrUpdateMel’lberAndRebalance()方法表示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

3 完成延迟操作相关的方法  
消费组状态为“准备再平衡”时(不管是从“稳定”状态进入，还是从“等待同步”状态进入)，协调者都会创建一个延迟的操作对象，并将延迟的操作加入到延迟缓存巾，同时会在prepareRebalance()方法中立即触发一次“尝试完成或监视延迟的操作”(tryCol’lpleteElseWatch()方法)。

那么，为什么在doJoi.nGroup()方法中，还要再执行一次“检查能杏完成延迟的操作”(checkAndCol’lplete()方法)。这是因为协调者处理“加入组请求”时，不一定每次都会创建延迟的操作，所以不一定每次都会调用tryCol’lpleteElseWatch()方法。因为与延迟操作相关的外部事件有可能会完成它，所以只要有触发它的外部事件，就应该调用checkAndCol’lplete()尽快完成延迟的操作。

如图5-17所示，如果添加或更新消费者成员时，消费组状态已经是“准备再平衡”，则不需要再执行“准备再平衡操作”。因为协调者在处理上一个消费者时，在将消费组状态改为“准备再平衡”时就已经执行了一次“准备再平衡操作”。比如，5.4.3节第二小节最后一个示例中，第二个消费者发送“加入组请求”，协调者会启动一次“准备再平衡操作”。第三个消费者发送“加入组请求”，以及第一个消费者重新发送“加入组请求”则不会再启动“准备再平衡操作”。后两个消费者作为外部事件，由于都可能有机会完成延迟操作，所以要通过延迟缓存检查能杏完成延迟操作。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]

协调者在创建延迟操作的同时，也会立即尝试完成一次延迟的操作：如果这次尝试不能成功完成，则将延迟的操作放入缓存中；如果可以成功完成，会调用延迟操作的onCoMpleteJoi.n()方法，并将消费组状态更改为“等待同步”。当回到doJoi.nGroup()方法的最后一步，判断到消费组状态不是“准备再平衡”，就不需要检查延迟操作能否完成，因为延迟操作已经完成了。

1.  消费组状态机与添加或更新消费者成员  
    协调者处理消费者的“加入组请求”还要考虑消费者是否已经在消费组中。如果消费者以“未知编号”发送“加入组请求”，协调者会先创建对应的消费者成员元数据，然后将其加入到消费组元数据中。在协调者返回给消费者第一次的“加入组响应”结果中，要带有分配的成员编号，这样下一次消费者会以分配的成员编号发送“加入组请求”，协调者就可以从消费组元数据中找出对应编号的消费者成员元数据。

前面示例中，新的消费者发送“加入组请求”时，会触发消费组状态更改为“准备再平衡”。已有的消费者重新发送“加入组请求”时，协调者处理这些已有消费者的“加入组请求”，都是从“准备再平衡”状态进入updateMeMberAndRebalance()方法。但协调者处理新消费者的“加入组请求”流程与之不同，两者的区别如作。

*  消费组状态是“稳定”，新消费者加入后，会启动一次新的再平衡操作。新消费者从“稳定”进入addMeMberAndRebalance()方法，并更改状态为“准备再平衡”。已有消费者需要重新发送“加入组请求”。
 *  消费组状态是“等待同步”，新消费者加入后，不会启动新的再平衡操作。新消费者从“等待同步”进入addMeMberAndRebalance()方法，并更改状态为“准备再平衡”。已有消费者也需要重新发送“加入组请求”。

注意：addMeMberAndRebalance()和updateMeMberAndRebalance()方法创建或更新消费者元数据，都会调用prepareRebalance()方法。在一次再平衡操作中，协调者处理消费组中所有消费者的“加入组请求”，其中只有一次机会调用prepareRebalance()方法。一旦调用该方法，消费组状态更改为“准备再平衡”，协调者处理其他消费者的“加入组请求”只能从“准备再平衡”的状态入- 进去。

协调者处理消费者的“加入组请求”，根据消费组的不同状态分别执行不同的分支流程。相关代  
码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]

先来看消费组状态入- 是“准备再平衡”时，消费者成员编号是否是“未知编号”的示例。第一个消费者(Cl)加入组，从“稳定”到“准备再平衡”再到“等待同步”；第二个消费者(C2)加入组，从“等待同步”进入到“准备再平衡”；在这之后，新(第三个，C3)旧(第一个，Cl)消费者都有可能从“准备再平衡”进入。如图5-18(中)所示，新消费者先于旧消费者发送“加入组请求”，对应的处理流程如下。

(1)新消费者在旧消费者重新发送“加入组请求”之前，先发送了“加入组请求”，对应添加成员
    方法。
    (2)旧消费者重新发送“加入组请求”，对应更新成员方法(updateMel'lberAndRebalance())。

如图5-18(右)所示，旧消费者重新发送“加入组请求”后，新消费者才发送“加入组请求”。协调者会先处理旧消费者的“加入组请求”，处理完成后才开始处理新消费者的“加入组请求”。但由于协调者处理旧消费者的“加入组请求”时，就完成延迟操作，并更改消费组状态为“等待同步”，因此协调者处理新消费者的“加入组请求”时，不再从“准备再平衡”状态进入，而是从“等待同步”进入。

正常情况下，协调者处理消费者的“加入组请求”，如果从消费组的“稳定”状态进入，这之后消费组没有完成本次的再平衡操作，其他消费者都不会再从“稳、定”状态进入了，只能根据不同场景从其他两个状态“准备再平衡”和“等待同步”进入。比如图5-18中，在步骤(1)之后，本次的再平衡操作没有结束之前，后续的步骤(2)到步骤(4)都从另外两个状态进入。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]

如图5-19所示，从“准备再平衡”进入的状态有两个：“稳定”和“等待同步”。心形图的左半部分是一次正常的再平衡操作，并且可以一直按照这种状态轨迹循环下去。右半部分表示再平衡操作处于“等待同步”状态时，就有新的消费者加入组。不过后者只要满足完成“延迟操作”的条件，最终也会走左半部分的流程。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]

再来看消费组状态人- 是“等待同步”时，消费者成员编号是杏是“未知编号”的示例。未知编号对应添加消费者成员，这在前面已经分析过多次了，比如图5-18中的第二个消费者、图5-19巾，心形图的右半部分。但是消费组状态入- 是“等待同步”，而且消费者成员是已知的，目前还没看到这样的示例。

正常来说，在“等待同步”状态时，新消费者加入组很好理解，但是旧消费者加入组就有点异常。如图5-20(左)所示，正常的流程是消费者发送“加入组请求”(步骤(1))，协调者发送“加入组响应”给每个消费者。如图5-20(右)所示，消费组状态更改为“等待同步”，而且协调者也向每个消费者都发送了“加入组响应”。但如果消费者没有收到“加入组响应”(步骤(2))，它就会重新发送“加入组请求”(步骤。))。协调者处理已知消费者的“加入组请求”并不会更改消费组状态，也不需要执行添加或更新消费者成员方法，只需要直接返回“加入组响应”(步骤(4))。注意：步骤(3)到步骤(4)并不会改变消费组的状态。

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

最后再来看消费组状态人- 是“稳定”时，消费者成员加入消费组时编号已知的场最。如果消费者发送了“加入组请求”，但没有收到“加入组响应”，最终的状态转为“等待同步”；如果消费者发送了“同步组请求”，但没有收到“同步组响应”，最终的状态仍然是“稳、定”状态。

注意：协调者给每个消费者都发送完响应结果，并不保证消费者就一定能收到响应结果。如果是消费者内部自己的问题，就需要重新发送请求：比如消费者没有收到“加入组响应”，需要重新发送“加入组请求”；没有收到“同步组响应”，需妥重新发送“同步组请求”。

总结下协调者处理消费者的加入组请求和同步组请求。协调者的doJolnGroup()方法最多只能将消费组状态转换到“等待同步”，从“等待同步”状态转变到“稳定”状态，只能在doSyncGroup()方法中做到。消费组到达“稳定”状态后，表示协调者处理了主消费者发送的“同步组请求”，但这并不意味着其他消费者也都发送了“同步组请求”。如图5-21所示，普通消费者发送了“加入组请求”，但没有收到“加入组响应”，需要重新发送“加入组请求”。协调者处理这种消费者重新发送的“加入组请求”时，因为消费组状态已经是“稳定”，所以它会立即返回“加入组响应”给消费者。具体步骤如下。

(1)3个消费者发送“加入组请求”给协调者，消费组状态从“准备再平衡”到“等待同步”。
    (2)协调者返回“加入组响应”给3个消费者，前两个消费者收到“加入组响应”，第三个没收到
    响应。
    (3)第二个消费者是普通消费者，它发送“同步组请求”，协调者处理时只是设置对应的回调方法。
    (4)第一个消费者是主消费者，它完成分区分配，发送“同步组请求”给协调者，消费组状态改
    为“稳定”。
    (5)协调者返回“同步组响应”给两个消费者，因为只有前两个？肖费者的回调方法不为空。
    (6)第三个？肖费者没收到“加入组响应，，，它重新发送“加入组请求”
    (7)协调者处理第三个？肖费者的“加入组请求”，？肖i费组状态是“稳、定”，立即返回“加入组响应”。
    (8)第三个？肖费者因为不是主消费者，它收到“加入组响应”，会立即发送“同步组请求”。
    (9)协调者处理第三个消费者的“同步组请求”，消费组状态是“稳定”，立即返回“同步组响应“。

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

上面的示例是普通消费者没有收到“加入组响应”，但如果是主消费者没有收到“加入组响应”就不同了。如图5-21所示，由于主消费者没有收到“加入组响应”，它就不可能执行分区分配工作，也不会发送“同步组请求”。协调者也不会将消费组状态更改为“稳定”，因此仍然停留在“等待同步”，只能等主消费者及时地重新发送“加入组请求”。这种场景下，整个过程的具体步骤如下。

(1)3个消费者发送“加入组请求”给协调者，消费组状态从“准备再平衡”到“等待同步”。
    (2)协调者返回“加入组响应”给3个消费者，前两个消费者收到“加入组响应“，第三个没收到
    (3)前两个消费者都是普通消费者，它发送“同步组请求”，协调者处理时只是设置对应的回调
    方法。
    (4)主消费者没有收到“加入组响应”，重新发送“加入组请求”。
    (5)协调者处理主消费者的“加入组请求”，消费组状态是“等待同步”，立即返回“加入组响应”。
    (6)主消费者收到“加入组响应”，执行分区分配丁．作，并发送“同步组请求”。
    (7)协调者处理主消费者的“同步组请求”，返回响应给所有消费者，更新消费组状态为“稳定”。

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

消费组的一次再平衡操作对于每个消费者而言，都需要发送“加入组请求”和“同步组请求”。上面分析协调者处理“加入组请求”的相关状态机变化，其实也都有涉及“同步组请求”的状态变化。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/7cb0c84dd6624fc0bb3940fc07a7622c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/f13fb83af58d4553a4a0617adcc6fa91.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/5df25a3ca4804748b5a9df351e8b32d0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/1a0ed5d36fcf4d139c1795606841e999.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/5f684464f2f64c24b8ff14d06ed4a6d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/50479937aee842b3af3191199261d766.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/66917801fc4b4d8597b808cb966290a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]: /images/20221022/4554da79cfb24d65ac53e5e7d63d653c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 8]: /images/20221022/a014ee0b135c4752b7b287809b736403.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 9]: /images/20221022/ea342d083f72495dbf54decc86f0cb31.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 10]: /images/20221022/35a9a094e4634625b1506fda4f3ce674.png