3.3.2 抽象拉取线程

╰半橙微兮° 2023-01-13 04:24 142阅读 0赞

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

#### 3.3.2 抽象拉取线程 ####

抽象拉取线程(AbstractFetcherThread)定义了拉取工作的主要流程，不同的实现类通过抽象方法嵌套在主流程中，保证了主流程执行的一致性。拉取线程工作时，首先要确定数据源(即拉取状态)，每次拉取到消息还要更新拉取状态，确保下一次拉取请求时获得的拉取状态是最新的。

1.  构建拉取请求

拉取管理器后台线程调用拉取线程的addPartitians()方法，partitionMap变量保存了每个分区的拉取状态。拉取线程的运行方法会根据拉取状态构建井处理拉取请求。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]  
注意：由于一个拉取线程要处理多个分区的拉取请求，因此抽象拉取线程要对partitionMap的操作加锁。同时它使用partitionMapCond作为锁的条件，用来在有数据时触发，没有数据时则等待。

1.  处理拉取请求

消费者和备份副本的拉取工作都一样，拉取线程向服务端拉取消息的步骤如下。

(1)buildFetchRequest(partitionMap）根据partitionMap构建拉取请求。
    (2)fetch(fetchRequest）根据拉取请求向目标节点拉取消息，并返回响应结果。
    (3)processPartitionData(partitionData）处理拉取到的分区结果数据。

相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
注意：抽象的拉取线程类调用一次doWork()方法表示执行一次拉取，但并不表示拉取线程只执行一次就结束了。因为抽象的拉取线程类继承了ShutdownableThread，而后者继承了Thread类，它的run方法会循环调用抽象的doWork()方法，所以ShutdownableThread的实现类（比如抽象的拉取线程类和LeaderFinderThread）在doWork()方法中不需妥考虑线程循环运行的问题。

拉取钱程每i用用一次doWork()，都会根据partitionMap构建并发送拉取请求。在拉取到消息之后、处理拉取结果之前，要更新part1tionMap中所有分区的拉取状态。part1tionMap一旦发生变化，依赖于part1tionMap的拉取请求就必须重新构建。那么part1tionMap拉取状态的变化对拉取请求有什么影响呢？

再平衡操作为消费者重新分配新的分区，有些分区可能之前并不属于当前消费者。在消费者拉取线程第一次拉取分区的消息时，需要从ZK读取分区的拉取状态（即偏移量），作为第一次拉取请求的拍取偏移盘。比如，Pl原先被分配给消费者l，经过一次再平衡后被分配给消费者2。消费者2的拉取线程开始工作时，需要从ZK中读取Pl的偏移量，作为第一次拉取请求的Pl拉取状态。后续的拉取请求如果没有发生再平衡操作，则可以直接从part1tionMap中获取Pl分区的拉取状态。每个拉取线程都在自己本地保存了当前负责所有分区的拉取状态，拉取线程每次在收到拉取请求的响应结果后，会用本次拉取消息集最后一条消息的偏移量来更新part1tionMap，下一次会从上一次拉取的最后一个位置继续拉取。这样就可以保证拉取线程的每一次拉取请求总是拉取新的消息集，而且不会重复。所以，除了第一次拉取请求获取拉取状态是从ZK读取，后面的拉取请求都直接从part1tionMap中读取。

注意：由于partitionMap拉取状态只是拉取线程中的内存数据结构，如果拉取线希呈挂了，分区的消费状态就会丢失，所以客户端需要定时地保存消费状态。对于消费者客户端，通过开启autoCornmit开关，定时提交分区的偏移量到ZK中；对于备份副本，则会将偏移量写到本地的检查点文件中。

抽象拉取线程的运行方法中定义了构建拉取请求、拉取消息、处理拉取结果的接口，下面我们来看消费者拉取线程对这3个接口方法的具体实现。

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/0b04803275444b1a9d036cb9b9112d74.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/947103244f1b499ea81950b8bc1ac020.png