4.1.4 消费者拉取消息

墨蓝 2023-01-13 04:28 103阅读 0赞

#### 4.1.4 消费者拉取消息 ####

> 消费者创建拉取请求的准备工作，和生产者创建生产请求的准备工作类似，它们都必须和分区的主副本交互。一个生产者写人的分区和消费者分配的分区都可能有多个，同时多个分区的主副本有可能在同一个节点上为了减少客户端和服务端集群的网络连接，客户端并不是以分区为粒度和服务端交互，而是以服务端节点为粒度如果分区的主副本在同一个节点上，应当在客户端先把数据按照节点整理好，把属于同一个节点的多个分区作为一个请求发送出去一个消费者可以允许同时向多个主副本节点发送请求，这个请求包括属于这个主副本节点的多个分区。

创建拉取请求的一个重要数据，是需要指定从分区的什么位置开始拉取。消费者的订阅状态中保存了分配给消费者所有分区的状态信息，其中就包括拉取偏移量（position变量）。如图4-16所示，从消费者分配分区开始，结合使用集群的元数据（Cluster）和消费者的订阅状态（SubscriptionState),向每个目标节点发送拉取请求，具体步骤如下。

(1）消费者向协调者申请加入消费组，并得到分配给它的分区。
    (2）集群的元数据记录了分区及其所属主副本节点的信息。
    (3）消费者的订阅状态记录了分区及其最近拉取偏移量的信息。
    (4）拉取器t作时，会将所有的分区按照主副本节点整理。
    (5）每个主副本对应一个拉取请求，消费者向服务端节点发送拉取请求。

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

1.  发送拉取请求  
    拉取器的准备工作做好后，接着通过消费者网络客户端（ConsumerNetworkClient）将请求发送出去。调用send（）方法实际上并不会将请求通过网络发送到服务端，只有等到轮询的时候才会真正发送出去。这也是消费者轮询时在发送请求后，需要调用客户端轮询方法的原因。拉取器的发送拉取请求方法还添加了一个监昕器，它的作用类似于4.1.3节第5小节“轮询与结果”中的回调对象。当客户端轮询到拉取请求对应的响应结果，便会调用这个监昕器的回调方法，来处理响应结果。下面3段代码是Kafka消费者、拉取器、消费者网络客户端完成发送拉取请求调用逻辑的过程。相关代码如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

客户端发送请求后需要获取响应结果，在旧消费者中，客户端发送拉取请求，会阻塞直到服务端返回结果。这里新消费者采用“异步和轮询”的方式：客户端调用client.send（）方法不会被阻塞，而是返回一个异步对象。同时，为了能够处理服务端返回的响应结果，在返回的异步对象上添加l一个处理拉取响应的监昕器，当收到服务端的响应结果时，监昕器里的回调方法将执行。

注意：关于网络层客户端、异步请求对象、监听器等相关的调用流程，4.2节再详细分析。

1.  处理拉取晌应  
    客户端发到指定服务端节点的拉取请求可能包括多个分区，所以拉取请求对应的响应结果也包含了多个分区的数据。拉取请求对象（FetchRequest）的分区数据表示客户端从分区的哪个位置开始’拉取，拉取响应对象（FetchResponse）的分区数据表示服务端返回了哪些消息。监听器调用的处理拉取响应方法（handleFetchResponse()），会将每个分区数据的字节数组记录集（recordSet），转换成一条条可以被消费者直接读取的消费者记录（ConsumerRecord），并添加到拉取器的this.records全局成员变量中。由于响应结果已经按照分区组织好了，所以拉取器构造的分区记录集（PartitionRecords）也是每个分区对应一批拉取结果。相关代码如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]

在解析分区数据之前，还要再次查询消费者订阅状态，看这个分区是否可以拉取。因为有可能消费者发送完拉取请求后，在没有收到拉取结果之前，就取消了对某个分区的拉取（比如调用消费者的pause（）方法暂停分区的拉取）。由于服务端无法感知这个动作，它还是会返回之前的所有分区数据。所以拉取器在处理拉煎结果时，还需要再次判断分区数据是不是客户端想要的。比如消费者分配了\[PO,Pl,P2］这3个分区，然后发送包含这3个分区的拉取请求，接着暂停了Pl分区。服务端会返回3个分区的数据，但是实际上客户端现在只需要［PO,P2］两个分区。

拉取结果集按照分区进行封装后放在了全局变革this.records中，客户端还要主动调用拉取器的fetchedRecords（）方法才能获取到拉取结果。虽然客户端轮询、异步请求对象、回调方法这些组合都是异步的，但是为了获取消息，消费者必须主动调用，而不能向动获取结果。这也是客户端主动拉取消息，而不是服务端惟送消息给客户端的体现。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/9141cc5268464c778a6044859db58b98.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210415231648134.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/0a4d9aa8a1fa49bd8f42b4ecf0717e3f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210415231725377.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/5dc78f224a954e68af31eebc0ad53fff.png