前面我们讲到，在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的，其实就是要事先知道元数据的，那么看上面的代码，我详细分析一下它的工作原理，以前元数据的数据结构
- 首先我们看一下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的信息。
网友评论