消息拉起主要入口为：KafkaConsumer#poll方法，其声明如下：

代码＠１：参数为超时时间，使用 java 的 Duration 来定义。 代码＠２：调用内部的 poll 方法。

KafkaConsumer#poll

代码＠１：首先先对其参数含义进行讲解。

代码＠２：检查是否可以拉取消息，其主要判断依据如下：

代码＠３：如果当前消费者未订阅任何主题或者没有指定队列，则抛出错误，结束本次消息拉取。

代码＠４：使用 do while 结构循环拉取消息，直到超时或拉取到消息。

代码＠５：避免在禁止禁用wakeup时，有请求想唤醒时则抛出异常，例如在下面的@8时，会禁用wakeup。

代码＠６：更新相关元数据，为真正向 broker 发送消息拉取请求做好准备，该方法将在下面详细介绍，现在先简单介绍其核心实现点：

这里会有一个更新元数据是否占用消息拉取的超时时间，默认为 true。

代码＠７：调用 pollForFetches 向broker拉取消息，该方法将在下文详细介绍。

代码＠８：如果拉取到的消息集合不为空，再返回该批消息之前，如果还有挤压的拉取请求，可以继续发送拉取请求，但此时会禁用warkup，主要的目的是用户在处理消息时，KafkaConsumer 还可以继续向broker 拉取消息。

代码＠９：执行消费拦截器。

接下来对上文提到的代码＠６、＠７进行详细介绍。

1.1 KafkaConsumer updateAssignmentMetadataIfNeeded 详解

KafkaConsumer＃updateAssignmentMetadataIfNeeded

要理解这个方法实现的用途，我们就必须依次对 coordinator.poll 方法与 updateFetchPositions 方法。

1.1.1 ConsumerCoordinator#poll

代码＠1：执行已完成的 offset (消费进度)提交请求的回调函数。

代码@2：队列负载算法为自动分配（即 Kafka 根据消费者个数与分区书动态负载分区）的相关的处理逻辑。其实现关键点如下：

代码@3：用户手动为消费组指定负载的队列的相关处理逻辑，其实现关键如下：

代码@4：如果开启了自动提交消费进度，并且已到下一次提交时间，则提交。Kafka 消费者可以通过设置属性 enable.auto.commit 来开启自动提交，该参数默认为 true，则默认会每隔 5s 提交一次消费进度，提交间隔可以通过参数 auto.commit.interval.ms 设置。

接下来继续探讨 updateAssignmentMetadataIfNeeded (更新元数据)的第二个步骤，更新拉取位移。

1.1.2 updateFetchPositions 详解

KafkaConsumer#updateFetchPositions

代码@1：如果订阅关系中的所有分区都有有效的位移，则返回 true。

代码@2：如果存在任意一个分区没有有效的位移信息，则需要向 broker 发送请求，从broker 获取该消费组，该分区的消费进度。相关的实现细节将在后续文章【Kafka 消费进度】专题文章中详细介绍。

代码@3：如果经过第二步，订阅关系中还某些分区还是没有获取到有效的偏移量，则使用偏移量重置策略进行重置，如果未配置，则抛出异常。

代码@4：发送一个异步请求去重置那些正等待重置位置的分区。有关 Kafka 消费消费进度、重平衡等知识将会在后续文章中深入探讨，本文只需了解 poll 消息的核心处理流程。

从 KafkaConsumer#poll 中流程可以看到，通过 updateAssignmentMetadataIfNeeded 对元数据、重平衡，更新拉取偏移量等工作处理完成后，下一步就是需要向 broker 拉取消息了，其实现入口为：KafkaConsumer 的 pollForFetches 方法。

1.2 消息拉取

KafkaConsumer#pollForFetches

代码@1：计算本次拉取的超时时间，其计算逻辑如下：

代码@2：如果数据已经拉回到本地，直接返回数据。将在下文详细介绍 Fetcher 的 fetchedRecords 方法。

代码@3:组装发送请求，并将存储在待发送请求列表中。

代码@4：如果已缓存的分区信息中存在某些分区缺少偏移量，如果拉取的超时时间大于失败重试需要阻塞的时间，则更新此次拉取的超时时间为失败重试需要的间隔时间，主要的目的是不希望在 poll 过程中被阻塞【后续会详细介绍 Kafka 拉取消息的线程模型，再来回顾一下这里】。

代码@5：通过调用NetworkClient 的 poll 方法发起消息拉取操作（触发网络读写）。

代码@6：更新本次拉取的时间。

代码@7：检查是需要重平衡。

代码@8：将从 broker 读取到的数据返回（即封装成消息）。

从上面消息拉取流程来看，有几个比较重要的方法，例如 Fetcher 类相关的方法，NetworkClient 的 poll 方法，那我们接下来来重点探讨。

我们先用一张流程图总结一下消息拉取的全过程：

接下来我们将重点看一下 KafkaConsumer 的 pollForFetches 详细过程，也就是需要详细探究 Fetcher 类的实现细节。

Fetcher 封装消息拉取的方法，可以看成是消息拉取的门面类。

2.1 类图

我们首先一一介绍一下 Fetcher 的核心属性与核心方法。

接下来我们将按照消息流程，一起来看一下 Fetcher 的核心方法。

2.2 Fetcher 核心方法

2.2.1 Fetcher#fetchedRecords

Fetcher#fetchedRecords

代码@1：首先先解释两个局部变量的含义：

代码@2：循环去取已经完成了 Fetch 请求的消息，该 while 循环有两个跳出条件：

代码@3、@4 主要完成从缓存中解析数据的两个步骤，初次运行的时候，会进入分支@3，然后从 调用 parseCompletedFetch 解析成 PartitionRecords 对象，然后代码@4的职责就是从解析 PartitionRecords ，将消息封装成 ConsumerRecord，返回给消费端线程处理。

代码@3的实现要点如下：

从上面可知，上述方法的核心方法是：parseCompletedFetch。

代码@4的实现要点无非就是调用 fetchRecords 方法，按分区组装成 Map，供消费者处理，例如供业务处理。

接下来将重点探讨上述两个方法的实现细节。

2.2.1.1 Fetcher#parseCompletedFetch

在尝试探讨该方法之前，我们首先对其入参进行一个梳理，特别是先认识其主要数据结构。

1、CompletedFetch 相关类图

从上图可以看出，CompleteFetch 核心属性主要如下：

分区的数据是使用 PartitionData 来进行封装的。我们也来简单的了解一下其内部数据结果。

2、parseCompletedFetch 详解