5.2.2 消费者和消费组元数据

向右看齐 2023-01-13 15:57 88阅读 0赞

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

#### 5.2.2 消费者和消费组元数据 ####

> 消费者加入组过程发送的“加入组请求”和“同步组请求”，都会指定消费组编号(groupid)和消费者成员编号(l’lel’lberId)，同一个消费组编号只对应一个“消费组元数据”(GroupMetadata，下文简称“组元数据”)。服务端使用“消费者成员元数据”(Mel’lberMetadata，下文简称“成员元数据”)表示每个泊费者发送的元数据信息，并添加到对应的“组元数据”中。

注意：协调者处理“加入组请求”和“同步组请求”，不需妥为每种请求都定义一个“成员元数据”。协调者只用了一个统一的“成员元数据”表示“这个消费者在加入组过程中，在服务端保存的相关状态数据”。

1.  消费者成员元数据

“成员元数据”类的构造函数参数和消费者发送的“加入组请求”数据是一样的，比如都有成员编号、消费组编号、协议元数据集。会话超时时间也是由客户端发送“加入组请求”时指定的，latestHeartbeat变量记录了该消费者最近一次发送心跳的时间。另外，“成员元数据”最重要的一个信息是：当前这个消费者到底分配到了哪些分区(assignl’lent变量)，因为消费者加入组的最终目的就是从协调者获取到分区。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

“成员元数据”还定义了两个值对象，它们分别对应服务端在处理请求时定义的两个发送响应回调方法：  
![在这里插入图片描述][20210412212436982.png]

这两个值对象使用var定义为“成员元数据”的一个变量，而不是用def定义为一个方法。因为处理“加入组请求”的回调方法参数是Joi叫roupResult，没有返回值，所以awaitingJoinCallback值对象的类型是：JoinGroupResult=>Unit。值对象除了赋值操作和普通的成员变量一样，我们还可以把值对象当作一个方法，传递正确的参数即可。比如下面的代码中要把值对象awaitingJoinCallback当作方法使用，传递的参数必须是JoinGroupResult:

![在这里插入图片描述][20210412212457252.png]

为了方便将“发送响应回调方法”作为消费者“成员元数据”值对象的一个变量，可以把回调方法定义成一个高级的类型。这样看起来，值对象的使用方式其实和普通的类型类似。相关代码如下：

![在这里插入图片描述][20210412212507390.png]

KafkaApis虽然在一开始处理请求时就定义了sendResponseCallback()回调方法，但是因调方法只能通过上面那种“传递参数给调用值对象，把值对象当作方法使用”的方式调用。一旦调用回调方法，就表示服务端会发送响应结果给客户端，说明服务端已经处理完客户端发送的请求。

注意：在调用完值对象的回调方法(倒数第二行)后，妥重直值对象为空(最后一行)。不过，虽然“成员元数据”的值对象为空，但“成员元数据”仍然在“组元数据”中。消费者一旦发送一次“加入组请求”，就会一直被协调者对应的“组元数据”管理(除非这个消费者没有及时发送心跳给协调者而被移除掉)。

1.  消费组元数据

一个“组元数据”管理了所有消费者的“成员元数据”。消费组协调者对象的addM阴阳rAndRebalance()方法表示“组元数据”中没有该消费者，需要添加消费者的“成员元数据”。updateMel’lberAndRebalance()方法表示“组元数据”中已经有该消费者的元数据，只需做更新。

这两个方法都传递了“组元数据”对象(GroupMetadata)，说明必须先有“组元数据”。如果添加“成员元数据”时都还没有“组元数据”(更新时一定存在“组元数据”)，就会先创建“组元数据”。创建“组元数据”是必须的，如果没有“组元数据”，即使有“成员元数据”，也是没有意义的。相关代码如下：

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

注意：添加或更新消费者的成员元数据，都会执行消费纽状态相关的maybePrepareRebalance()方法。

“组元数据”在消费者需要加入或更新时，除了更新对应消费者的“成员元数据”，还会记录一些其他数据。比如，协调者会为消费组选择一个主消费者，来代替它执行分区分配工作。另外，每个消费者发送“加入组请求“时，都会指定一个会话超时时间。协调者会从消费组的所有消费者中，选择一个最大的会话超时时间，作为”再平衡操作的超时时间“。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

注意：notYetRejoinedMef!lberS方法会获取消费纽元数据中awaitingJoinCallback值对象为空的消费者成员。正常来说，协调者刚开始处理消费者的“加入组请求”时，会设直“成员元数据”的awaitingJoinCallback为事先定义的“发送响应回调方法”，这时候该方法不会收集到满足条件的成员。但如果消费者成员已经存在于消费组的元数据中，而且值对象被更新为空，就会被选择出来。

消费组元数据中还有一个很重要的数据：“消费组的当前状态”。因为每个消费者加入消费组都分成“加入组”和“同步组”两个步骤，所以协调者在处理不同消费者的这两种请求时，都需要改变消费组的状态。消费组元数据的状态机有4种状态：“稳定状态”(Stable)、“准备再平衡状态”(preparingRebalance)、“等待同步状态”(AwaitingSync)、“离开状态”(Dead)。协调者新创建一个消费组元数据，这个消费组元数据的初始状态为“稳定状态”。

注意：“稳定状态”分成两种，一种是刚刚创建消费纽元数据，并且消费组元数据中还没有管理任何一个消费者成员。另一种是完成一次分区分配工作后，消费组管理了所有的消费者，这些消费者都分配到了分区，也进入了稳定状态。前者的“稳定”表示没有消费者，也就不会为消费者分配分区。后者的“稳定”表示已经有消费者，并且也都分配到了分区。注意：消费组元数据的状态代表整个消费组级别。

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/eee2b87ed8df4a66b374b0d0201f0482.png
[20210412212436982.png]: /images/20221022/ab0e1e492e3841da82f625b5fb894017.png
[20210412212457252.png]: /images/20221022/ba7e7d7e49bf448cb5d90083d751987b.png
[20210412212507390.png]: /images/20221022/e4d36221cc2f46d38e5c1b6e67c58419.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/f89f17671ba247e0b76b6ebfe5e635f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/d31f9137dba84aa4affcc187ea008dd5.png