深度解析kafka生产者发送消息（拉取元数据）-蒲公英云

原创不易，转载请注明出处

文章目录

- - 前言
  - 1.拉取元数据的时机
  - 2.业务线程等待元数据
  - 3.sender线程拉取元数据
  - 总结

前言

在之前的文章中我们介绍了，kafka 消息生产者端有2种线程，一种是业务线程，一种是sender线程，这个业务线程是支持多个的，也就是支持并发发送，然后这个sender 一个KafkaProducer实例就一个，业务线程不停的往一个内存缓冲区（消息累加器，因为这里要把多个消息打成batch）追加消息，然后sender线程干的事就多了，发送消息，接收响应，建立连接，获取元数据。在解析sender发送消息之前我们先要有topic 分区的元数据信息，这样子才能知道消息发送到的partition在哪台机器上，本文先介绍下拉取元数据的时机，也就是什么情况下会拉取元数据，接着就是看看业务线程是怎样等着sender线程拉取消息的。

1.拉取元数据的时机

（需要注意的，并不是KafkaProducer初始化完成就会去拉取元数据，也不是一个单独的定时任务管着拉取（RocketMQ是采用单独定时任务来拉取的））

一个是业务线程获取topic分区信息元数据的时候发现没有，就会立马唤醒sender线程去拉取元数据
sender线程检查已经准备好的batch的时候，发现没有partition 没有leader副本的信息
一个是距离上次更新元数据已经过了5分钟了，这个时候就会去更新下，这个参数配置metadata.max.age.ms，默认是5分钟
broker 响应发送消息结果里面带着异常，InvalidMetadataException，就是元数据异常。

下面我们就按照kafka消息生产者第一次发送消息的时候，发现没有对应topic的partition信息，唤醒通知sender线程去拉取的这么一个场景

2.业务线程等待元数据

在调用KafkaProducer#send方法发送消息的时候，先会经过拦截器的处理，接着就是调用doSend方法来进行选择分区，key，value序列化，追加到消息缓冲区（消息累加器）中。在doSend方法最前面，有这么一行代码

long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);

这行代码就是检查消息对应topic 的partition元数据信息是不是存在，如果不存在的话，就会不断尝试获取并且 “通知”sender线程去更新这个元数据信息。
我们稍微看下这个waitOnMetadata 方法的实现。
在这里插入图片描述
这里面的细节我们不用关注，只需要知道个大概就可以了，可以看到先根据topic 获取一下看看本地有没有partition元数据信息的缓存，如果有的话，就直接return返回了，如果没有程序接着往下执行，一个whlie循环，只要本地缓存中获取不到就不停下，我们看下while循环里面，

int version = metadata.requestUpdate();
 sender.wakeup();
 metadata.awaitUpdate(version, remainingWaitMs);

第一行是很重要的，在metadata里面设置了needUpdate = true，这就相当于告诉sender线程元数据需要更新下。
第二行，唤醒sender线程，这里可以先略过去，如果你知道这个sender都干什么了的话，你就能知道这里为啥要唤醒这个sender线程，其实sender线程就是基于java nio事件驱动网络模型实现的发送消息，接收响应这些东西，如果对java nio 了解的话，就会知道，在select的时候是可以阻塞等待事件到来的，就是为了唤醒这个步骤的堵塞。
第三行，metadata.awaitUpdate(version, remainingWaitMs); 这行就是将version传了进去，然后version不变的话就会一直循环，sender拉取到了元数据信息，然后更新本地的缓存的时候，会将这个version+1，这个时候，它就会逃离这个循环。

3.sender线程拉取元数据

