图解 Kafka 生产者元数据拉取管理全流程

大家好，我是 华仔, 又跟大家见面了。,在上一篇中，正式开启了「Kafka的源码之旅」，主要讲述了 KafkaProducer 初始化时用到的核心组件以及消息发送的核心流程，带你梳理生产者初始化整体的源码分析脉络，并通过「场景驱动」的方式带大家一点点的对 Kafka 源码进行深度剖析，一起掌握 Kafka 源码核心架构设计思想。,今天这篇我们就来聊聊生产者是会如何拉取和管理元数据的，带你梳理生产者元数据管理整体的源码分析脉络。,认真读完这篇文章，我相信你会对 Kafka 生产获取和管理元数据源码有更加深刻的理解。,这篇文章干货很多，希望你可以耐心读完。,消息想从 Producer 端发送到 Broker 端，必须要先知道 Topic 在 Broker 的分布情况，才能判断消息该发往哪些节点，比如：「Topic 对应的 Leader 分区有哪些」、「Leader分区分布在哪些 Broker 节点」、「Topic 分区动态变更」等等，所以及时获取元数据对生产者正常工作是非常有必要的。,元数据获取涉及的底层组件比较多，主要分为：「KafkaProducer 主线程加载元数据」、「Sender 子线程拉取元数据」。,接下来我们逐一分析元数据在生产者端是如何被加载和拉取、更新的。为了方便大家理解，所有的源码只保留骨干。,首先我们来看下 KafkaProducer 主线程是如何加载元数据的。,在上一篇中《图解Kafka生产者初始化核心流程》我们分析知道集群元数据的初始化是在 KafkaProducer 主线程的构造函数中来完成的，我们来看一下相关源码：,从上述源代码我们可以看出在 KafkaProducer 的构造方法中，如果metadata为空就会初始化集群元数据类「ProducerMetadata」，然后通过调用 「this.metadata.bootstrap」这个方法来启动引导程序，这时 metaData 对象里并没有具体的元数据信息，因为客户端还没发送元数据更新的请求，后面会通过唤醒 Sender 线程进行发送请求获取元数据的。,这里的 this.meta 其实就是 Kafka 内存中的一个对象，底层会做一层缓存，因此并不会一直请求 Kafka Broker 端进行获取。,我先给大家总结下元数据初始化及启动的调用关系图，口说无凭，我们来扒开源码瞅一瞅，这样更真实。,通过上述的调用关系图，我们可以看出：,接下来会挨个类展开来进行讲解。,在主线程中初始化了 ProducerMetadata 类的对象，我们先来看看这个类都做了哪些事情。,github 源码地址如下：,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java。,我们可以看出这里只是调用了父类的构造函数进行类属性的初始化，接下来我们深度分析下 ProducerMetadata 类中的几个比较重要的方法。,该方法主要用来向元数据主题集合 topics 中添加主题，主要用在「KafkaProducer 主线程」以及「Sender 子线程」中，我们来看下是如何添加的，具体逻辑如下：,此时主题被添加到元数据主题主题集合中，但是如果集合里面数据过期了该怎么办?接下来我们看另外一个方法是如何判断的。,该方法用来判断元数据中是否该保留该主题，会在 handleMetadataResponse 即处理元数据响应结果的时候进行调用，我们来看下它是如何判断的。,该方法用来判断是全量更新元数据还是部分更新元数据，逻辑相对比较简单，主要用在 KafkaProducer 主线程元数据同步等待时调用，后续小节会详细分析。,该方法用来更新生产端元数据的，具体逻辑如下:,从上述 update()方法 中可以看出最后调用 notifyAll() 来唤醒阻塞的线程， 那么它是如何唤醒的呢，这就是接下来要分析的方法。,该方法用来实现线程阻塞的功能，用在主线程中如果发现主题对应的元数据不存在时，阻塞并等待 Sender 线程将元数据更新完成。,重点是调用了 time.waitObject() 方法来实现阻塞功能，它的实现还是有一些技巧的，它的底层通过调用 SystemTime 包里面的 waitObject() 实现的，源码如下：,如果你的项目中也需要类似的功能要实现一个锁，可以参考这里的代码，实现非常巧妙。,接下来我们来重点分析下元数据基类。,首先它是一个线程安全的类，因此Metadata通过synchronized修饰几乎所有方法来保证线程安全，里面封装了元数据基本信息以及元数据的相关操作，主要被用在生产端的 「KafkaProducer 主线程」、「Sender 子线程」中，对于生产者来说只需要获取自己发送的主题集合的元数据，这样可以有效降低网络传输的数据量。因此，它内部只维护部分主题的元数据。,github 源码地址如下：,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/Metadata.java。,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/MetadataCache.java。,我们先来看下这个类的重要字段：,介绍完字段后，我们来深度分析下几个元数据用到的重要方法。,该方法用来引导启动程序的，即在第一次使用前进行初始化的工作。,我们来看下元数据缓存的启动程序都做了哪些事情。,此时获取元数据被启动了，但是还未获取元首，所以元数据缓存都是空的集合，接下来我们来分析下元数据是如何被更新以及如何解析响应的，这里涉及到以下几个方法。,该方法用来设置全部主题更新的标记，上来就先将 needFullUpdate 置为 true，即要全部更新主题，然后返回更新版本。 主要在「ProducerMetadata」、「sender 子线程」、「NetworkClient线程」中使用。,该方法用来设置新集合主题更新的标记，这里并未真正发送更新元数据的请求，只是设置标识位的值，Kafka必须确保在第一次拉消息前元数据是可用的，即必须更新一次元数据。,该方法用来获取集群元数据的，默认从元数据缓存中返回元数据。,该方法主要做了几件事情，具体如下：,最后调用了 handleMetadataResponse 这个重要方法来解析元数据响应，接下来重点分析下它做了些什么。,该方法用来解析元数据响应，我们具体分析下主流程的逻辑：,首先初始化相关集合：「内部主题集合」、「无效主题集合」、「无权限主题集合」。,遍历主题的元数据响应。,将该主题添加到元数据主题集合中。,判断是否保留主题元数据，如果过期可能就没必要保留了。,判断是否是内部主题，如果是内部主题就添加到内部主题集合。,判断是否元数据响应error为空的话,否则如果元数据响应error不为空的话,判断是否部分主题更新响应,至此，元数据相关类的重要方法已经分析完毕，剩余没分析到的方法待到后续场景中进行分析，接下来我们看下主线程是如何加载元数据的。,github 源码地址如下：,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/common/Cluster.java。,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/common/Node.java。,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/common/PartitionInfo.java。,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/common/TopicPartition.java。,Cluster类基本保存了所有的kafka集群相关的信息。,每个topic和每个分区组成的唯一索引，代表一个分区标识。,综上映射关系可以看出，kafka 是以「分区」为最小管理单元，然后分区中的 Leader 负责交互。「Cluster」代表整个kafka集群的实体类，「MetaData」的角色相当于「Cluster」的在客户端的一个维护者。,我们先来看加载元数据大体过程。,接下来分析详细源码，首先客户端可以直接调用「 producer.send() 」进行发送，底层调用 doSend() 方法。,从上述发送的源码中，可以看出来主线程在发送消息前需要先获取元数据，这样才能知道消息要发送到哪些节点。,中间通过调用 waitOnMetadata() 获取元数据的相关信息，为后续发送消息提供支持，接下来我们重点分析下 waitOnMetadata() 这个方法。,该方法用来获取元数据以及元数据消耗时间，我们具体分析下主流程的逻辑：,从元数据缓存中获取元数据，如果metadata不存在当前topic的元数据，会触发一次强制刷新，metaData中的needUpdate置为true。,判断该主题是否是无效的主题， 如果是抛异常。,将该主题放入元数据的主题列表中，通过 Sender 线程定时更新这些主题的元数据。,从元数据缓存中获取主题对应的分区数。,判断元数据缓存是否能满足需要，就是说要能够找到要发送消息的主题分区，条件是：partitionsCount != null && (partition == null || partition < partitionsCount) 即「主题对应的分区数不能为空且发送的分区ID要小于主题的分区数」。,不停的轮询唤醒 sender 线程更新元数据，这里有两个条件要满足其一才可以。,将主题 topic及过期时间添加到元数据主题列表中,标记元数据更新标识，并获取当前元数据版本号，提醒 Sender 线程更新元数据。,唤醒 sender 子线程进行元数据更新以及消息发送。,阻塞主线程并等待Sender线程更新元数据成功。,待 「Sender 子线程更新元数据成功」或者「阻塞超时解除阻塞」，再次获取元数据。,计算等待更新完成元数据消耗时间。,如果超时，抛超时异常。maxWaitMs 最大1分钟。,获取元数据分区数。,返回元数据以及消耗时间。,聪明的读者可能发现在生产者中获取元数据都是基于topic的，主要原因就是对于生产者来说，没必要拉取全部的元数据，只拉取自己需要的主题元数据就可以了。,至此，元数据加载已经分析完了， 接下来我们看下 Sender 子线程是如何拉取元数据的。,由于 Sender 线程的深度分析不是本文的重点，这里先简单的给大家带一带，后续会有专门篇章去分析。,github 源码地址如下：,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java。,https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java。,我们来看看 Sender 线程拉取元数据的大体过程。,接下来我们详细分析具体的源码。,通过源码可以得知 Sender 是一个 Runnable 对象，那整个 Sender 线程执行的核心逻辑就在 run() 方法中，run() 方法中的第一段代码就是循环调用 runOnce() 方法：,上述源码中的 runOnce() 方法是 Sender 线程一个执行的周期，在这个周期中会进行一次批量的请求发送，并进行一次响应的处理。,接下来我们看里面2个涉及到元数据的重要方法：,该方法用来进行 Sender 线程创建请求的核心，这里只分析下元数据涉及的部分：,从元数据缓存中获取元数据。,通过元数据cluster获取要发送的节点 Leader 分区信息。,如果主题的 Leader 分区对应的节点不存在就强制更新元数据。,该方法用来发送网络IO请求，在请求之前先执行尝试更新元数据的请求。,从初始化函数中可以得出元数据更新组件是调用 NetworkClient 内部类即 DefaultMetadataUpdater，前面已多次提到更新集群元数据的场景，而这些更新操作都是在标记集群元数据是否需要更新，而真正执行更新的操作是这里。接下来我们看下这个类的 maybeUpdate() 方法。,从这段源码中可以看出上来先计算下次要更新元数据的时间，我们看下这个计算过程。,元数据是否过期 timeToExpire，计算方式：,允许更新的时间点 timeToAllowUpdate，计算方式：,最后计算这俩值的最大值作为下次更新元数据的时间。,分析完元数据更新时机，最后调用 maybeUpdate(now, node) 发送更新元数据的请求。,该方法用来发送更新元数据的请求，具体的实现逻辑如下：,从上述源码可以看出：更新流程一直在重试，直到元数据更新成功为止。,如果元数据更新成功以后，KafkaProducer 主线程就不会被阻塞，当 NetworkClient 接收到服务端对 Metadata 请求的响应后，就会更新 Metadata 信息， 即 poll() 方法后续的操作。,该方法用来判断成功接收服务端返回的响应，根据不同的响应返回做不同的操作，这里只看下跟元数据更新有关的逻辑。,该方法用来对成功返回信息进行处理，主要是对元数据信息的更新。,查看 response 返回信息的 error。,判断broker信息是否为空,至此，Sender 子线程拉取并更新元数据分析完毕。,最后通过一张图来描述整个元数据拉取的全过程：,这里，我们一起来总结一下这篇文章的重点。,1、通过「场景驱动」的方式从元数据的使用场景出发，抛出主线程加载元数据和子线程拉取元数据的过程是怎样的?,2、带你梳理了「主线程如何加载元数据源码全貌」，包括 ProducerMetadata 类、Metadata 元数据基类、cluster类的几个重要方法源码分析，最后分析主线程加载元数据过程。,3、又带你梳理了「Sender 线程拉取元数据」，包括 Sender 周期执行、元数据更新触发时机的几个重要方法源码分析。,3、最后通过一张元数据流程图来勾勒出元数据拉取和更新的全貌。