3.Kafka源码深入解析之元数据结构

前面我们讲到，在KafkaProducer初始化时，初始化了一个非常核心的组件Metadata

this.metadata = new Metadata(retryBackoffMs, config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG),
                    true, true, clusterResourceListeners);
this.metadata.update(Cluster.bootstrap(addresses), Collections.<String>emptySet(), time.milliseconds());

这个组件前面也详细说明过，是去broker上 拉取一次集群的元数据过来的，我们知道当一条消息从producer端发送到broker时，一定是先必须知道这条消息所在的leader partitions node的，其实就是要事先知道元数据的，那么看上面的代码，我详细分析一下它的工作原理，以前元数据的数据结构

1. 首先我们看一下Metadata这个类

public final class Metadata {

    private static final Logger log = LoggerFactory.getLogger(Metadata.class);

    public static final long TOPIC_EXPIRY_MS = 5 * 60 * 1000;
    private static final long TOPIC_EXPIRY_NEEDS_UPDATE = -1L;
    // metadata 更新失败时,为避免频繁更新 meta,最小的间隔时间,默认 100ms
    private final long refreshBackoffMs;
    // metadata 的过期时间, 默认 60,000ms
    private final long metadataExpireMs;
    // 每更新成功1次，version自增1,主要是用于判断 metadata 是否更新
    private int version;
    // 最近一次更新时的时间（包含更新失败的情况）
    private long lastRefreshMs;
    // 最近一次成功更新的时间（如果每次都成功的话，与前面的值相等,
    // 否则，lastSuccessulRefreshMs < lastRefreshMs)
    private long lastSuccessfulRefreshMs;
    private AuthenticationException authenticationException;
    // 集群中一些 topic 的信息
    private Cluster cluster;
    // 是都需要更新 metadata
    private boolean needUpdate;
    // topic 与其过期时间的对应关系
    private final Map<String, Long> topics;
    private final List<Listener> listeners;
    //当接收到 metadata 更新时, ClusterResourceListeners的列表
    private final ClusterResourceListeners clusterResourceListeners;
    // 是否强制更新所有的 metadata
    private boolean needMetadataForAllTopics;
    // 默认为 true, Producer 会定时移除过期的 topic,consumer 则不会移除
    private final boolean allowAutoTopicCreation;
    private final boolean topicExpiryEnabled;

包含的各字段意思上面注解非常详细，大家可以看一下，其中有一个Cluster cluster是什么呢，我们在看：

public final class Cluster {

