4.3.5 心跳和协调者的关系

╰半夏微凉° 2023-01-13 15:56 120阅读 0赞

#### 4.3.5 心跳和协调者的关系 ####

客户端调用心跳任务的reset()方法会创建第一个延迟任务，这个方法的调用链如下。

*  确保协调者是已知的，即消费者客户端必须连接上管理消费组的协调者。
 *  确保消费组是活动的，即消费者必须分配到分区。

注意：上面两个调用的方法都定义在对应的请求回调处理器中，前者是“获取消费组的协调者”请求(GroupCoordi.natorRequest)，后者是“加入消费组”请求(Joi.叫roupRequest)。消费者和协调者进行交互操作，必须确保消费者已经知道并且连接上协调者所在的节点，如果都没有连接上协调者，心跳等其他操作都不会正常进行。连接上协调者后，就可以立即向协调者发送一次心跳。另外，如果消费者需要重新加入消费组，在分配到分区后，也要重置心跳任务。

消费者发送心跳，正常来说应当只是通知一下服务端协调者而已。不过在分布式系统中，通信的双方可能都会存在一些问题，比如协调者可能会突然挂掉。这时服务端应该为每个消费组重新选择一个协调者，但如果此后消费者连接的还是原来的协调者就有问题了(它应该连接最新的协调者节点)，  
这种情况应该让消费者重新获取协调者。服务端如果能够在客户端定时发送的心跳任务中附带这种信息，客户端就能够及时知道应该再去找最新的协调者。消费者针对不同错误码的处理方式如下。

*  协调者挂掉了，客户端设置“消费组的协调者”对象为空，消费者需要重新发送“获取消费组的协调者”请求获取新的协调者。
 *  协调者没有挂掉，客户端设置“需要重新加入组”变量为true，消费者需要向协调者重新发送“加入组请求”加入消费组。

相关代码如下：

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

心跳的响应处理中并不执行具体的错误处理操作，比如让客户端连接新的协调者或者让消费者重新加入消费组，心跳只是更新这些后续错误处理操作相关的条件变量。当客户端轮询时会监听到条件变量发生了变化，从而让轮询操作主动地执行对应的操作。这里可以把心跳看作任务通知，必须依赖客户端的轮询来确保能及时捕获到错误情况进行错误处理，不在心跳中处理任务是因为任务的操作时间可能比心跳间隔长得多。所以，客户端的轮询非常重要，它不仅仅驱动了数据的不断拉取，还可以根据心跳结果执行不同的任务。

心跳任务(HeartbeatTask)定义在抽象的协调者类(AbstractCoordi.nator)中。抽象的协调者有两种实现一一消费者的协调者、连接器的协调者，这两者都需要发送心跳给服务端的协调者节点，它们都有组的概念。后者是OIO版本新增的Kafka连接器，会在第8章分析。另外，消费者的协调者因为需要定时存储分区的消费进度，还有一个自动提交偏移量的定时任务。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/b66f3a7082ba4aaa8aba9a1a06941b9b.png