5.1.2 消费者的加入组和同步组

素颜马尾好姑娘i 2023-01-13 15:56 88阅读 0赞

### 5.1.2 消费者的加入组和同步组 ###

> 消费者向协调者发送“加入组请求”获取分区和现实生活中的任务分配很相似。为了帮助理解分区分配的过程，我们以软件开发常见的任务分士为例如图5-1所示。公司要开发一个新项目，有3个开发人员向项目经理申请开发任务；项目经理执行任务分配，将项目的3个模块分别交给3个开发人员。这个示例中开发人员是消费者，项目经理是协调者，项目有哪些模块是集群的元数据，项目经理分配任务是分区分配算法，项目经理将不同模块分给不同的开发人员等价于不同分区分配给不同消费者。将这个示例对应到消费者分区分配的流程，具体步骤如下。

(1)消费者发送订阅信息给协调者。
    (2)协调者收集所有的消费者，以及它们对应的订阅信息。
    (3)协调者执行任务分配算法，即具体如何将不同的分区分配给不同的消费者。
    (4)分配结果确定后，协调者将分区返回给消费者，消费者分配到分区开始t作。

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

在Kafka的实现中，协调者在收集完所有的消费者及其订阅信息后，并不执行具体的任务分配算法，而是交给其中一个消费者执行分区分配任务，这个消费者叫作主消费者。

注意：这里的主消费者只是负责执行分区分配，并没有其他相关的状态数据。协调者通常会将第一个发送“加入组请求”的消费者选作主消费者。但后续过程中，如果这个主消费者挂掉了，协调者就会选择下一个消费者作为主消费者。所以这里协调者选择主消费者的策略比较简单，不像分区的主消费者选举会有很多的依赖、限制条件。

1.  主消费者执行任务分配

> 如图5-2所示，我们仍然以前面的示例为例，项目经理会选择一个开发人员作为组长(主消费者)，让组长帮他执行任务的分配t作。组长分配完任务后，再将分配结果告诉给项目经理。最后项目经理再将任务分配给每个开发人员。改进后的具体步骤如下。

(1)消费者发送订阅信息给协调者。
    (2)协调者收集所有的消费者，以及它们对应的订阅信息。
    (3)协调者将所有的消费者成员列表及其订阅信息发送给主消费者。
    (4)主消费者执行具体的分区分配算法。
    (5)主消费者将分配结果同步回协调者。
    (6)协调者收到主消费者的分配结果，将分区返回给每个消费者。

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

协调者选择主消费者帮它执行具体的分区分配算法，这种做法的好处是可以减少协调者本身的负  
担，但缺点是客户端计算完结果后，需要把分配结果同步回协调者，再由协调者将结果返回给各个消  
费者客户端。这种方式在实现上需要权衡和考虑下面4个设计上的问题。

*  协调者如何选择主消费者，是否需要依赖什么外部的条件？
 *  协调者选择主消费者执行分区分配，协调者要怎么处理主消费者失败的情况？
 *  协调者将“所有消费者成员及其订阅信息”作为“加入组请求”的响应结果分配给主消费者，是否需要发送“加入组响应”给其他普通的消费者？
 *  假设每个消费者都会收到“加入组响应”但这个响应结果并不表示分配结果，每个消费者如何得到分配给它们的分区，是否需要再次发送请求？

权衡结果分别如下。

*  协调者在选择主消费者上不需要做过多的判断，通常会选择第一个发送“加入组请求”的消费者作为主消费者。执行分区分配的算法交给任何一个消费者都是可以完成的。
 *  不管什么类型的消费者失败，都会遵照第4章心跳超时的处理方式。失败的消费者会被协调  
    者从消费组中移除，并触发再平衡操作，剩余存活的每个消费者都要重新加入消费组。
 *  每个消费者发送“加入组请求”后，协调者在收集完所有的消费者及其订阅信息后，会返回“加入组响应”给每个消费者，但这个响应结果并不是分区分配结果。
 *  每个消费者收到“加入组响应”后，都会发送“同步组请求”给协调者来获取分配的分区。主消费者会在完成分区的分配任务后才发送“同步组请求”。普通消费者会立即发送“同步组请求”，但因为主消费者还没有将分配结果返回给协调者，普通消费者的“同步组请求”在服务端会被延迟处理。协调者收到主消费者带有分配结果的“同步组请求”后，会将分配结果分配给每个消费者。

