一、前言
我们上一篇说了 聊聊 Kafka：Producer 源码解析，这一篇我们来说下 Producer Metadata 的读取与更新机制。上一篇从宏观上介绍了 Producer 的宏观模型，其中通过 waitOnMetadata() 方法获取 topic 的 metadata 信息这一块东西很多，所以单独拎一篇出来讲。
二、Metadata
2.1 什么是 Metadata
Metadata 是指 Kafka 集群的元数据，包含了 Kafka 集群的各种信息，直接看源码便可知：
public class Metadata implements Closeable { private final Logger log; // retry.backoff.ms: 默认值为100ms，它用来设定两次重试之间的时间间隔，避免无效的频繁重试。 private final long refreshBackoffMs; // metadata.max.age.ms: 默认值为300000，如果在这个时间内元数据没有更新的话会被强制更新。 private final long metadataExpireMs; // 更新版本号，每更新成功1次，version自增1,主要是用于判断metadata是否更新 private int updateVersion; // 请求版本号，每发送一次请求，version自增1 private int requestVersion; // 上一次更新的时间（包含更新失败） private long lastRefreshMs; // 上一次更新成功的时间 private long lastSuccessfulRefreshMs; private KafkaException fatalException; // 非法的topics private Set<String> invalidTopics; // 未认证的topics private Set<String> unauthorizedTopics; // 元数据信息的Cache缓存 private MetadataCache cache = MetadataCache.empty(); private boolean needFullUpdate; private boolean needPartialUpdate; // 会收到metadata updates的Listener列表 private final ClusterResourceListeners clusterResourceListeners; private boolean isClosed; // 存储Partition最近一次的leaderEpoch private final Map<TopicPartition, Integer> lastSeenLeaderEpochs; }
MetadataCache：Kafka 集群中关于 node、topic 和 partition 的信息。（是只读的）
public class MetadataCache { private final String clusterId; private final Map<Integer, Node> nodes; private final Set<String> unauthorizedTopics; private final Set<String> invalidTopics; private final Set<String> internalTopics; private final Node controller; private final Map<TopicPartition, PartitionMetadata> metadataByPartition; private Cluster clusterInstance; }
关于 topic 的详细信息（leader 所在节点、replica 所在节点、isr 列表）都是在 Cluster 实例中保存的。
// 保存了Kafka集群中部分nodes、topics和partitions的信息 public final class Cluster { private final boolean isBootstrapConfigured; // node 列表 private final List<Node> nodes; // 未认证的topics private final Set<String> unauthorizedTopics; // 非法的topics private final Set<String> invalidTopics; // kafka内置的topics private final Set<String> internalTopics; private final Node controller; // partition对应的信息，如：leader所在节点、所有的副本、ISR中的副本、offline的副本 private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition; // topic和partition信息的对应关系 private final Map<String, List<PartitionInfo>> partitionsByTopic; // topic和可用partition(leader不为null)的对应关系 private final Map<String, List<PartitionInfo>> availablePartitionsByTopic; // node和partition信息的对应关系 private final Map<Integer, List<PartitionInfo>> partitionsByNode; // 节点id与节点的对应关系 private final Map<Integer, Node> nodesById; // 集群信息，里面只有一个clusterId private final ClusterResource clusterResource; }
// topic-partition: 包含 topic、partition、leader、replicas、isr public class PartitionInfo { private final String topic; private final int partition; private final Node leader; private final Node[] replicas; private final Node[] inSyncReplicas; private final Node[] offlineReplicas; }
看源码不难理解 Metadata 的主要数据结构，我们大概总结下包含哪些信息：
- 集群中有哪些节点；
- 集群中有哪些 topic，这些 topic 有哪些 partition；
- 每个 partition 的 leader 副本分配在哪个节点上，follower 副本分配在哪些节点上；
- 每个 partition 的 AR 有哪些副本，ISR 有哪些副本；
2.2 Metadata 的应用场景
Metadata 在 Kafka 中非常重要，很多场景中都需要从 Metadata 中获取数据或更新数据，例如：
- KafkaProducer 发送一条消息到指定的 topic 中，需要知道分区的数量，要发送的目标分区，目标分区的 leader，leader 所在的节点地址等，这些信息都要从 Metadata 中获取。
- 当 Kafka 集群中发生了 leader 选举，节点中 partition 或副本发生了变化等，这些场景都需要更新Metadata 中的数据。
三、Producer 的 Metadata 更新流程
Producer 在调用 doSend() 方法时，第一步就是通过 waitOnMetadata 方法获取该 topic 的 metadata 信息。
总结一下以上代码：
- 首先会从缓存中获取 cluster 信息，并从中获取 partition 信息，如果可以取到则返回当前的 cluster 信息，如果不含有所需要的 partition 信息时就会更新 metadata；
- 更新 metadata 的操作会在一个 do ….while 循环中进行，直到 metadata 中含有所需 partition 的信息，该循环中主要做了以下事情：
- 调用 metadata.requestUpdateForTopic() 方法来获取 updateVersion，即上一次更新成功时的 version，并将 needUpdate 设为 true，强制更新；
- 调用 sender.wakeup() 方法来唤醒 Sender 线程，Sender 线程中又会唤醒 NetworkClient 线程，在 NetworkClient 中会对 UpdateMetadataRequest 请求进行操作，待会下面会详细介绍；
- 调用 metadata.awaitUpdate(version, remainingWaitMs) 方法来等待 metadata 的更新，通过比较当前的 updateVersion 与步骤 1 中获取的 updateVersion 来判断是否更新成功；
3.1 org.apache.kafka.clients.NetworkClient#poll
上面提到调用 sender.wakeup() 方法来唤醒 Sender 线程，Sender 线程中又会唤醒 NetworkClient 线程，在 NetworkClient 中会对 UpdateMetadataRequest 请求进行操作。在 NetworkClient 中真正处理请求的是 NetworkClient.poll() 方法，接下来让我们通过分析源码来看下 NetworkClient 是如何处理请求的。
3.2 org.apache.kafka.clients.NetworkClient.DefaultMetadataUpdater#maybeUpdate(long)
我们来看下 metadata 是如何更新的
这里你可能会问，老周啊，最小负载节点是啥呀？
别急，我们来看下面这张图，你就理解了。
LeastLoadedNode 指 Kafka 集群中所有 Node 中负载最小的那一个 Node，它是由每个 Node 在 InFlightRequests 中还未确定的请求数决定的，未确定的请求越少则负载越小。如上图所示，Node1 即为 LeastLoadedNode。
3.3 org.apache.kafka.clients.Metadata#updateRequested
下次更新元数据信息的时间：当前 metadata 信息即将到期的时间即 timeToExpire 和 距离允许更新 metadata 信息的时间 即 timeToAllowUpdate 中的最大值。
timeToExpire：needUpdate 为 true，表示强制更新，此时该值为 0；否则的话，就按照定时更新时间，即元数据信息过期时间（默认是 300000 ms 即 5 分钟）进行周期性更新。
timeToAllowUpdate：默认就是 refreshBackoffMs 的默认值，即 100 ms。
3.4 org.apache.kafka.clients.NetworkClient.DefaultMetadataUpdater#maybeUpdate(long, org.apache.kafka.common.Node)
我们继续跟一下 maybeUpdate 方法：
因此，每次 producer 请求更新 metadata 时，会有以下几种情况：
- 通道已经 ready，node 可以发送请求，那么就直接发送请求。
- 如果该 node 正在建立连接，则直接返回。
- 如果该 node 还没建立连接，则向 broker 初始化连接。
而 KafkaProducer 线程一直是阻塞在两个 while 循环中的，直到 metadata 更新：
- sender 线程第一次调用 poll，初始化与 node 的连接。
- sender 线程第二次调用 poll，发送 metadata 请求。
- sender 线程第三次调用 poll，获取 metadataResponse，并更新 metadata。
3.5 接收 Server 端的响应，更新 Metadata 信息
handleCompletedReceives 是如何处理任何已完成的接收响应，如下：
之后进一步调用 handleSuccessfulResponse。
四、总结
Metadata 会在下面两种情况下进行更新：
- 强制更新：调用 Metadata.requestUpdate() 将 needFullUpdate 置为 true 来强制更新。
- 周期性更新：通过 Metadata 的 lastSuccessfulRefreshMs 和 metadataExpireMs 来实现，一般情况下，默认周期时间就是 metadataExpireMs，5 分钟时长。
在 NetworkClient 的 poll() 方法调用时，会去检查两种更新机制，只要达到一种，就会触发更新操作。
Metadata 的强制更新会在以下几种情况下进行：
- initConnect 方法调用时，初始化连接；
- poll() 方法中对 handleDisconnections() 方法调用来处理连接断开的情况，这时会触发强制更新；
- poll() 方法中对 handleTimedOutRequests() 来处理请求超时时；
- 发送消息时，如果无法找到 partition 的 leader；
- 处理 Producer 响应（handleProduceResponse），如果返回关于 Metadata 过期的异常，比如：没有 topic-partition 的相关 meta 或者 client 没有权限获取其 metadata。
强制更新主要是用于处理各种异常情况。
好了，Producer Metadata 读取与更新机制就说到这，我们下一期再见。
欢迎大家关注我的公众号【老周聊架构】，Java后端主流技术栈的原理、源码分析、架构以及各种互联网高并发、高性能、高可用的解决方案。
