5.4.1 再平衡操作与监昕器

喜欢ヅ旅行 2022-11-19 06:26 5阅读 0赞

#### 5.4 消费组状态机 ####

> 协调者保存的消费组元数据中记录了消费组的状态机，消费组状态机的转换主要发生在“加入组请求”和“同步组请求”的处理过程中。除此之外，协调者处理“离开消费组请求”“迁移消费组请求”“心跳请求”“提交偏移量请求”也会更新消费组的状态、机，或者依赖消费组的状态进行不同的处理。

消费者要加入消费组，需要依次发送“加入组请求”和“同步组请求”给协调者。消费者加入组的过程叫作再平衡，因为协调者处理这两种请求会更新消费组状态，所以再平衡操作跟消费组状态也息息相关。

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

#### 5.4.1 再平衡操作与监昕器 ####

> 为了验证“消费组的状态转换步骤”和“协调者在不同时间处理不同消费者的加入组请求”两者的关系，我们可以使用5.1.4节的“消费者再平衡监昕器”，通过监控监昕器的回调方法来验证。下面的再平衡监听器仅仅打印了分区信息，监昕器回调方法和再平衡操作的执行顺序如下。

(1)消费者准备加入组，调用“消费者再平衡监昕器”的onPart"i.ti.onsRevoked()方法。
    (2)消费者发送“加入组请求”，协调者开始处理消费者的“加入组请求”，执行再平衡操作。
    (3)消费者完成加入组，调用“消费者再平衡监昕器”的onParti.ti.onsAssi.gned()方法。

相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

注意：“消费者再平衡监听器”回调方法的流程是在客户端中调用的。消费者的协调者  
(ConsumerCoordinator)在onJo1.nPrepare()中调用监听器的onPartitionsRevoked()方法，在  
onJo1.nCol’lplete()中调用监听器的onPartitionsAss1.gned()方法。因为一次再平衡操作只会调用一次onJo1.nPrepare()和onJo1.nCol’lplete()方法，所以监听器的两个回调方法在一次再平衡操作中也只会调用一次。

要模拟协调者在不同时间处理不同消费者的“加入组请求”，我们可以通过启动3个终端程序完成。但要做到消费者同时加入消费组，这种方式很难实现。我们在一个示例程序中同时启动3个消费者线程，就可以达到“同时加入组”的效果。注意：为了保证每个消费者都分配到分区，主题至少要有3个分区。相关代码如下：

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

下面模拟3个消费者按照不同顺序加入组的场景，每种场景发生的再平衡操作次数都不同。

*  3个消费者同时加入消费组。这种方式只启动一次再平衡，但第一个消费者发送多次加入组请求。
 *  第一个消费者加入组，睡眠一秒后第二个消费者加入组，再睡眠一秒后第三个消费者加入组。
 *  第一个消费者加入组，睡眠一秒后第二个消费者加入组，再睡眠两秒后第三个消费者加入组。

1.  3个消费者，一次再平衡操作

> 第一种场景：3个消费者同时发送了“加入组请求”。下面的日志中，消费者调用onPartitionsRevoked()方法发送“加入组请求”之前3个消费者可以同时发送加入组请求。但协调者处理请求时，因为在“消费组元数据上”加了锁的保护，只能一次处理一个消费者的“加入组请求飞即第一个消费者收到加入组响应之前，第二个消费者的加入组请求不会被处理。

日志中3个消费者的onPartitionsRevoked()方法后，紧接着是3个消费者的onPartH1.onsAss1.gned()方法。这说明该场景下一共只发生了一次再平衡操作，即只执行了一次分区分配工作，3个消费者最后都收到了分配的分区。只执行一次再平衡操作，并不意味着消费组的状态为“准备再平衡”只有一次：第一个消费者加入组，消费组状态从“稳定”到“准备再平衡”再到“等待同步”；第二个消费者加入组时，消费组状态会从“等待同步”到“准备再平衡气相关日志如下：

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

为了验证协调者处理3个消费者发送请求的处理顺序，我们在GroupCoordi.nator源码的doJoi.nGroup()和doSyncGroup()方法中添加了两处日志，然后在IDEA中启动Kafka服务，而不是通过命令行启动。相关日志如下：

下面是再平衡操作时Kafka服务端相关的日志记录，日志对应的请求发送和处理步骤如下。