消费者客户端要发送两种类型的请求采用422节“4异步请求的链式调用”的异步请求连接模式来完成。

1.  发送“加入组请求”

> 如果是协调者负责分区的分配工作，消费者发送完“加入组请求”后，就可以从“加入组响应”中获取到分配给它们的分区。那么可以在发送“加入组请求”返回的异步请求对象上，使用组合模式加上一个适配器。在适配器的回调方法中，解析“加入组响应”中分配的分区，完成异步请求对象，最后协调者就可以获取异步请求的结果，这个结果就是分配给消费者的分区。相关代码如下：

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

但因为协调者并不执行分区分配，所以它返回的“加入组响应”没有分配结果。协调者返回给每个消费者的“加入组响应”是不同的，主消费者收到的是“所有消费者成员列表及其对应的订阅信息”，而普通消费者并没有这些数据，因为普通消费者并不会执行分区分配的工作。

“加入组的响应处理器”继承了“协调者响应处理器”(CoordinatorResponseHandler)，它本质上是一个用于组合模式的异步请求适配器。在回调方法中，由于消费者接收的“加入组响应”不是分配的分区，所以不能直接完成“加入组”的异步请求，而应该再次发送“同步组请求”。

但如果“加入组响应”有错误，就不需要继续发送“同步组请求”，而应该对“加入组”的异步请求调用ra"ise()方法，表示“加入组”的异步请求有异常。客户端轮询完成，但异步请求没有成功，就不会执行。『iJoinC0111plete()回调方法，消费者需要重新发送“加入组请求”。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

不同的消费者会根据“加入组响应结果”的isleader()调用不同的回调方法，但是链式模式链接的异步请求都是“加入组”的异步请求。

注意：消费者客户端要获取分配给它们的分区，这个数据会存放在“加入组”的异步请求中。一旦调用“加入组”异步请求的co111plete()方法完成它，就表示“加入组”的异步请求完成，消费者就可以从异步请求中获取数据，得到分配给它们的分区。注意，“加入纽请求”指的是JoinGroupRequest，而“加入组”的异步请求指的是“加入纽请求”的异步结果对象。消费者调用圭i\]JoinGroupResponseHandler的回调方法，虽然表示消费者收到“加入纽响应”，但这时“加入组”的异步请求还没有到被完成的火候。

1.  发送“同步组”请求

> 普通消费者在收到“加入组响应结果”后，会立即发送“同步组请求”给协调者。而主消费者在收到“加入组响应结果”后，会从“加入组响应结果”中获取执行分区分配过程中需要用到的数据，然后调用perfor111Assign111ent()执行分区分配。只有这个执行过程完成后，主消费者才会开始发送“同步组请求”给协调者。相关代码如下：

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

“加入组响应处理器”的回调方法调用时，表示消费者收到“加入组响应结果”，但“加入组”的异步请求还没有完成。“同步组响应处理器”的回调方法调用时，表示？肖费者收到“同步组响应结果”由于“同步组响应结果，，表示的数据就是分配给消费者的分区信息，所以可以完成“同步组”的异步请求，并一起完成了“加入组”的异步请求。这时消费者读取“加入组”异步请求的结果就是分配给消费者的分区信息。如图5-3所示，消费者获取协调者的分配结果，总体上可以分为下面4个步骤。

