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

今天药忘吃喽~ 2024-04-08 10:09 65阅读 0赞

今天这篇我们就来聊聊**生产者是会如何拉取和管理元数据的**，带你梳理生产者元数据管理整体的源码分析脉络。

认真读完这篇文章，我相信你会对 Kafka 生产获取和管理元数据源码有更加深刻的理解。

这篇文章干货很多，希望你可以耐心读完。

![5117da237f82de642f04445c3af65480.png][]

### 01 总的概述 ###

消息想从 Producer 端发送到 Broker 端，必须要先知道 Topic 在 Broker 的分布情况，才能判断消息该发往哪些节点，比如：「**Topic 对应的 Leader 分区有哪些**」、「**Leader分区分布在哪些 Broker 节点**」、「**Topic 分区动态变更**」等等，所以及时获取元数据对生产者正常工作是非常有必要的。

元数据获取涉及的底层组件比较多，主要分为：「**KafkaProducer 主线程加载元数据**」、「**Sender 子线程拉取元数据**」。

接下来我们逐一分析元数据在生产者端是如何被加载和拉取、更新的。为了方便大家理解，所有的源码只保留骨干。

### 02 主线程如何加载元数据 ###

首先我们来看下 **KafkaProducer 主线程是如何加载元数据的**。

**集群元数据的初始化**是在 **KafkaProducer 主线程的构造函数**中来完成的，我们来看一下相关源码：

// 初始化 Kafka 集群元数据，元数据会保存到客户端中，并与服务端元数据保持一致
    if (metadata != null) {
        this.metadata = metadata;
    } else {
        // 初始化集群元数据
        this.metadata = new ProducerMetadata(retryBackoffMs,
              // 元数据过期时间：默认5分钟
              config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG),
              // topic最大空闲时间，如果在规定时间没有被访问，将从缓存删除，下次访问时强制获取元数据
              config.getLong(ProducerConfig.METADATA_MAX_IDLE_CONFIG),
              logContext,
              clusterResourceListeners,
              Time.SYSTEM);
        // 启动metadata的引导程序    
        this.metadata.bootstrap(addresses);
    }

从上述源代码我们可以看出在 KafkaProducer 的构造方法中，如果**metadata为空**就会初始化集群元数据类「**ProducerMetadata**」，然后通过调用 「**this.metadata.bootstrap**」这个方法来启动引导程序，这时 **metaData 对象里并没有具体的元数据信息**，因为客户端还没发送元数据更新的请求，**后面会通过唤醒 Sender 线程进行发送请求获取元数据的**。

这里的 **this.meta** 其实就是 Kafka 内存中的一个对象，底层会做一层缓存，因此并不会一直请求 **Kafka Broker** 端进行获取。

我先给大家总结下元数据初始化及启动的调用关系图，口说无凭，我们来扒开源码瞅一瞅，这样更真实。

![eb7f6196280748eca05468c47b04de26.png][]

通过上述的调用关系图，我们可以看出：

> 1.  ProducerMetadata 类是 MetaData 的子类。
> 2.  Metadata 类是元数据基类，**封装了元数据的具体信息、版本控制、更新标识、响应解析**等。
> 3.  元数据的信息其实最终是保存在**元数据缓存即 MetadataCache** 中，而 它最核心的是 **Cluster** ， 保存了元数据基础信息。

接下来会挨个类展开来进行讲解。

#### 02.1 初探 ProducerMetadata ####

在主线程中初始化了 ProducerMetadata 类的对象，我们先来看看这个类都做了哪些事情。

github 源码地址如下：

> https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java

public class ProducerMetadata extends Metadata {
        // 主题元数据过期时间，如果在这个时间段内未被访问，它就会从缓存中删除。
        private final long metadataIdleMs;
    
        /* Topics with expiry time */
        // map集合，生产者的元数据主题集合，里面保存着
        // 主题和主题过期时间的对应关系，即 topic, nowMs + metadataIdleMs，
        // 过期了的主题会被踢出去。
        private final Map<String, Long> topics = new HashMap<>();
        // 新的主题集合, set集合，即第一次要发送的主题
        private final Set<String> newTopics = new HashSet<>();
        private final Logger log;
        private final Time time;
        public ProducerMetadata(long refreshBackoffMs,
                                long metadataExpireMs,
                                long metadataIdleMs,
                                LogContext logContext,
                                ClusterResourceListeners clusterResourceListeners,
                                Time time) {
            //调用父类 Metadata 的构造函数
            super(refreshBackoffMs, metadataExpireMs, logContext, clusterResourceListeners);
            ....
        }
    }

我们可以看出这里只是调用了父类的构造函数进行类属性的初始化，接下来我们深度分析下 ProducerMetadata 类中的几个比较重要的方法。

02.1.1 add()

public synchronized void add(String topic, long nowMs) {
            // 判断对象是否为空
            Objects.requireNonNull(topic, "topic cannot be null");
            if (topics.put(topic, nowMs + metadataIdleMs) == null) {
                // 添加主题到新主题集合中
                newTopics.add(topic);
                // 更新元数据标记 属于Metadata类方法，后面小节分析
                requestUpdateForNewTopics();
            }
     }

