5.4.6 再平衡超时与会话超时

「爱情、让人受尽委屈。」 2023-01-12 15:52 206阅读 0赞

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

#### 5.4.6 再平衡超时与会话超时 ####

> 消费组成员\[Cl\]\[Cl,C2\](CI,C2\](Cl:Pl,C2:P2\]\[Cl\]\[CI\]\[Cl:\{Pl,P2\}\]当消费组状态是“准备再平衡”，协调者会创建一个“延迟的加入组”对象。这个对象表示协调者在处理消费者的“加入组请求”时，由于还没有收集完整“重新发送加入组请求”的消费者，所以还不能返回“加入组响应”给发送了“加入组请求”的消费者。当消费组中的所有消费者都发送(或重新发送)了“加入组请求”，“延迟的加入组”对象就可以完成，协调者这时才会返回“加入组响应”给所有的消费者。

协调者等待“延迟操作”完成有一个时间限制，它会选择消费组巾所有消费者会话超时时间的最大值，作为“再平衡操作的超时时间”，也叫作“延迟操作的超时时间”。如果是第一个消费者加入组，再平衡操作的时间等于第一个消费者的会话超时时间。但因为目前消费组中只有第一个消费者，所以协调者刚刚创建的延迟操作可以马上完成。当第二个消费者加入组后，再平衡操作的时间会选择两个消费者的会话超时时间最大值。比如第一个消费者的会话超时时间是10秒，第二个消费者的会话超时时间是5秒，再平衡操作的超时时间等于10秒，延迟的加入操作会最多等待10秒，等待第一个消费者在这段时间内可以重新发送“加入组请求”。

为延迟操作设置超时时间是为了防止延迟操作一直无法完成。假设原有的消费者迟迟没有重新发送“加入组请求”，协调者就无法确定何时才可以返回“加入组响应”给已经发送了“加入组请求”的消费者。从创建延迟操作，经过了“再平衡操作超时时间”之后，延迟操作会被强制完成。在完成延迟操作时，协调者会找出那些没有在规定时间内重新发送“加入组请求”的消费者，将它们从消费组中移除摊。相关代码如下：

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

在完成“延迟的加入组”对象中，协调者会返回“加入组响应”给消费组中的所有消费者。这也是在onCo叩leteJoi.n()方在中，要事先移除掉超时消费者的原因。在发送“加入组响应”时，消费组中的所有消费者一定都在“再平衡操作超时时间”内及时发送了“加入组请求”。协调者返回“加入组响应”给每个消费者后，都会立即完成本次“延迟的心跳”，并调度下一次  
“延迟的心跳”。“延迟的心跳”和“延迟的加入组”概念上相同，前者因为是消费者级别，超时时间是消费者自己的会话超时时间；后者因为是消费组级别，超时时间是所有消费者的最大会话超时时间。相关代码如下：  
![在这里插入图片描述][20210414214456190.png]

如图5-23所示，在消费组的一次再平衡操作过程中，服务端的协调者只有一个延迟的加入对象(DelayedJoi.n)，并且它会为每个消费者保存一个延迟的心跳对象(DelayedHeartbeat)，用来监控消费者是否及时地发送心跳，具体步骤如下。

(1)消费者发送“加入组请求”时会指定会话的超时时间(简称“会话时间”)。
    (2)协调者不能立即返回“加入组响应”给消费者，创建一个消费组级别的“延迟加入”。
    (3)“延迟加入”可以完成，协调者返回“加入组响应”给消费组中的每个消费者。
    (4)协调者为每个消费者都创建一个“延迟心跳”，并监控每个消费者是否存活。

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

协调者在处理完消费者的“加入组请求”后，会返回“加入组响应”给消费者。消费者收到“加入组响应”后，就应该在会话时间内及时发送“同步组请求”给协调者；杏则，协调者就会认为消费者出现了故障。协调者在处理“同步组请求”时，有多个地方调用了“完成若调度下一次心跳”方法。

(1)状态为“等待同步”，在设置成员元数据的回调方法后调用。
    (2)状态为“稳定”，在发送“同步组响应”给消费者后调用。
    (3)状态为“等待同步”，收到主消费者的“同步组请求”，给每个消费者发送“同步组响应”后调用。

第三处的用法和协调者处理“加入组请求”时，给每个消费者发送“加入组H向应”后调用“完成并调度下一次心跳”方法类似。它们都针对所有消费者，而不是单个消费者。但前面两个用法，只针对一个消费者。相关代码如下：

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

协调者创建完“延迟操作”对象后，一个很重要的步骤是：当“延迟操作”相关的外部事件发生时，就需要通过延迟缓存尝试完成延迟的操作。对于“延迟的加入组”，外部事件是消费者发送了“加入组请求”；对于“延迟的心跳”，外部事件则是协调者和消费者之间有网络通信。不管是协调者处理消费者发送的请求，还是协调者发送响应给消费者，协调者都会完成本次延迟的心跳，并开始调度下一次延迟的心跳。如图5-24所示，协调者完成“延迟加入”操作的时间是10:00:00，它为每个消费者创建的“延迟心跳”，会根据当前时间和消费者的会话超时时间，设置下一次“延迟心跳”的截止时间(deadli.ne)。

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

如图5-25所示，消费者在收到“加入组响应”后，会发送“同步组请求”给协调者。协调者依次处理每个消费者的“同步组请求”，会先完成延迟缓存中已有的“延迟心跳”，然后创建新的“延迟心跳”。这里假设协调者最后才处理主消费者的“同步组请求”。如果普通消费者发送“同步组请求”比主消费者还要晚，它也会先完成已有的“延迟心跳”，并创建新的“延迟心跳”。但协调者在调用PropagateAssi.gnment()方法时，就不会为这样的普通消费者再次调用“完成并调度下一次心跳”。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]

如图5-26所示，协调者在返回“同步组响应”给每个消费者后，会为每个消费者都调用一次“完成并调度下一次心跳”。注意：协调者收到主消费者的时间是10:00:38，但返回“同步组响应”的时间是10:00:40，在这段时间内，协调者会将“同步组请求”的“消费组分配结果”保存到内部主题。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]

“延迟加入”完成时，协调者发送“加入组响应”给所有的消费者。下面分析“延迟的心跳”在完成时的动作。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/3305e001664947b3af651fedb7f41d8e.png
[20210414214456190.png]: /images/20221022/625122fea77042bebea3f5435a9417a5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/48350134f3ed4c87882caabd6c9d57cf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/a2a2b0e75b6f4e34b4b4f09ff4bde67b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/e99a4e214d3e4be2bebae3bd442f438c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/275b8dddf89c414290a4cd06121a745e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/48df62ec8d0f4af1a3f18f3db6470656.png