(1)每个消费者都发送“加入组请求”给协调者节点。
    (2)协调者收到所有消费者发送的“加入组请求”，返回“加入组响应”给每个消费者，还会将执
    行分区分配算法需要的数据(比如消费者成员列表)传给主消费者。
    (3)主消费者执行完分区分配算法后，将“分配结果”通过“同步组请求”的方式发送给协调者
    节点。其他消费者也会发送“同步组请求”给协调者，但是它们的请求中并没有“分配结果”数据。
    (4)每个消费者从“同步组响应”的结果数据中获取到分配给它们的分区。

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

在“加入组”和“同步组”两种请求的“回调处理器”中，回调方法的第二个参数即异步请求，表示的分别是“加入组”和“同步组”的异步请求。“加入组响应处理器”没有完成“加入组”的异步请求，“同步组响应处理器”则完成了“同步组”的异步请求。那么，“加入组”的异步请求在什么时候完成呢？实际上，通过调用“加入组”异步请求的链接方法，将“同步组”异步请求的结果设置为“加入组”异步请求的结果，从而完成“加入组”的异步请求。这个过程更详细的步骤如下。

(1)每个消费者都向协调者发送“加入组请求”，申请加入消费组。
    (2)协调者接收每个消费者的“加入组请求”，收集消费组的消费者成员列表。
    (3)协调者选举一个消费者客户端作为主消费者。
    (4)协调者向发送“加入组请求”的每个消费者返回响应结果，其中包含所有消费者成员列表。
    (5)步骤(3)的主消费者会做额外的分区分配算法，并在计算完成后发送同步组请求给协调者。
    (6)普通消费者收到包含成员列表的“加入组响应”结果不做计算，立即发送同步组请求给协
    调者。
    (7)不管是主消费者还是普通消费者，发送同步组请求的目的都是向协调者申请分区。
    (8)协调者收到主消费者在步骤(5)的分配结果，向每个发送同步组请求的消费者返回分区。

“加入组请求”和“同步组请求”以及相关的异步请求、回调处理器的调用流程在4.2.2节“4异步请求的链式调用”中分析过了，这里就不再赘述。下面总结了这两个i1’T求调用流程相关的特点。

*  客户端收到服务端返回的晌应结果后，才会调用回调处理器的回调方法。比如，主消费者客户端收到的“加入组响应结果”除了消费者的成员编号，还有所有消费者的成员信息及其订阅信息等；普通消费者客户端收到的“加入组响应结果”包含了成员编号。不管是主消费者，还是普通消费者，它们收到的“同步组响应结果”都是分配给消费者的分区。
 *  使用组合模式返回的异步请求对象，只有调用其完成方法才表示异步请求已经完成。通常，在响应处理器的回调方法中完成异步请求但如果异步诸求还没到完成的时候(一般是不能从响应结果中获取出异步请求需要的结果数据)，就不能调用其完成方法。比如，“川人组响应处理器”中没有完成“加入组”的异步请求，“同步组响应处理器”则完成了“同步组”的异步请求。“加入组”异步请求的完成是在“同步组”异步请求的完成之后，调用“同步组”异步请求的链接方法才被完成的。
 *  “加入组请求”先于“同步组请求”发送给协调者。消费者接收到“如l入组响应”后，才会发送“同步组请求”。“加入组”的异步请求也先于“同步组”的异步请求产生，但“‘加入组’的异步请求却后于‘同步组’的异步请求完成”，即“同步组”的异步请求完成后，才完成“加入组”的异步请求。

这里分析了客户端发送“加入组请求”和“同步组请求”的调用流程，下面分析相关的业务逻辑处理。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/cdca1cf3d5d04d3fa5746ca1f9dfed4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/a526ebf8fb8c46119625c14ffdc93ba0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/c85e810eb56f47038b1b0b2f79d4b3f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/d1f0188a6f134a4fa120f90f132923ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/f5685e624a674838ab71109aa3b73cd4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/22d48baab53441f185d8bdc8564690cf.png