3.3.1 拉取线程管理器

川长思鸟来 2022-10-22 09:17 32阅读 0赞

### 3.3 消费者拉取数据 ###

消费者连接器通过再平衡操作分配到的分区相当于t作任务，任务需要由工作线程完成。生产者要写消息到服务端的分区，这是通过Sender工作线程完成的，消费者要读服务端分区的消息则通过拉取管理器的拉取线程完成。下面我们来分析消费者客户端拉取消息的具体实现。

#### 3.3.1 拉取线程管理器 ####

消费者的拉取管理器(ConsumerFetcherManager)管理了当前消费者的所有拉取线程，这些拉取线程会从服务端的分区拉取消息。前面我们知道每个消费者都会分配到分区信息集合，这些分区会被拉取管理器的startConnections（)方法使用。partitionMap变量表示被消费者拉取管理辑所管理的分区集合。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]  
Kafka的生产者和消费者都只能和分区的主副本通信，所以消费者再平衡后分配到分区信息，需要找到分区的主副本。拉取管理器会启动一个后台的LeaderFinderThread线程，不断找IH已经存在主副本的分区，被选中的分区会被加入对应的拉取线程。如图3-18所示，消费者连接器触发ZKRebalancerListener监听器的再平衡操作，将分区信息传递给拉取管理类，后台线程会选中已经准备好的分区交给不同的拉取线程，然后拉取线程才会真正开始工作。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

1.  选择有主副本的分区

LeaderFinderThread后台线程是抽象拉取管理器（AbstractFetcherManager）的内部类，它的工作是找出已经有主副本的芬区。初始时假设分配给消费者的所有分区（topicinfos）都找不到主副本，topicinfos会被加入候选集合（noleaderPartitionS时，没有主副本的分区）中。后台线程每次运行时，只需要判断候选集合中的分区是存有主副本，如果找到有主副本的某个分区，就将这个分区从候选集合中移除。后台线程下一次运行时，上次已经被选出并移除的分区就不会存在于候选集合，所以每个分区都只会有一次被加入拉取线程的机会，并不会被重复加入。

拉取管理器的后台线程将有主副本的分区分配给拉取线程，而拉取线程还需要知道分区的拉取位置才能正常t作。前面再平衡操作分配给消费者的分区信息，会保存在拉取管理器的partitionMap变革’中，分区信息中包含了拉取位置，所以可以从partitionMap中读取分区的拉取位置。addFetcherForPartitions()方法接收的参数类型是：分区→BrokerAndinitialOffset（分区的主副本节点，分区的拉取位置）。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]  
分区是否有主副本这个信息并不存在于客户端中，客户端需要向Kafka集群的任意一个消息代理节点发起主题元数据请求←ToplcMetadata）。因为一个主题有多个分区，所以主题元数据也包括了所有分区的元数据，这其中就有每个分区的主副本信息。有了分区的主副本后，拉取管理器会把属于同一个主副本的不同分区分配给相同的拉取线程。

注意：拉取线程本身不需要管理分区，它知道自己要负责哪些分区就可以了。拉取线程开始工作时一定是有分区的，如果没有分配分区给拉取线程，即使存在拉取线程也应该关闭掉。

总结一下。后台线程找出有主副本的分区、创建拉取线程、为拉取线程指定分区都是由拉取管理器完成的，因为只有管理器才有资格管理所有的线程。拉取管理器管理所有的拉取线程，而每个拉取线程则管理自己的分区和偏移量，每个角色都各司其职。拉取管理器不需要关心底层分区的偏移盘，拉取线程向己会根据偏移茧，执行分区的拉取任务。

1.  创建拉取线程

客户端涉及和分区相关的工作线程，通常以消息代理节点为粒度，让一个线程管理这个消息代理节点上的多个分区。工作线程通常都是重量级的对象，不适合每个分区都启动一个单独的线程，能够合并的分区尽量要放在同一工作线程中处理。

比如生产者发送到同一个节点的多个分区，会被包装成一个请求发送到服务端。拉取管理器为分区选择所属的拉取线程，每一个拉取线程也会负责多个分区。将分区添加到拉取线程上，首先要调用createFetcherThread()创建拉取线程，然后调用addParti.ti.ans()添加分区到线程上。addFetcherForPartitions()方法的输入参数partitionAndOffsets除了用来确定到底需要有多少个线程，以及每个线程要负责哪些分区外，还需要知道每个分区的初始偏移盘值。

