上一节我们详细解析了生产端拉取元数据的数据结构，其实也可以看出来，无非就是topic ,partition,node之间的对应关系，同时我们在第二章节KafkaProducer初始化的时候有过如下的代码：
this.metadata.update(Cluster.bootstrap(addresses), Collections.<String>emptySet(), time.milliseconds());
ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config);
实际上我们看了一下update方法，在kafakPrducer初始化的时候并没有真正的
去拉取topic的元数据，但是他肯定是对集群元数据做了一个初始化的，把你配置的那些broker地址转化为了Node,放在Cluster对象实例化中。
那么本章节就真正的去分析一下，Kafka是如何拉取元数据的。
在本课节最开始的时候，就给出一个producer发送消息的api demo,其中如下：
ProducerRecord record = new ProducerRecord<String, String>("topic1", "userName", "lc");
//发送记录
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if(Objects.isNull(exception))
System.out.println("success");
}
});
前两章节已经分析了KafkaProducer的初始化，接着我们看发送消息的时候其实是调用了send()方法：
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
// intercept the record, which can be potentially modified; this method does not throw exceptions
ProducerRecord<K, V> interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
return doSend(interceptedRecord, callback);
}
这里看到最后其实调用了doSend()方法，在这里我们多说一句，上面是不是有一个拦截器的加载判断，为什么一开始就要做拦截器的配置呢？很简单，无用的数据最早过滤，避免后面无用的数据做序列化操作。
接下来看doSend()方法，这里我们还是一行一行的进行分析：
//这里同步拉取元数据
ClusterAndWaitTime clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
//这里我们已经拉取到元数据了，cluster里也有对应的值了
Cluster cluster = clusterAndWaitTime.cluster;
上面在dosend()前面马上调用了waitOnMetadata()方法，可以看出来这是一个等待同步元数据的方法，为什么在发消息刚开始就要调用这个方法呢，大家想一下，要是没有获取到这个topic的元数据，你的消息发到哪里去？
waitOnMetadata()中有一个参数，我们需要明白：
max.block.ms:默认1分钟，决定了你调用send方法的时候，最多会被阻塞多长时间
接下来查看waitOnMetadata()这个方法：
metadata.add(topic);
Cluster cluster = metadata.fetch();
Integer partitionsCount = cluster.partitionCountForTopic(topic);
// Return cached metadata if we have it, and if the record's partition is either undefined
// or within the known partition range
/**
* 这里partitionsCount是根据当前topic,去元数据metadata里找到topic的partitions数量
* 当不为null，说明找到了
* 这里partition通常为null:
* ProducerRecord record =
* new ProducerRecord<String, String>("topic1", "userName", "lc");
* //发送记录
* producer.send(record);
* 这里找到对应的元数据就直接返回，花的时间是0
*/
if (partitionsCount != null && (partition == null || partition < partitionsCount))
return new ClusterAndWaitTime(cluster, 0);
我们可以想一下，当第一次进行这个方法的时候，上面的代码其实metadata里没有我们的元数据信息的，那么我们接着向下走:
long begin = time.milliseconds();
//余下多少时间，默认值给的是最多等待的时间
long remainingWaitMs = maxWaitMs;
//已经花了多少时间
long elapsed;
do {
log.trace("Requesting metadata update for topic {}.", topic);
metadata.add(topic);
//获取当前元数据的版本
//在producer管理元数据的时候，对于他来说，数据是有版本号的
//每次成功更新元数据，都会递增这个版本号
//把needUpdate设置为true
int version = metadata.requestUpdate();
//唤醒sender线程,开始执行拉取元数据操作，
//拉取元数据是由Sender线程完成的
sender.wakeup();
try {
//同步等待sender线程拉取元数据
metadata.awaitUpdate(version, remainingWaitMs);
} catch (TimeoutException ex) {
// Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs
throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
}
// 获取一下集群的元数据，(上面可能成功更新了集群的元数据信息)
cluster = metadata.fetch();
//计算一下拉取元数据花的时间
elapsed = time.milliseconds() - begin;
if (elapsed >= maxWaitMs)
throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
//如果已经获取了元数据，但是发现topic没有授权
if (cluster.unauthorizedTopics().contains(topic))
throw new TopicAuthorizationException(topic);
remainingWaitMs = maxWaitMs - elapsed;
//如果这个值不为null,就要跳出do while
partitionsCount = cluster.partitionCountForTopic(topic);
} while (partitionsCount == null);
上面这段代码非常的核心，可以看出是一个do while循环，里面做了两个重要的事情，
其一，sender.wakeup(); 这个是唤醒sender线程，我们知道在前两章节在初始化的时候新建了一个sender线程，sender线程就是用于拉取元数据的，这里唤醒就是让他开始执行拉取工作
其二，metadata.awaitUpdate(version, remainingWaitMs); 这里不难看出是一个等待更新元数据的方法，其实我们看这个方法的实现，就是一个阻塞主线程的操作:
public synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs) throws InterruptedException {
if (maxWaitMs < 0) {
throw new IllegalArgumentException("Max time to wait for metadata updates should not be < 0 milliseconds");
}
long begin = System.currentTimeMillis();
long remainingWaitMs = maxWaitMs;
/**
* 这里判断上一个标志位大于等于现在的version时，表示sender还没有拉取成功元数据
* 这里一直循环下去，等待SENDER拉取元数据，当没有时间就报异常
*/
//this.version <= lastVersion 表示元数据还没有拉取成功
while (this.version <= lastVersion) {
AuthenticationException ex = getAndClearAuthenticationException();
if (ex != null)
throw ex;
if (remainingWaitMs != 0)
//这里虽然还没有可能sender线程的源码,但是我们应该可以猜想它一定会有这样的操作
//如何更新元数据成功了，那么一定会唤醒该线程
//TODO 这里等待
wait(remainingWaitMs);
long elapsed = System.currentTimeMillis() - begin;
if (elapsed >= maxWaitMs)
throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
remainingWaitMs = maxWaitMs - elapsed;
}
}
看到上面的源码我们可以看出，在while循环里，当前版本小于等于最后的版本号，表示元数据还没有更新，里后面调用了wait方法，这里阻塞了主线程。
接合上面的方法waitOnMetadata(),我们明白了，就是唤醒sender线程去拉取元数据，同时阻塞当前线程，一直等到sender拉取元数据回来，或者等待timeout，才会接着工作，当前如果timeout了，这里就报异步了，那么timeout默认多长时间呢，是哪个参数呢，正是我们上面给出的值max.block.ms,默认一分钟。
接着我们看下面的源码：
// 获取一下集群的元数据，(上面可能成功更新了集群的元数据信息)
cluster = metadata.fetch();
//计算一下拉取元数据花的时间
elapsed = time.milliseconds() - begin;
if (elapsed >= maxWaitMs)
throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
//如果已经获取了元数据，但是发现topic没有授权
if (cluster.unauthorizedTopics().contains(topic))
throw new TopicAuthorizationException(topic);
remainingWaitMs = maxWaitMs - elapsed;
//如果这个值不为null,就要跳出do while
partitionsCount = cluster.partitionCountForTopic(topic);
如果程序走到这里，说明 我们已经成功拉取元数据回来，并且把值赋给了cluster,这里有一个值elapsed 要注意，就是花了多少时间去拉取元数据的，这个时间partitionsCount 不为null了，那么上面的代码do while循环就跳出了，最后组装成ClusterAndWaitTime对象及花费时间返回：
return new ClusterAndWaitTime(cluster, elapsed);
这里我是从主线程观点来分析执行流程的，并没有看到sender是如何把元数据拉回来的，事实上拉取broker上的元数据一定要有网络连接，这个我们下一章节详细分析
网友评论