深度解析kafka生产者发送消息（拉取元数据）

你的名字 2023-01-06 12:59 432阅读 0赞

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

### 文章目录 ###

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

--------------------

### 前言 ###

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

### 1.拉取元数据的时机 ###

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

1.  一个是业务线程获取topic分区信息元数据的时候发现没有，就会立马唤醒sender线程去拉取元数据
2.  sender线程检查已经准备好的batch的时候，发现没有partition 没有leader副本的信息
3.  一个是距离上次更新元数据已经过了5分钟了，这个时候就会去更新下，这个参数配置`metadata.max.age.ms`，默认是5分钟
4.  broker 响应发送消息结果里面带着异常，InvalidMetadataException，就是元数据异常。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
    下面我们就按照kafka消息生产者第一次发送消息的时候，发现没有对应topic的partition信息，唤醒通知sender线程去拉取的这么一个场景

### 2.业务线程等待元数据 ###

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

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

这行代码就是检查消息对应topic 的partition元数据信息是不是存在，如果不存在的话，就会不断尝试获取并且 “通知”sender线程去更新这个元数据信息。  
我们稍微看下这个waitOnMetadata 方法的实现。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
这里面的细节我们不用关注，只需要知道个大概就可以了，可以看到先根据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 这个实现类。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
上面前四行其实就是计算下次更新元数据的时间，我们是第一次更新，然后业务线程又把needUpdate设置成了true，其实这个metadataTimeout就是0 ，也就是立马更新元数据，首先是选取一个node（这个node通俗的理解就是你配置的broker server 地址），选取node的方法就是哪个最闲使用哪个，判断最闲的方法就是找出那种最少等待响应请求数量，就是看看哪个node，等待响应请求数量最少，一开始都没有请求，随便选个了，接着就是调用maybeUpdate 方法发送请求。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
先是看看能不能发送，也就是这个canSendRequest方法，主要就是看看连接建立了没有，等待响应的请求是不是超了（这里默认是最多5个等待响应的请求），如果没有建立连接的话，就会走到下面这个else if 中，进行连接的建立。我们这里就当一切都ok了。  
先是更新metadataFetchInProgress 为true，表示已经有程序在请求更新元数据了，接着就是获取要更新元数据的topic，默认是指更新你需要的那些topic的元数据信息，接着就是封装请求实体，调用doSend方法进行发送。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
this.inFlightRequests.add(request); 这一行就是记录已经发送出去的响应，我们上面判断每个node 等待响应的请求数就是通过inFlightRequests里面请求数量，这个inFlightRequests 里面一个node 对应一个队列，请求对应的响应收到之后，就会将对应的请求移除掉。最后调用selector的send方法进行发送，这里我们就不进去看看，实际上只是将请求信息发送放到了对应channel的暂存区里面，然后关注write事件，后面在进行select的时候，真正发送出去，我们这里就不看了。

接着就是处理接收到的元数据响应  
再次回到NetworkClient的poll 方法中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
接收到的响应都会被放到completedReceives 集合中，通过handleCompletedReceives 方法来处理这堆响应，我们看下这个方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
就是遍历completedReceives 这个集合里面的响应，然后从inFlightRequests 找到之前与之对应的请求，解析响应体，然后调用metadataUpdater的maybeHandleCompletedReceive方法来判断是不是拉取元数据的响应，如果是直接处理了，如果不是就添加到responses 集合中等待后续业务逻辑的处理。  
我们看下metadataUpdater的maybeHandleCompletedReceive方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
判断，如果是拉取metadata 的请求，就会调用handleResponse方法来处理响应  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]  
这里会先metadataFetchInProgress 重置为false，表示没有正在更新元数据的请求。  
接着就是解析元数据响应，从响应中获取cluster集群对象，这个cluster集群就是所谓的元数据的。调用metadata的update方法更新本地缓存。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
这里就是更新了几个时间，然后version版本+1 ，通知监听器，**更新metadata 维护的cluster对象，换成了响应回来的那个**，唤醒那些等待在metadata的线程。  
这个时候，业务线程会被唤醒，然后法相version版本变化了，拿到对应topic的partition元数据信息就接着往下执行了。

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

### 总结 ###

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221119/d9205d4d2cc24abca5c77ddc063f3b56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221119/267ba9bfbffd457dbce1533ac99de5dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221119/c53d3c299e0443068b06ba7efadc928d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210115151842102.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210115152446759.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/2021011516140394.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20210115161528762.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20210115161817230.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20210115161941989.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20210115162112209.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/20210115163528774.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20210115164102613.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70