    private final boolean isBootstrapConfigured;
    /**
     *  关于 topic 的详细信息（leader 所在节点、replica 所在节点、isr 列表）都是在 Cluster 实例中保存的
     * node，Kafka broker节点，一台机器，
     * 包括了：
     *   private final int id; --broker.id
     *     private final String idString;
     *     private final String host; --主机
     *     private final int port;   --port
     *     private final String rack;  --机架
     *
     *   broker.id 与 node 的对应关系；
     *   topic 与 partition （PartitionInfo）的对应关系；
     *   node 与 partition （PartitionInfo）的对应关系。
     */
    //一个kafka集群每个节点
    private final List<Node> nodes;
    /**
     * 没有被授权访问的topic列表，如果你的客户端没有被授权访问某个topic,
     * 就会放在这个列表里
     */
    //没有授权的topic列表
    private final Set<String> unauthorizedTopics;
    //内置的topic列表
    private final Set<String> internalTopics;
    //controller所在node
    private final Node controller;
    /**
     * 代码一个分区，里面就是它的topic名字，以及他在topic里的分区号
     * ，每个分区都有多个副本 ，leader副本 在哪个broker上，follower在哪些
     * broker里，isr列表都在哪
     *    private final String topic;
     *     private final int partition;
     *     private final Node leader;
     *     private final Node[] replicas;
     *     private final Node[] inSyncReplicas;
     *     private final Node[] offlineReplicas;
     */
    //每个分区的信息
    private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition;
    /**
     * 每个topic有哪些分区
     */
    private final Map<String, List<PartitionInfo>> partitionsByTopic;
    /**
     * 每个topic有哪些当前可用的分区，如果某个分区没有leaeer是存活的，那
     * 此时这个分区不可用
     */
    private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;
    /**
     * 每个broker上有哪些分区
     */
    private final Map<Integer, List<PartitionInfo>> partitionsByNode;
    /**
     * broker.id ->Node
     * brokerid与node对应关系，如broker0 --> node1
     */
    private final Map<Integer, Node> nodesById;
    //kafka的集群id
    private final ClusterResource clusterResource;

这里明白了，这些元数据里的结构了吧，其实可以看到好多信息是冗余了，这正是为了查询数据方便。
接下来我们看一下下面的代码：

this.metadata.update(Cluster.bootstrap(addresses), Collections.<String>emptySet(), time.milliseconds());

可以看到是去更新元数据的，那么它的实现原理是什么呢？

public synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now) {
        Objects.requireNonNull(cluster, "cluster should not be null");

        this.needUpdate = false;
        this.lastRefreshMs = now;
        this.lastSuccessfulRefreshMs = now;
        this.version += 1;

        //默认为true
        if (topicExpiryEnabled) {
            // Handle expiry of topics from the metadata refresh set.
            for (Iterator<Map.Entry<String, Long>> it = topics.entrySet().iterator(); it.hasNext(); ) {
                Map.Entry<String, Long> entry = it.next();
                long expireMs = entry.getValue();
                if (expireMs == TOPIC_EXPIRY_NEEDS_UPDATE)
                    entry.setValue(now + TOPIC_EXPIRY_MS);
                else if (expireMs <= now) {
                    it.remove();
                    log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", entry.getKey(), expireMs, now);
                }
            }
        }

        for (Listener listener: listeners)
            listener.onMetadataUpdate(cluster, unavailableTopics);

        String previousClusterId = cluster.clusterResource().clusterId();

        //默认是false
        if (this.needMetadataForAllTopics) {
            // the listener may change the interested topics, which could cause another metadata refresh.
            // If we have already fetched all topics, however, another fetch should be unnecessary.
            this.needUpdate = false;
            this.cluster = getClusterForCurrentTopics(cluster);
        } else {
            /**
             * 直接把传进来的Cluser对象赋值给metadata,其实只有节点信息kafka集群地址
             * 初使化的时候，update方法并没有到broker去拉取元数据
             */
            //这里cluster其实只是获取了我们配置的kafka addresses
            this.cluster = cluster;
        }

        // The bootstrap cluster is guaranteed not to have any useful information
        if (!cluster.isBootstrapConfigured()) {
            String clusterId = cluster.clusterResource().clusterId();
            if (clusterId == null ? previousClusterId != null : !clusterId.equals(previousClusterId))
                log.info("Cluster ID: {}", cluster.clusterResource().clusterId());
            clusterResourceListeners.onUpdate(cluster.clusterResource());
        }
        notifyAll();
        log.debug("Updated cluster metadata version {} to {}", this.version, this.cluster);
    }

可以看到，其实在初始化的时候，metadata并没有去broker拉取元数据信息，只是把我们配置的broker addrss设置进去了。

这里有一个notifyAll()，大家看到了吧，同时看上面的方法，是不是一个synchronized 来修饰的，这里是什么意思呢？
其实这里大家想一下，synchronized 加锁后，同一时间我们只有一个线程可以进入update()来执行，这里也说明是线程安全的，其他的线程进入等待，上面的notifyAll不就是释放锁，唤醒别的线程吗

上面的needMetadataForAllTopics 意思很显示，是否需要把集群所有的topic的元数据都拉取过来呢，默认是false,这里也可以知道我们只需要我们的topic的信息。