那么如何对所有分区进行分组，保证分组后同一组的所有分区都共用一个拉取线程？分组条件是BrokerAndFetcherId，其中Broker表示分区的主副本节点，拉取编号FetcherId的计算方法是：对主题进行散列化加上分区编号的结果，再和线程数numFetchers进行取模。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]  
注意：因为拉取线程会和服务端进行网络通信，客户端针对不同服务端节点应该使用不同的拉取线程。所以拉取线程和分区的主副本有关，不同主副本一定有不同的拉取线程。另外，取模计算和主题、分区、线程数有关。默认的线程数为1时，相同主题不同分区计算出来的拉取编号为0（任何数%1都等于0）。结合消息代理节点，线程数为l时，消费者对每个目标节点都只会创建一个拉取线程。

Kafka中有两种对象会拉取消息一一消费者和l备份副本，拉取管理器也有对应的两种实现。消费者管理器会创建消费者的拉取线程，副本管理器会创建副本的拉取线程。C「eateFetche「Thread（）方法创建拉取线程的参数sou「ceB「oke「表示拉取线程要连接的目标节点，因为拉取消息时必须知道从哪个目标节点上拉取。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]  
消费者和备份副本的拉取线程都需要知道分区的偏移盐，这样拉取钱程才能知道要从哪里开始拉取消息。消费者的拉取偏移量来向于分区信息对象的fetchedOffset，即在消费者再平衡时从ZK读取。所以创建消费者拉取线程时，需要传递带有分区信息对象的partitionMap。而备份副本的偏移量并没有保存在ZK中，而是保存在备份副本的本地内存中。

1.  拉取线程的拉取状态

消费者拉取管理器创建消费者拉取线程时，会把它持有的‘‘代表分配给当前消费者的所有分区信息“，即partitionMap全集数据传给每一个拉取线程。因为分区信息对象中的队列会用来存放分区的拉取结果，如果没有把分区信息传给每个拉取线程，拉取钱程就无法获取；J,1:111的队列，就没有地方来存放拉取到的消息。如图3-19所示，ZKRebalancerListener将分区信息集合传给拉取管理器（杏则管理器也不知道它到底要拉取哪些分区），拉取管理器再把分区信息、集合传给每个拉取线程。

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

拉取管理器的LeaderFinderThread后台线程会将分区添加到对应的拉取钱程。具体过程为：后台钱程选择有主副本的分区后，将分区添加到所属的拉取线程；如果拉取线程不存在，就会创建拉取线程，再将分区添加到拉取线程中。抽象的拉取线程中有一个表示分区及其对应分区拉取状态的partitionMap变量，它的含义和抽象拉取管理器中表示分区及其对应分区信息的partitionMap不同。LeaderFinderThread并非一次性就能找到全部有主副本的分区，可能有些分区还没有主副本，就只能在下次被找到，而且这次找到的不会被再次查找。抽象拉取线程调用addPartitions（)~夺分区添加到拉取线程中，也只是每次找到的那部分分区，不过抽象拉取线程的partitionMap保存的是：目前为止所有存在主副本的分区及其对应的分区拉取状态。相关代码如下：

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

注意：消费者的全集分区信息对象会传递给抽象拉取管理器的partition问ap，抽象拉取管理器会再把partitionMap传给每个消费者拉取线程。但是抽象拉取线程中的partitionMap表示的是属于当前消费者分区的拉取状态，它是全局分区中的子集数据。如果再推广到所有消费者，每个消费者也只是分配到主题所有分区中的子集。你可能觉得拉取线程有拉取管理器传递给它的partitionMap，和它本身管理的partitionMap拉取状态容易各淆。实际上，这两个对象表示的含义不同，前者是分区信息，后者是拉取状态。拉取状态会用在拉取工作的执行中，而分区信息用在其他地方（参见333节的消费者拉取线程)。

总结一下拉取管理器的LeaderFinderThread后台线程的主要工作，具体步骤如下。

(1)后台线程调用抽象拉取管理器的addFetcherForPartitians（)方法。
    (2)addFetcherForPartitians（)方法调用CreateFetcherThread（)抽象方法。
    (3)消费者拉取管理器的CreateFetcherThread（)创建具体的消费者拉取线程。
    (4)消费者拉取管理器调用抽象拉取线程的addPartitians（)将分区添加到步骤（3)的拉取钱程。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/4ea4b38bb497486c841cac61550ab33f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/771b3f2d8e3c4401aab6a9298c7e9c7d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/2dc67fac7c2643efbbc4532a823ffa62.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/feb4058773524cbe865f7a9005329031.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/11c0b2bb9b0f4efc9cf781563374b296.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/a14ec3e2785946d5b866c44df998a545.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/fe5eb7e9246c45f3995ca10190bff67b.png