(1)3个消费者同时发送加入组请求，consuf'lter－3(第一个消费者)的请求优先被协调者处理。
    (2)consuf'lter－3(主消费者)完成延迟操作，消费组状态改为“等待同步”
    (3)处理co『lSl』『~er－2的“加入组请求，，，消费组状态改为“准备再平衡”，延迟操作不能完成。
    (4)处理co『lSl』f'ller－1的“同步组请求"，延迟操作不能完成；处理consumer-3的同步组请求。
    (5)消费组状态为“再平衡”，conSl』lη1er－3重新发送“加入组请求”。
    (6)延迟操作可以完成，返回“加入组响应”给3个消费者；consuf'lter’3发送“同步组请求”。
    (7)协调者保存consuf'lter－3发送的分配结果时，也会处理普通消费者的“同步组请求”。
    (8)消费组的分配结果保存完毕后，返回“同步组响应”给所有的消费者，消费组状态为“稳定”。
    (9)关闭应用程序，消费组状态为“离开”，3个消费者都会从消费组中移除。

相关日志如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

注意：日志中第三次再平衡操作并不是因为协调者处理第三个消费者“加入组请求”，而是关闭应用程序后消费纽状态更改为“离开”引起的。协调者在处理第二个消费者的“加入组请求”时，会等待第一个消费者重新发送“加入组请求”。这时，协调者没有对“消费组元数据”进行加锁，第三个消费者发送“加入组请求”可以先于第一个消费者重新发送“加入组请求”。等到协调者收到第一个消费者重新发送的“加入组请求”，延迟操作可以完成，消费组中就有3个消费者了。

第一种场景“3个消费者同时发送加入组请求”有以下几个特点。

*  每个消费者再平衡监听器的onParti.ti.onsRevoked()方法和onParti.ti.onsAssi.gned()方法都只调用一次。这说明3个消费者加入消费组，总共只发生一次再平衡操作，并且只分配一次分区。
 *  第一个消费者和第二个消费者都会调用prepareRebalance()方法，它们都将消费组状态改为“准备再平衡”。第一个消费者更改前的状态是“稳定”，第二个消费者更改前的状态是“等待同步”。
 *  在这一次再平衡操作中，消费组的状态变化依次是：稳定→准备再平衡→等待同步→准备再平衡→等待同步→稳定。最后关闭应用程序，消费组状态为“离开”。
 *  第一个消费者发送了两次“加入组请求”，第二个和第三个消费者各向发送了一次“加入组请求”。

1.  3个消费者，两次再平衡操作  
    第二种场景：第一个消费者启动后，过一秒启动第二个消费者。第二个消费者启动后，再过一秒启动第三个消费者。这么做的目的是让第一个消费者收到“加入组响应”保证它发送“同步组请求”、收到“同步组响应”之前，都不会有新消费者加入消费组(睡眠了一秒，足够第一个消费者来完成这些操作)。

第一种场景中消费组状态还是“等待同步”，第一个消费者还没发送“同步组请求”。第二个消费者发送“加入组请求”，会等待第一个消费者重新发送“加入组请求”。第二种场景下第一个消费者发送了“同步组请求”，并收到“同步组响应”，更改消费组状态、为“稳定”。在这之后，第二个消费者才发送了“加入组请求”，和第一种场景一样，同样要完成延迟操作，协调者会等待第一个消费者重新发送“加入组请求”。相关日志如下：

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

第一个消费者调用了两次onPartlti.onsRevoked()和onParti.ti.onsAssi.gned()方法，说明第一个消  
费者参与了两次再平衡操作。第二个消费者和第三个消费者都只调用了一次回调方法，说明它们只参  
与了一次再平衡操作。第一次再平衡操作只有第一个消费者，所以分配结果是3个分区都分配给了第  
一个消费者。第二次再平衡操作有3个消费者参与，所以每个消费者都分配到了一个分区。

在第二个和第三个消费者调用onParti.ti.onsRevoked()方法，协调者还没有开始处理它们的请求之前，即还没有开始第二次再平衡操作，第一个消费者因为在第一次再平衡操作分配到了分区，所以它可以拉取到消息(Message\_Z)。但是当协调者开始处理第二个和第三个任何一个消费者的请求时，开始第二次再平衡操作，第一个消费者就不能再拉取消息了。因为消费组状态被更新为“准备再平衡”，第一个需要重新发送“加入组请求”才能参与第二次再平衡操作。第二种场景消费组的状态变化过程：稳定→准备再平衡→等待同步→稳定→准备再平衡→等待同步→稳定。

下面是第二种场景的服务端日志，它和第一种场景的区别是发生了两次再平衡操作，具体步骤如下。