该方法主要用来**向元数据主题集合 topics 中添加主题**，主要用在「**KafkaProducer 主线程**」以及「**Sender 子线程**」中，我们来看下是如何添加的，具体逻辑如下：

> 1.  往元数据主题集合 topics 中添加主题和对应的过期时间（当时时间+过期时间段「**默认值：5分钟**」）。
> 2.  如果元数据主题集合中不存在该主题时，说明是第一次就把该主题添加到新主题集合中。
> 3.  标记要更新新主题元数据的属性字段「**lastRefreshMs**」 、「**needPartialUpdate**」 、「**requestVersion**」，以便后续唤醒 Sender 线程去服务端拉取新主题的元数据。

此时主题被添加到元数据主题主题集合中，**但是如果集合里面数据过期了该怎么办？**接下来我们看另外一个方法是如何判断的。

02.1.2 retainTopic()

public synchronized boolean retainTopic(String topic, boolean isInternal, long nowMs) {
            // 获取该主题的过期时间
            Long expireMs = topics.get(topic);
            // 如果为空表示该主题不在元数据主题集合中
            if (expireMs == null) {
                return false;
            // 判断该主题是否在新集合中
            } else if (newTopics.contains(topic)) {
                return true;
            // 判断是否超过了过期时间
            } else if (expireMs <= nowMs) {
                log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", topic, expireMs, nowMs);
                // 超过后直接从元数据主题集合中删除该主题
                topics.remove(topic);
                return false;
            } else {
                return true;
            }
      }

该方法用来判断元数据中是否该保留该主题，**会在 handleMetadataResponse 即处理元数据响应结果的时候进行调用**，我们来看下它是如何判断的。

> 1.  先判断元数据主题集合中是否存在该主题，如果不存在直接返回false。
> 2.  然后判断该主题是否在新主题集合中，如果存在直接返回true。
> 3.  再判断该主题是否超过了过期时间，如果超过了，就从元数据主题集合中删除该主题，再请求元数据的时候就不用带上该主题，可以有效的减少网络传输数据大小。

02.1.3 requestUpdateForTopic()

public synchronized int requestUpdateForTopic(String topic)   {
         // 如果新主题集合中存在该主题
         if (newTopics.contains(topic)) {
            // 针对新主题集合标记部分更新，并返回版本
            return requestUpdateForNewTopics();
         } else {
            // 全量更新，并返回版本
            return requestUpdate();
         }
      }

该方法用来判断是全量更新元数据还是部分更新元数据，逻辑相对比较简单，**主要用在 KafkaProducer 主线程元数据同步等待时调用**，后续小节会详细分析。

02.1.4 update()

public synchronized void update(int requestVersion, MetadataResponse response, boolean isPartialUpdate, long nowMs) {
            // 调用父类的update方法
            super.update(requestVersion, response, isPartialUpdate, nowMs);
            // 如果新主题集合不为空，则遍历响应元数据找出已经获取元数据的主题，并从新主题集合中删除
            if (!newTopics.isEmpty()) {
                for (MetadataResponse.TopicMetadata metadata : response.topicMetadata()) {
                    newTopics.remove(metadata.topic());
                }
            }
            // 唤醒等待元数据更新完成的线程
            notifyAll();
     }

该方法用来更新生产端元数据的，具体逻辑如下:

> 1.  先调用父类的 update() 方法。
> 2.  然后判断新主题集合是否不为空，如果不为空则遍历响应元数据找出已经获取元数据的主题，并从新主题集合中删除。
> 3.  最后调用 notifyAll() 来唤醒等待元数据更新完成的线程。

从上述 **update()方法** 中可以看出最后调用 **notifyAll()** 来唤醒阻塞的线程， **那么它是如何唤醒的呢**，这就是接下来要分析的方法。

02.1.5 awaitUpdate()

public synchronized void awaitUpdate(final int lastVersion, final long timeoutMs) throws InterruptedException {
            long currentTimeMs = time.milliseconds();
            long deadlineMs = currentTimeMs + timeoutMs < 0 ? Long.MAX_VALUE : currentTimeMs + timeoutMs;
            // 通过调用 time.waitObject 来实现线程阻塞
            time.waitObject(this, () -> {
                // Throw fatal exceptions, if there are any. Recoverable topic errors will be handled by the caller.
                maybeThrowFatalException();
                return updateVersion() > lastVersion || isClosed();
            }, deadlineMs);
            if (isClosed())
                throw new KafkaException("Requested metadata update after close");
      }

该方法用来**实现线程阻塞**的功能，用在主线程中如果发现主题对应的元数据不存在时，阻塞并等待 Sender 线程将元数据更新完成。

重点是调用了 **time.waitObject()** 方法来实现阻塞功能，它的实现还是有一些技巧的，它的底层通过**调用 SystemTime 包里面的 waitObject()** 实现的，源码如下：

public void waitObject(Object obj, Supplier<Boolean> condition, long deadlineMs)

[5117da237f82de642f04445c3af65480.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/31b2e62639ca4dc0be7f4e4421dde824.png
[eb7f6196280748eca05468c47b04de26.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/caba718b990b4c6ba898619e51a94588.png