5.1.3 主消费者执行分配任务

一时失言乱红尘 2023-01-13 15:57 65阅读 0赞

#### 5.1.3 主消费者执行分配任务 ####

> 消费者发送的“加入组请求”(Jo1nGroupRequest)的内容包括：消费组编号、消费者成员编号、协议类型、协议内容和元数据(protocolMetadata)。其中，协议内容是分区分配算法的名称，元数据是消费者订阅的主题列表。“加入组响应”对象的内容包括：消费者成员编号、统一的消费组协议、主消费者编号、协调者执行分区分配工作的次数、消费者成员列表。

*  客户端发送的协议与服务端返回的“消费组协议”(groupProtocol)。虽然“加入组请求”中的“协议名称”包括了系统支持的所有协议类型(范围分配和循环分配)，但且正执行具体的分区分配时只允许一种协议。协调者会负责统一所有消费者的协议，选择一个大家都支持认可的协议作为“消费组协议”。协调者发送“加入组响应”给每个消费者的“消费组”协议都是一样的，虽然只有主消费者会使用这个协议来做实际的分配t作。
 *  消费者成员编号(mel’\\berId)。消费者发送的“加入组请求”需要指定消费者成员编号，当消费者初次加入消费组时这个编号是UNKNOWN\_MEMBER。协调者处理每个消费者发送的“加入组请求”，会为每个消费者指定唯一的消费者成员编号，并包含在“加入组响应”中运回给消费者。后续消费者需要重新加入消费组时，发送“加入组请求”巾的消费者成员编号，就是协调者之前分配给它的编号。
 *  主消费者编号(leaderId)。协调者选择的主消费者编号，如果消费者的成员编号和主消费者编号相等，那么这个消费者就是主消费者。
 *  所有消费者成员信息C111e111bers)。协调者会将它收集到的所有消费者信息，都发送给主消费者。注意：111e111ber5不仅包括所有的消费者成员编号，还包括每个消费者订阅的主题。必须要包含订阅信息，否则只有消费者成员编号，不知道订阅了哪些主题，主消费者还是无法执行分区分配工作。
 *  纪元编号(generat-Lon)。只在每个消费者每次需要重新加入组时，才会在协调者端进行更新，它表示协调者从启动至今一共发生了多少次分区分配的工作。每次消费组发生再平衡操作时，协调者都会发起一次分区分配的工作。虽然分区分配工作是由主消费者执行的，但主消费者有可能变化，所以要由服务端的协调者来记录这个编号。

普通消费者收到“加入组响应”会调用onJoinFollower()方法，立即发送“同步组请求”给协调者，并给返回的“同步组”异步请求链接上“加入组”的异步请求。当消费者收到“同步组响应”后，会完成“同步组”的异步请求，再完成“加入组”的异步请求，这样普通消费者就可以从“加入组”  
的异步请求结果中获取分配给它的分区。主消费者在收到“加入组响应”时会调用。『1Joinleader()方法，也会发送“同步组请求”给协调者。它也会给返回的“同步组”异步请求链接上“加入组”异步请求，后续流程和普通也费者类似，分别是：收到“同步组响应”、完成“同步组”异步请求、完成“加入组”异步请求、获取“加入组”异步请求结果。

消费者发送“同步组请求”(SyncGroupRequest)的内容包括：消费组编号、纪元编号、消费者成员编号、消费组的分配结果。其中前3个信息都在协调者返回给消费者的“加入组响应”结果中，“消费组的分配结果”只有主消费者会传递。主消费者在收到“加入组响应”后，并不会立即发送“同步组请求”给协调者，而是要等到执行分区分配的工作完成后才发送“同步组请求”。主消费者发送的“同步组请求”带有“消费组的分配结果”(groupAssign111ent)，普通消费者发送的“同步组请求”没有分配结果，因为它并没有执行分区分配工作。

1.  执行任务获取消费组的分配结果

主消费者调用perfor111Assign111ent()方法执行分区分配工作。其中，members是所有消费者的订阅信息，它的键是每个消费者的成员编号，值是消费者的订阅信息。这个数据是由协调者在收集完所有消费者发送的订阅信息后，作为“加入组响应”传给主消费者的。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

主消费者在执行分区分配工作时，根据每个消费者发送的订阅信息，会通过分区分配器的assign()方法计算出每个消费者分配的分区结果。如表5-1所示，这个过程产生了一些字典数据结构，它们的键都是字符串类型的消费者成员编号，值的类型不同，表示的含义也不同。后两个的值都表示每个消费者的分区分配结果，不同点是前者是一个对象，后者是序列化的字节数组。  
![在这里插入图片描述][20210412204720186.png]

1.  抽象分区分配器类

“抽象分区分配器类”(AbstractPartitionAssignor)实现了“分区分配器接口”(PartitionAssignor)自"Jassign()分区分配方法，但它又定义了一个参数类型不同的assign()抽象方法。这样具体的“分区分配实现类”不需要实现“分区分配器接口”，只需继承并实现抽象类的assign()方法。这个方法的两个参数的含义如下，它和“分区分配器接口”的两个参数类似，但做了一点处理。比如，不使用“集群元数据”，而是从集群元数据获取主题对应的分区数；不使用“订阅信息对象”，而是使用订阅信息的订阅主题列表。

*  partitionsPerTopic。每个主题有多少个分区。从传入分区分配器的集群元数据获取数据。
 *  subscriptions。每个消费者的订阅主题。和分区分配器的subscriptions类似，但把订阅信息对象(Subscription)转为字符串列表，实际上订阅信息就是由主题列表组成的。

相关代码如下：

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

注意：具体分区分配器的实现类和1日API的算法类似(详见322节)，这里不再分析。图5-4总结了消费者发送“加入组请求”给协调者，到获取到分区列表的过程，具体步骤如下。

(1)消费者发送“加入组请求”，得到一个“加入组”的异步请求。
    (2)消费者获得“加入组响应”结果，表示协调者已经收集到所有发送了“加入组请求”的消费者。
    (3)主消费者会执行分区分配任务，返回结果是消费组中所有消费者及其对应的分区列表。
    (4)每个消费者都会发送“同步组请求”，得到一个“同步组”的异步请求。
    (5)每个消费者获得“同步组响应”结果，表示分配给当前消费者的分区列表。
    (6)完成“同步组”的异步请求，并通过模式完成“加入组”的异步请求。
    (7)消费者获取“加入组”异步请求的结果，这个数据表示的就是分配给消费者的分区。

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/4186ef1718b443b1ae9d202461636904.png
[20210412204720186.png]: /images/20221022/5b7251238afb4ca7ba816fa46faec836.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/a8a7cce3633f45f082c51fa8878f0dc2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/b0999f91ad74415a9607f84df9f9c326.png