3.3.3 消费者拉取线程

向右看齐 2023-01-13 04:24 74阅读 0赞

#### 3.3.3 消费者拉取线程 ####

消费者拉取管理器在创建拉取线程时，会将表示分区及其分区信息对象的全局partitionMap作为类级别的变量传给每个拉取线程，但每个拉取线程在拉取时实际上只会负责一部分的分区。拉取线程在拉取到分区数据后，需要将拉取结果保存到分区信息的队列中。因为每个拉取线程都持有全局的partitionMap引用，所以processPartitionData()方法在处理拉取结果时，可以获取到分区信息中的队列，并将拉取结果填充到队列中。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]  
注意：拉取线程的抽象类在拉取工作中也有一个partitionMap变量，但这个变量表示的是：分配给当前消费者的所有分区的拉取状态。它的作用域仅仅在抽象拉取线程中，和消费者拉取线程没有关联。

消费者拉取线程构建拉取请求后，通过Si111pleConsu111er代表和服务端的网络连接。Si111pleConsu111er使用同步类型的阻塞通道发送请求和接收响应。相关代码如下：

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

消费者和备份剧本的拉取线程在收到拉取消息后处理方式不同，比如备份副本会把数据写到自己本地的日志文件中，消费者则会把数据填充到分区信息对象的队歹lj中供消费者客户端应用程序获取。

注意：除了拉取请求（FetchRequest),Si111pleConsu111er还提供了其他两种请求的发送：提交偏移量的请求（OffsetC0111111itRequest）、获取偏移量的请求（OffsetFetchRequest）。这些方法不仅作为高级A凹的内直实现，也可以提供给低级API进行手动控制。

1.  分区信息的队列保存拉取的消息

消费者拉取线程的fetch()方法，通过Si111pleConsu111er向服务端发起请求并返回所有分区及其数据(PartitionData），然后处理每个分区的拉取结果。ProcessPartitionData()方法的参数是分区数据的底层消息集，即从服务端拉取到的分区消息对象。它会根据分区得到分区信息对象，调用其enqueue()方法，将消息集包装成数据块（FetchedDataChunk）放入分区信息对象的队列中。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]  
如图3-20所示，分区信息对象作为消费者应用程序和拉取线程的中间桥梁，保存了“拉垠偏移盘”和“队列”两个重要的信息。拉取偏移量用在拉取钱程中，表示要从分区的什么位置拉取消息，拉取钱程拉取到数据后将拉取结果填充到队列中。回顾一下消费者连接器在一开始创建了队列和消息流时，队列是空的。现在，分区信息对象的队列有数据后，消费者应用程序可以通过消息流从队列中取得数据。

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

2.拉取出现错误的处理方式

拉取线程向服务端发送拉取请求如果收到UoFFSET\_our\_oF\_RANGE错误码表示扣，取请求的拉取偏移iiJ超H－＼服务端分区的范围，拉取线程就要根据消费者设置的重置策略设置拉取偏移盐，并且更新分区的拉取状态。下一次发送拉取请求时，拉取线程使用重茸的偏移盐拉取分区的消息。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]  
消费者拉取线程拉取消息过程中还可能遇到其他的错误，通常是分区的主副本发生变化，导致拉取线程不能再从之前的节点上读取数据。此时，拉取线程会调用handlePartitionsWithErrOrs()抽象方法进行处理。首先，这个分区不应该继续拉取，所以要将其从拉取状态集合中移除，这样下次拉取请求就不会存在这个错误的分区了。然后，将分区加入到消费者拉取管理器的noleaderPartitionset中，这样LeaderFinderThread就会重新选择分区的主副本，让拉取线程连接最新的节点。相关代码如下：

![在这里插入图片描述][20210325215809583.png]

如图3-21所示，总结从分配分区给消费者，到拉取线程拉取消息返回给消费者的具体步骤如下。

(1)再平衡操作将分区分配给消费者，读取ZK的偏移量作为分区信息的拉取偏移量。
    (2)分区信息的队列用来存储结果数据，拉取偏移量作为拉取线程初始的拉取位置。
    (3)拉取线程拉取分区的数据，初始时从拉取偏移量开始拉取消息。
    (4)partitionMap表示分区的最新拉取状态，每次拉取数据后都要更新拉取状态。
    (5)拉取线程创建拉取请求，并通过SiMpleConsL』附r发送请求和接收响应结果。
    (6)拉取钱程技取到分区消息后，将分区数据的消息集填充到分区信息对象的队列。
    (7)创建消费者连接对象时，会创建队列和消息流，一个队列关联了一个消息流。
    (8)消费者客户端从消息流中迭代读取结果数据，实际上就是从队列中拉取消息。

目前为止，虽然拉取线程从服务端成功拉取到了最新消息，并放到分区信息对象的队列里，但是客户端其实“还没有开始读取队列中的消息”。消费者的客户端应用程序需要通过“迭代消息流”，才能从队列中读取IH消息。而只有消费者客户端成功消费到数据，才表示消息已经到达客户端。否则在这之前尽管数据已经在客户端进程中，但是还没有到达客户端应用程序，就不算做被消费，只能说“正在等待被消费”。

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/1983622cf645474dabc37091d73bd171.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/10c3e87ca2504cd38ac4351a5a68c0c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/9fe12410687f467eab26e26bdfff44e8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/3cb5a5046d904c0fb3250edd6468f90a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/2242aa1fd4ed459dbf2acad75f64d4b5.png
[20210325215809583.png]: /images/20221022/1351ccb7fc01494a97b6348154613fe9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/849b629c8f1c41fd81b9da570fb95d6c.png