5.2.4 协调者调用回调方法发送晌应给客户端

傷城~ 2023-01-13 15:57 52阅读 0赞

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

#### 5.2.4 协调者调用回调方法发送晌应给客户端 ####

> 在5.2.2节，消费者成员元数据定义了两个值对象：awaitingJo1nCallback和awaitingSyncCallback。我们也分析了把值对象当作方法调用时，实际上会调用KafkaApis中事先定义的“发送响应的回调方法”，说明协调者处理完请求，发送“响应结果”给客户端。

注意：下面的方法会涉及“消费组的状态”，但都只会粗略地一笔带过，后面会详细分析状态机的实现。

1.  发送“加入组响应”给消费者

协调者要返回“加入组响应”给消费组下的所有消费者之前，会增加纪元编号 、 选择 H·＼一个统一的消费组协议 、 将消费组状态更改为“等待同步” 。 因为是要返回“加入组响应”，而消费者成员元数据的 awaltlngJolnCallback值对象保存了 “发送响应的回调方法”，所以只要用调用方法的方式调用awaltlngJolnCallback值对象，就可以调用到“发送加入组响应的回调方法” 。 相关代码如下 ：

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

消费组管理了所有的消费者成员，协调者发送“加入组响应”时，是一次性一起发送响应结果给每个消费者。但是协调者处理消费者发送的“加入组请求”并不是同时进行的，这说明协调者在处理某些消费者的“加入组请求”时，并不会立即返回“加入组响应”。实际上，这是通过“延迟操作”来实现的，“延迟操作”类似于延迟的任务，它和消费组的状态机也有关系。

注意：协调者在发送“加入组响应”之前就更新消费组状态为“等待同步”。另外，协调者往每个消费者发送响应结果之后，就立即针对该消费者启动服务端的心跳监控。

1.  发送“同步组响应”给消费者

普通的消费者在收到“加入组响应”后，会立即发送“同步组请求”。但主消费者只有执行完分区分配工作后，才会发送“同步组请求”。协调者没有同时处理每个消费者的“同步组请求”，但最后同时发送了“同步组响应”。说明协调者在处理某些消费者的“同步组请求”时，并不会立即返回“同步组响应”。

当协调者收到主消费者的“同步组请求”后，它会立即返回“同步组响应”给所有的消费者(包括主消费者和普通消费者)。协调者发送“同步组响应”给消费组每个消费者的方式和发送“加入组响应”类似，把调用的值对象改为awaitingSyncCallback即可。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

注意：协调者将“加入组响应”返回给每个消费者后，都会重直消费者“成员元数据”的awaitingSyncCallback为空，并立即针对该消费者启动服务棉的心跳监控。任何涉及协调者和消费者的网络通信，服务端的协调者节点都会启动一次新的心跳监控。

1.  协调者保存消费组任务

协调者在返回“同步组响应”给消费者之前，会先把“消费组分配结果，，(groupAssigne111nt)以普通消息的形式持久化到内部主题(\_consumer\_offets)中。如果协调节点出行问题需要进行故障迁移，新的协调者可以从“内部主题”中读取持久化的消息，重建“消费组分配结果”

一个协调者可以充当多个消费组的协调节点，并使用“消费组缓存”(groupsCache)保存它管理的所有“消费组元数据”。迁移协调者时会读取内部主题的“消费组分配结果”，重新加载到“消费组缓存”中。协调者处理“加入组请求”和“同步组请求”时，根据消费者客户端传递的消费组编号查询“消费组元数据”，会先从“消费组缓存”中查询，如果“消费组元数据”已经存在，直接使用现有的数据。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

协调者处理“同步组请求”时， 将发送“同步组响应结果”给消费者的逻辑，作为一个回调方法传给了“延迟存储”对象（ DelayedStore ）。 “延迟存储”由消息集和回调方法组成，协调者会先存储完“消费组分配结果” 代表的消息、集，然后才调用回调方法。 即，协调者将“消费组分配结果”保存到 内部主题之后 ， 才会发送“同步组响应”给每个消费者。 协调者保存“消费组分配结果”和保存生产者发送的消息一样，都落到了副本管理器的追加消息流程上。 服务端处理生产者消息的回调方法是发送响应给生产者，协调者保存完“消费组分配结果”后的回调方法是将“同步组响应”返回给消费者 。 相关代码如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

前面主要分析的就是：协调者如何处理消费者的“加入组请求”和“同步组请求”，直到返回“加入组响应”和“同步组响应”给消费者。但在这个过程中，我们省略了一些细节，并遗留了下面这些问题。

*  调用消费者成员元数据的回调方法后，为什么要重置回调方法为空，并且还要启动一次心跳监控？
 *  协调者为什么在发送“加入组响应”之前就更新状态为“等待同步”，而在发送“同步组响应”之后才更新状态为“稳、定”？另外，消费组的其他状态比如“准备再平衡”是什么时候发生的？
 *  不同消费者在不同时刻发送请求给协调者，协调者并不会立即发送响应结果给消费者。比如协调者会等到收集完所有的消费者成员信息才发送“加入组响应”，等到收到主消费者发送的“消费组分配结果”才发送“同步组响应”，那么通过什么方式来做到“延迟发送响应结果”给消费者，以及如何判断请求已经处理完成可以发送响应结果了？要回答上面这些问题，归根到底是要理解下面这两个概念。
 *  “延迟的加入组”操作对象(DelayedJoin，下文简称“延迟操作”)。
 *  消费组的状态机转换。有4种状态：“稳定”“准备再平衡”“等待同步”“离开”。

这两个操作互相关联：在创建延迟操作对象之前，会更新消费组状态为“准备再平衡”；当延迟操作对象完成后，会更新消费组状态为“稳定”。由于消费组状态在不同时间内发生了变化，因此协调者在处理不同消费者的请求时，会根据当前消费组的状态进入不同的执行流程。下面先来看下“延迟的操作”。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/0a5f38118ea246d18f81fd25714d4e2b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/5d63958bf8cb4333a8d695b22db8bf4a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/61542742ca064ebbb7baa4d0bbb55425.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/ec1a0b0d0abd4d8b98596ccb1d9f71e0.png