sender是个Runnable ，我们直接看它的run方法就可以了。这里不会过多的介绍sender线程处理逻辑，有些内容点到为止，到拉取元数据的部分我们会详细介绍下。
在Sender的run方法中，最下面有一行 this.client.poll(pollTimeout, now);这个poll方法就是处理建立连接，发送请求，处理响应，拉取元数据，然后在它的poll方法的第一行 long metadataTimeout = metadataUpdater.maybeUpdate(now);就是在检查拉取元数据。我们详细看下这个maybeUpdate方法，元数据更新器metadataUpdater 默认是使用的DefaultMetadataUpdater 这个实现类。
在这里插入图片描述
上面前四行其实就是计算下次更新元数据的时间，我们是第一次更新，然后业务线程又把needUpdate设置成了true，其实这个metadataTimeout就是0 ，也就是立马更新元数据，首先是选取一个node（这个node通俗的理解就是你配置的broker server 地址），选取node的方法就是哪个最闲使用哪个，判断最闲的方法就是找出那种最少等待响应请求数量，就是看看哪个node，等待响应请求数量最少，一开始都没有请求，随便选个了，接着就是调用maybeUpdate 方法发送请求。
在这里插入图片描述
先是看看能不能发送，也就是这个canSendRequest方法，主要就是看看连接建立了没有，等待响应的请求是不是超了（这里默认是最多5个等待响应的请求），如果没有建立连接的话，就会走到下面这个else if 中，进行连接的建立。我们这里就当一切都ok了。
先是更新metadataFetchInProgress 为true，表示已经有程序在请求更新元数据了，接着就是获取要更新元数据的topic，默认是指更新你需要的那些topic的元数据信息，接着就是封装请求实体，调用doSend方法进行发送。
在这里插入图片描述
this.inFlightRequests.add(request); 这一行就是记录已经发送出去的响应，我们上面判断每个node 等待响应的请求数就是通过inFlightRequests里面请求数量，这个inFlightRequests 里面一个node 对应一个队列，请求对应的响应收到之后，就会将对应的请求移除掉。最后调用selector的send方法进行发送，这里我们就不进去看看，实际上只是将请求信息发送放到了对应channel的暂存区里面，然后关注write事件，后面在进行select的时候，真正发送出去，我们这里就不看了。

接着就是处理接收到的元数据响应
再次回到NetworkClient的poll 方法中。
在这里插入图片描述
接收到的响应都会被放到completedReceives 集合中，通过handleCompletedReceives 方法来处理这堆响应，我们看下这个方法。

就是遍历completedReceives 这个集合里面的响应，然后从inFlightRequests 找到之前与之对应的请求，解析响应体，然后调用metadataUpdater的maybeHandleCompletedReceive方法来判断是不是拉取元数据的响应，如果是直接处理了，如果不是就添加到responses 集合中等待后续业务逻辑的处理。
我们看下metadataUpdater的maybeHandleCompletedReceive方法
在这里插入图片描述
判断，如果是拉取metadata 的请求，就会调用handleResponse方法来处理响应

这里会先metadataFetchInProgress 重置为false，表示没有正在更新元数据的请求。
接着就是解析元数据响应，从响应中获取cluster集群对象，这个cluster集群就是所谓的元数据的。调用metadata的update方法更新本地缓存。
在这里插入图片描述
这里就是更新了几个时间，然后version版本+1 ，通知监听器，更新metadata 维护的cluster对象，换成了响应回来的那个，唤醒那些等待在metadata的线程。
这个时候，业务线程会被唤醒，然后法相version版本变化了，拿到对应topic的partition元数据信息就接着往下执行了。

最后看下这个元数据都有啥，也就是Cluster类是什么样子的。
在这里插入图片描述
可以看到就是一堆partition 与partitionInfo，topic与partitionInfo，node 与partitionInfo的对应关系，最上面还有nodes这个所有node的集合信息，这里会将所有的node都返回回来，比如说你集群里面有3个broker ，然后你消息生产者参数bootstrap.servers只配置了某一台的信息，更新一次之后，就会把其他的node信息全带过来了。
我们这里需要看下partitionInfo里面都有啥东西
在这里插入图片描述
可以看到，partition所属的topic ，partition id ，leader 副本所在的node，所有副本对应node信息，isr副本对应node信息。

总结

本文主要是介绍了kafka消息生产者端获取元数据的流程，我们首先介绍了拉取元数据的时机，一个是用的时候发现没有对应的元数据，一个是每5分钟就会更新一下，一个是发送出去的消息请求，然后broker响应了一些关于元数据的异常，这个时候会去更新下等等，接着我们有介绍了业务线程因为没有元数据信息等待获取，最后介绍了sender线程拉取元数据的请求，包括请求的建立，响应的处理等等。