(1)协调者处理consuf"ler－1的加入组请求，并且完成第一次再平衡，分配到分区。
    (2)协调者处理consuf"ler－2的加入组请求，消费组状态为“准备再平衡”，延迟操作不能完成。
    (3)协调者处理consuf"ler－3的加入组请求，延迟操作还不能完成。
    (4)consuf"ler－1重新发送加入组请求，延迟操作完成，返回“加入组响应”给所有消费者。
    (5)consuf"ler－1是主消费者，它发送“同步组请求，，，返回“同步组响应，，主合consuf"ler－1。
    (6)consumer-2和consumer-3是普通消费者，协调者依次处理并返回“同步组响应”。
    (7)关闭应用程序，消费组状态为“离开”，3个消费者都会从消费组中移除。

相关日志如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]

1.  3个消费者，3次再平衡操作  
    第三种场景：第一个消费者启动后，过一秒启动第二个消费者，再过两秒启动第三个消费者。在第二个消费者加入消费组后，设置两秒的睡眠时间，是为了保证前面两个消费者完成第二次再平衡操作之前，暂时不让第三个消费者提前加入消费组。这种场景相比第二种又多了一次再平衡操作，即每个消费者加入消费组，都会触发一次再平衡操作。第一个消费者经历了3次再平衡，第二个消费者经历了两次再平衡，第三个消费者经历了一次再平衡：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]

从第三种场景服务端的日志也可以看刷，除去关闭应用程序，3个消费者依次力I.I人消费组，总共发生3次再平衡操作。消费组的状态变化过程：稳定→准备再平衡→等待同步『稳定(第一次再平衡结束)→准备再平衡→等待同步→稳定(第二次再平衡结束)→准备再平衡→等待同步→稳定(第三次再平衡结束)。具体步骤如下。

(1)协调者处理consul'ler－1的加入组请求，并且完成第一次再平衡，消费组状态为“稳定”。
    (2)协调者处理consul'ler－2的加入组请求，消费组状态为“准备再平衡”，延迟操作不能完成。
    (3)consul'le卜1重新发送加入组请求，延迟操作完成，返回“加入组响应”给两个消费者。
    (4)两个消费者都发送了“同步组请求”，协调者返回“同步组响应”给两个消费者，状态为“稳定”。
    (5)协调者处理consul'ler－3的加入组请求，延迟操作不能完成，等待前两个消费者重新加入组。
    (6)consul'ler－1和consul'ler－2重新发送“加入组请求”，延迟操作完成，状态为“等待同步”。
    (7)所有消费者都发送了“同步组请求”，协调者返回“同步组响应”给所有消费者，状态为“稳、定”
    (8)关闭应用程序，消费组状态为“离开”，3个消费者都会从消费组中移除。

相关日志如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]

上面是3个消费者按照不同顺序加入消费组的例子，其中消费组的状态变化也不一样。要理解不同消费者按照不同顺序加入消费组后，协调者处理请求的执行顺序，需要理解下面这些概念。

(1)协调者处理请求时，对消费组元数据加锁保护，保证不会同时处理多个请求。
    (2)消费组状态为“稳定”或“等待同步”，都可以转为“准备再平衡”，并创建延迟的操作。
    (3).消费组状态为“准备再平衡”时，主消费者发送“同步组请求”的分配结果是无效的。
    (4)如果消费组中有消费者，但它们还没有重新发送加入组请求，延迟操作不能完成。
    (5)一旦消费组中所有消费者的回调方法不为空，延迟操作可以完成，并返回加入组响应。

本节的3种场景示例除了第一种的消费组状态变化没有规律，后面两种消费组状态变化都很有规律。

*  第一种，稳定→准备再平衡→等待同步→准备再平衡→等待同步→稳定。
 *  第二种，两次再平衡操作都是：稳定→准备再平衡→等待同步→稳定。
 *  第三种，3次再平衡操作都是：稳定→准备再平衡→等待同步→稳定。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/477ef020b3044dd890ee1984720de806.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/c2dd7eb0301c41aba606829394ab7530.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/6b730f2002214296b54a1681584d6dd2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/bd5f254a43074716b2b6a04c109cee77.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/df8e3efc1aee44179d8617a4f64bcbae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/d5d1cd7f34df4f63bf64e8b9dd3c685a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/429aa26cfe644165b65c0f229b675921.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]: /images/20221022/8572aabab9524d059af953dda1d2f5e8.png