一文读懂kafka消息拉取机制｜线程拉取模型

发表于 3年以前 | 总阅读数：650 次

在详细介绍Kafka拉取之前，我们再来回顾一下消息拉取的整体流程：

在消费者加入到消费组后，消费者Leader会根据当前在线消费者个数与分区的数量进行队列负载，每一个消费者获得一部分分区，接下来就是要从Broker服务端将数据拉取下来，提交给消费端进行消费，对应流程中的pollForFetches方法。

要正确写出优秀的Kafka端消费代码，详细了解其拉取模型是非常重要的一步。

1、消息拉取详解

1.1 消费端拉取流程详解

消息拉取的实现入口为：KafkaConsumer的pollForFetches，接下来我们将详细剖析其流程，探讨kafka消息拉取模型，其实现如下所示：

整个消息拉取的核心步骤如下：

获取本次拉取的超时时间，会取自用户设置的超时时间与一个心跳包的间隔之中的最小值。
从拉取缓存区中解析已异步拉取的消息。
向Broker发送拉取请求，该请求是一个异步请求。
通过ConsumerNetworkClient触发底层NIO通信。
再次尝试从缓存区中解析已拉起的消息。

1.1 Fetch的sendFetches详解

经过队列负载算法分配到部分分区后，消费者接下来需要向Broker发送消息拉起请求，具体由sendFetches方法实现。

Step1:通过调用preparefetchRequest，构建请求对象，其实现的核心要点如下：

构建一个请求列表，这里采用了Build设计模式，最终生成的请求对象：Node为Key,FetchSessionHandler.FetchRequestData为Value的请求，我觉得这里有必要看一下FetchRequestData的数据结构：

其中ParitionData汇总包含了本次消息拉取的开始位点。
通过fetchablePartitions方法获取本次可拉取的队列，其核心实现要点如下：
从队列负载结果中获取可拉取的分区信息，主要的判断标准：未被暂停与有正确位点信息。
nextInLineRecords？
去除掉拉取缓存区中的存在队列信息(completedFetches)，即如果缓存区中的数据未被消费端消费则不会继续拉取新的内容。
获取待拉取分区所在的leader信息，如果未找到，本次拉取将忽略该分区，但是会设置需要更新topic路由信息,在下次拉取之前会从Broker拉取最新的路由信息。
如果客户端与待拉取消息的broker节点有待发送的网络请求(见代码@4)，则本次拉取任务将不会再发起新的拉取请求，待已有的请求处理完毕后才会拉取新的消息。
拉取消息时需要指定拉取消息偏移量，来自队列负载算法时指定，主要消费组的最新消费位点。

Step2:按Node依次构建请求节点，并通过client的send方法将请求异步发送，当收到请求结果后会调用对应的事件监听器，这里主要的是一次拉取最大的字节数50M。

值得注意的是在Kafka中调用client的send方法并不会真正触发网络请求，而是将请求放到发送缓冲区中，Client的poll方法才会真正触发底层网络请求。

Step3:当客户端收到服务端请求后会将原始结果放入到completedFetches中，等待客户端从中解析。

本篇文章暂时不关注服务端对fetch请求的处理，等到详细剖析了Kafka的存储相关细节后再回过来看Fetch请求的响应。

1.2 Fetcher的fetchedRecords方法详解

向服务端发送拉取请求异步返回后会将结果返回到一个completedFetches中，也可以理解为接收缓存区,接下来将从缓存区中将结果解析并返回给消费者消费。从接收缓存区中解析数据的具体实现见Fetcher的fetchedRecords方法。

核心实现要点如下：

首先说明一下nextInLineRecords的含义，接下来的fetchedRecords方法将从这里获取值，该参数主要是因为引入了maxPollRecords(默认为500)，一次拉取的消息条数，一次Fetch操作一次每一个分区最多返回50M数据，可能包含的消息条数大于maxPollRecords。

如果nextInLineRecords为空或者所有内容已被拉取，则从completedFetch中解析。
从completedFetch中解析解析成nextInlineRecords。
从nextInlineRecords中继续解析数据。

关于将CompletedFetch中解析成PartitionRecords以及从PartitionRecords提取数据成Map< TopicPartition, List< ConsumerRecord< K, V>>>的最终供应用程序消费的数据结构，代码实现非常简单，这里就不再介绍。

有关服务端响应SEND_FETCH的相关分析，将在详细分析Kafka存储相关机制时再介绍。在深入存储细节时，从消息拉取，消息写入为切入点是一个非常不错的选择。

2、消息消费端模型

阅读源码是手段而不是目的，通过阅读源码，我们应该总结提炼一下Kafka消息拉取模型(特点)，以便更好的指导实践。

首先再强调一下消费端的三个重要参数：

fetch.max.bytes

客户端单个Fetch请求一次拉取的最大字节数，默认为50M，根据上面的源码分析得知，Kafka会按Broker节点为维度进行拉取，即按照队列负载算法分配在同一个Broker上的多个队列进行聚合，同时尽量保证各个分区的拉取平衡，通过max.partition.fetch.bytes参数设置。
max.partition.fetch.bytes 一次fetch拉取单个队列最大拉取字节数量，默认为1M。
max.poll.records 调用一次KafkaConsumer的poll方法，返回的消息条数，默认为500条。

实践思考：fetch.max.bytes默认是max.partition.fetch.bytes的50倍，也就是默认情况一下，一个消费者一个Node节点上至少需要分配到50个队列，才能尽量满额拉取。但50个分区(队列)可以来源于这个消费组订阅的所有的topic。

2.1Kafka消费线程拉取线程模型

KafkaConsumer并不是线程安全的，即KafkaConsumer的所有方法调用必须在同一个线程中，但消息拉取却是是并发的，线程模型说明如下图所示：

其核心设计理念是KafkaConsumer在调用poll方法时，如果本地缓存区中(completedFeches)存在未拉取的消息，则直接从本地缓存区中拉取消息，否则会调用client#send方法进行异步多线程并行发送拉取请求，发往不同的broker节点的请求是并发执行，执行完毕后，再将结果放入到poll方法所在线程中的缓存区，实现多个线程的协同。