5.1.4 加入组的准备、完成和监昕器

我会带着你远行 2023-01-13 15:57 74阅读 0赞

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

#### 5.1.4 加入组的准备、完成和监昕器 ####

> 消费者重新加入消费组，在分配到分区的前后，都会对消费者的拉取工作产生影响。消费者发送“加入组请求”之前要停止拉取消息，在收到“加入组响应”中的分区之后要重新开始拉取消息。同时，为了能够让客户端应用程序感知消费者管理的分区发生变化，在加入组前后，客户端还可以设置自定义的“消费者再平衡监听器”，以便对分区的变化做出合适的处理。

1.  准备和完成“加入组请求”

消费者发送“加入每请求”给协调者，最终从协调者获取到的分配结果对象(Ass"ignment)表示分配给消费者的分区列表。消费者在加入消费组过程中调用onJo"inPrepare()方法，这表示消费者正准备加入消费组，正在等待分配分区。此时拉取器应该暂停拉取消息，而只有等消费者分配到分区，并将最新的分配结果更新到订阅状态中后，拉取器才可以开始发送拉取请求并拉取消息。相关代码如下：

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

下面将消费者加入组分配到分区的操作和拉取操作一起分析。因为客户端轮询采用的是cHentpoll(future)方式(详见423节)，所以执行“加入组分配到分区”会阻塞主流程：如果“加入组”操作没有完成，后续的流程都不会执行。可以看到，消费者重新加入消费组执行的分区分配作，为后续的拉取器拉取分区消息提供了数据来源。这两个动作通过消费者的订阅状态关联起来：消费者加入组完成后，将分区设置到订阅状态中，拉取器工作时获取订阅状态的分配结果，然后开始拉取消息。相关代码如下：

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

消费者在加入组的前后会对其他相关组件产生影响，并不仅仅是发送“加入组请求”，然后获取到分配的分区结果就结束了。比如，在加入组之前，需要执行下面两个操作。

(1)禁用向动提交任务，因为在加入组过程中不会拉取和消费新消息，所以没必要提交偏移量。
    (2)执行一次同步提交偏移盘，这个操作是阻塞的，确保提交偏移量能够成功完成。

在加入组之后，也要执行下面几个操作。

(1)更新订阅状态的needsFetchCorlmittedOffsets变量，表示需要刷新分区的提交偏移量。
    (2)更新订阅状态的分配结果，为每个分区新创建分区状态，这个对象用来记录分区的最新状态。
    (3)启动消费者的向动提交任务。

如图5-5所示，消费者加入消费组、拉取前的准备工作、拉取消息这3个步骤都在消费者的轮询操作中完成。加入组之前需要先采用同步方式提交分区偏移量给协调者。！V取准备工作会先从协调者获取分区的提交偏移量，然后更新分区的拉取偏移量，使消费者的拉取消息Tj年可以正常开始。

举例，消费者l在加入组之前分配到分区PO，它提交的偏移量为5；消费者2提交分区Pl的偏移量为10。两个消费者都重新加入组后，消费者l分配到分区Pl，消费者2分配到分区PO。通过拉取前的准备工作，消费者l从协调者获取分区Pl的提交偏移量为10，消费者2从协调者获取分区PO的拉取偏移量为5，接着它们各自从分区对应的最近消费位置开始拉取消息。

消费者重新加入消费组，会导致分配给消费者的分区发生变化。除了在消费者内部可以更新订阅状态外，客户端应用程序也可以定义一个“消费者再平衡监听器”。

说到再平衡操作，“再平衡操作、消费者重新加入消费组、分配分区”，这3个概念互相关联。再平衡操作发生时，消费者需要重新加入消费组，并从协调者获取分配的分区。通常，消费者的再平衡也意味着消费组的再平衡，这时消费组中的所有消费者都要再平衡。有下面几种事件触发再平衡操作。

*  消费者订阅的主题集合中任意一个主题的分区数量发生变化。
 *  创建或删除一个主题。
 *  消费组中已经存在的一个消费者成员挂掉了。
 *  一个新的消费者成员加入已经存在的消费组中。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

1.  消费者平衡监昕器

消费者在发送“加入组请求”之前，调用onJoinPrepare()方法会触发“消费者再平衡监听器”的onParttitionsRevoked()方法，在加入消费组后调用。onJoinComplete()方法会调用监听器的onParttitionsAssigned()。这两个方法的参数都是分区，前者是加入组之前分配的分区，后者是加入组之后分配的分区，所以这两个分区参数值会不一样。相关代码如下：

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

注意：消费者发送完“加入纽请求”，在协调者还没有返回结果之前，消费者之前拥有的分区仍然存在于订阅状态的分配结果中，即旧的分区并不会被真正删除。当消费者分配到新分区后，会先清空旧的分配结果，然后将新的分区添加到分配结果中。这里不需要在加入纽之前清空l日分配结果，因为分配到新分区之前，消费者的拉取操作会被阻塞，这时订阅状态中即使存在旧的分区也不会起作用。

“消费者再平衡监昕器”只适用于订阅模式的消费者API，如果使用手动分配分区模式，监昕器不会起作用。因为消费者如果指定消费固定的分区就不需要再平衡操作。使用向定义“消费者再平衡监昕器”的典型场景是：发生再平衡操作时，保存偏移量到外部存储系统中。下而这个示例和443节“保存偏移盘到文件系统”的原理是类似的。

消费者撤销分区时调用onPartitionsRevoked()方法，会通过consuf11erposition()方法获取分区的最近位置(这个变量实际上既是拉取偏移量，也能作为提交偏移I量)，然后保存到外部存储系统。消费者分配到新分区时调用onPartitionsAssigned()方法，会读取外部存储系统的分区偏移盘，然后调用consuf11erseek()方法定位到最近的位置。

自定义监昕器实现的“再平衡时保存偏移量”这个功能，实际上消费者内部已经有了，只不过它是保存到协调者节点。消费者发送“加入组请求”前执行一次同步模式的提交偏移盘操作，等价于监听器的“保存偏移量到外部存储系统”。消费者在拉取消息之前更新分区状态的拉取偏移盐，也等价于监昕器的“读取外部存储系统的偏移量，并定位到指定位置”。相关代码如下：

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

客户端应用程序向定义的“消费者再平衡监听器”一般只和IKafkaConsumer类交五。KafkaConsuf11er除了为消费者提供轮询方法，还有下面几个和偏移量相关的方法。

*  同步和异步提交偏移量：CommititSync(offsets)、CommititAsync(offsets)。
 *  获取分区的拉取和提交偏移量：position(partition)、cof111’1itted(partition)。
 *  定位到指定的位置，更新拉取偏移量：seek(partition,position)。

表5-2总结了消费者发送的请求和接收的响应结果数据内容。消费者发送“加入组请求”和获取“同步组响应”的数据是一样的，但获取“加入组响应”和发送“同步组请求”的数据则不同。主消费者需要从“加入组H向应”中获取所有的消费者成员信息，才能执行分区分配任务，并且它发送的“同步组请求”有“消费组的分配结果”。而普通消费者不会从“加入组响应”中获取成员列表，发送“同步组请求”时也没有“消费组的分配结果”这种数据。表中括号部分表示主消费者额外的数据。

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

主消费者要从“加入组响应”中获取消费者成员列表信息，而每个消费者都会发送向己的成员信息给协调者，所以看起来，协调者只需要“收集”每个消费者发送的成员信息，然后发给主消费者就可以了。而且分区分配丁，作也在主消费者客户端完成，协调者同样也只需要“接收”主消费者发送的“消费组的分配结果”就可以了。这样看来，协调者做的工作并不是很复杂。但实际上，协调者的工作不仅仅是收集和接收，它还要对所有消费者发送的两种请求进行协调。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/b0749f060e25473a9e48c4e7cd85346f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/1504478fb1684efaa0f45265d79c1c34.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/07cd6307c6a646568a89697d0fbfcec5.png
[20210412204956663.png]: /images/20221022/4e23a6658df949e7a2d99473f8d8a91a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/1b8287d9c81942d79410e09319399201.png
[20210412205044106.png]: /images/20221022/ace1e09bf1de49ada2d2bd036df5a3eb.png