5.1.1 元数据与分区分配器

一时失言乱红尘 2022-11-18 05:41 96阅读 0赞

> 第4章分析了消费者客户端轮询的3个步骤：发送拉取请求，客户端轮询，获取拉取结果。消费者在发送拉取请求之前，必须首先满足下面的两个条件。

*  确保消费者已经连接协调者，即找到服务端中管理这个消费者的协调者节点。
 *  确保消费者已经分配到分区，即获取到协调者节点分配给消费者的分区信息。

消费者客户端除了从协调者节点获取到分区，还会发送心跳请求(详见43节)、提交偏移量(详见44节)给协调者节点。其中，提交偏移革主要和消息的处理有关，协调者只是作为偏移盐的存储介质，因此不是本章的重点。而消费者发送心跳请求给协调者，则有可能归现各种各样的问题，如下。

*  消费者没有及时发送心跳，可能是消费者发生故障。这时协调者应该能够意识到有消费者离开了消费组，需要对消费组内的所有消费者重新分配分区。
 *  消费者发送心跳给协调者，但是服务端的协调者节点也可能出现故障。而消费者所有依赖协调者的工作都必须首先存在协调者，所以消费者会等待一段时间重新连接正确的协调者节点，然后由协调者节点再次分配分区。

注意：如果是协调者节点发生故障，服务端会有自己的故障容错机制，选出管理消费组所有消费者的新协调者节，点。消费者客户端没有权利做这个工作，它能做的只是等待一段时间，查询服务端是否已经选出了新的协调节点。如果消费者查到现在已经有管理协调者的协调节点，就会连接这个新协调节，哉。由于这个协调节点是服务端新选出来的，所以每个消费者都应该重新连接协调节点。

旧版本的消费者再平衡操作主要通过ZK监听器方式来触发，每个消费者为了获得分区，都会在客户端执行分区分配算法。有了协调者之后，分区的分配算法理所当然要交给协调者完成。消费者通过发送“加入组请求”给协调节点来获取分配给它们的分区。协调者可以向己执行这个分配算法(在服务端计算)，或者交给其中的一个消费者去完成(在客户端计算)。第二种方法的代价是负责执行分配算法的消费者，还要把分配结果同步给协调者，最终协调者会返回分配结果给每个消费者。

本章主要分析消费者发送“加入组请求”和“同步组请求”给协调者，协调者如何分配分区给消费者。另外，心跳请求也间接影响了消费者是否需要重新加入消费组，因此我们也会分析协调者如何监控客户端的心跳。

注意：为了让协调者能将分区分配给每个消费者，每个消费者都要发送“加入组请求”给协调者。协调者知道了消费组下有哪些消费者，就能决定把哪些分区分配给对应的消费者。

### 5.1 消费者加入消费组 ###

> 消费者发送“加入组请求”获取分区定义在抽象客户端协调者的ensureActiveGroup()方法，而该方法又定义在消费者的轮询操作中。即消费者每次轮询都会调用该方法，但并不是每次轮询都要发送“加入组请求”。从412节的分析中可知，主要通过rejoinNeeded变量和订阅状态的needsPartitionsAssigned变量控制是否需要重新加入组。通常在成功执行一次“加入组”的流程后，会设置这两个变量为false,这样消费者下次轮询时，调用ensureActiveGroup()方法就会立即返回。

消费者发送“加入组”请求会返回一个异步请求对象。为了确保消费者只有分配到分区之后，才可执行后续的拉取分区悄息操作，消费者需要通过客户端阻塞式地轮询(clientpoll(futur时，参见4.2.3节)等待异步请求完成。在异步请求完成后，我们可以在回调方法onJoinCol"lplete()中将“‘加入组响应’结果”，即分配的分区设置到消费者订阅状态的“分配结果”中。

注意：消费者妥拉取哪些分区的消息，是从消费者订阅状态(SubscriptionState，参见411节)中获取数据的。消费者从协调者获取到分配的分区后，妥将分配结果加入到订阅状态中。下面把ensureActiveGroup()方法中执行条件相关的部分(即判断是否需要重新加入消费组)都去掉，只保留“加入组请求”相关的业务逻辑，主要步骤如下。

(1)消费者加入消费组之前，需要做一些准备工作，比如同步提交一次偏移量，执行监昕器的回调。  
(2)消费者创建“加入组请求”，包括消费者的元数据作为请求的数据内容。  
(3)消费者发送“加入组请求”，采用组合模式返回一个新的异步请求对象，并定义回调处理器。  
(4)客户端通过轮询，确保组合模式返回的异步请求必须完成，这是一个阻塞的方法。  
(5)异步请求完成后，执行回调方法，将分区设置到消费者的订阅状态，并重置心跳定时任务。  
相关代码如下：

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

ensureActiveGroup()方法主要的3个子流程是：消费者准备加入消费组，发送“加入组请求”并定义响应处理器，消费者完成加入消费组。下面先来看创建“加入组请求”需要发送的元数据。

#### 5.1.1 元数据与分区分配器 ####

> 消费者客户端创建“加入组请求”，请求对象的变量有：消费组名称(groupid)、消费者成员编号(memberId)、协议类型(protocolType)、元数据(metadata)。

消费者客户端和第8章的连接器客户端都需要发送“加入组请求”从协调节点获取分区，所以有两种协议类型：消费者(consumer)、连接器(connect)。“协议内容”和“元数据”构成协议元数据(ProtocolMetadata)。元数据和协议类型有关，消费者的元数据是订阅的主题，连接器的元数据是一些配置信息。相关代码如下：

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

注意：ProtocolMetadata由Protocol和Metadata组成，前者是协议内容，后者是元数据内容。

消费者的协议内容是分区分配器(PartitionAssignor)的类名。目前Kafka消费者支持两种分区分配方式：范围(协议名称是range，对应的类是RangeAssignor)、循环(协议名称是roundrobin，对应的类是RoundRobinAssignor)。PartitionAssignor接口定义了assign()抽象的分配算法。不管是协调者还是消费者负责分配分区，调用assign()执行分区分配算法需要下面两个参数，表示分配算法必须传递下面两个数据才能工作。

*  subscriptions。哪些消费者订阅了哪些主题。
 *  metadata。消费者订阅的这些主题都有多少个分区。

subscriptions表示每个消费者的订阅信息，通过让每个消费者都发送向己的订阅信息给协调者，协调者就可以收集到所有消费者订阅的主题。metadata是集群的元数据，它记录了每个主题的相关信息，包括主题的分区数。这样协调者就可以将对应主题的分区，分配给所有订阅这些主题的消费者。相关代码如下：

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

分配方法的返回值包含了每个消费者对应的分配结果，分配结果是一个“主题分区集合”，表示分配给消费者的所有主题分区。消费者发送订阅信息(Subscriptlon)对象以及服务端返回分配结果(Assignment)对象，在网络传输时都需要进行序列化。

注意：Subscription是订阅信息，消费者订阅了哪些主题。Assignor是分区分配器，它会执行分区分配的算法，Assignment是分配结果，是执行分配分区后的结果。

消费者发送请求用到的元数据，分区分配器会用在具体分区分配的算法执行上。下面分析消费者发送完请求到调用分区分配器assign()方法的整个过程。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/f61191c0a2c4483c9e3dde42ef7153bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/2a82657595134756b0ed57b8702b204d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/fd4018b81de84d9d8a81e30b4716fa62.png