深度解析kafka生产者发送消息（原理篇）

痛定思痛。 2023-01-01 06:47 273阅读 0赞

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

### 文章目录 ###

*   *   *  前言
         *  1.回顾kafka生产者核心组件
         *  2.图解kafka发送消息原理
         *  总结

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

### 前言 ###

在《[深度解析kafka生产者启动流程][kafka]》一文中，我们大体上看了下kafkaProducer的初始化，初始化的时候都创建了哪些核心的组件，包括Metadata，Partitioner，RecordAccumulator，NetworkClient，Sender。本文主要是先回顾下kafka生产者端一些核心的组件，然后再介绍下kafka生产者端发送消息的整个流程，看看它那些组件是怎样运作的。

### 1.回顾kafka生产者核心组件 ###

1.  Metadata 这个组件是管理元数据的，包括集群有哪些node ，然后topic有哪些partition，每个partition的leader 在哪个node 上，follower在哪些node上等等。
2.  Partitioner 是个选择partition的组件，比如说你一个topic 有多个partition，它就会按照规则为某条消息选择一个partition。
3.  RecordAccumulator 缓冲区，累加器，它会将多个发往同一个partition的消息打成一个batch，这样就能够减少传输次数，一次传输成发送多条消息。达到批量发送的效果。
4.  NetworkClient 网络通信组件，封装了jdk nio一套api，建立连接，发送消息，接收响应，断开连接等等。
5.  Sender 就是将消息从RecordAccumulator 缓冲区里面将消息拿出来，然后使用NetworkClient 这个网络通信组件进行消息的发送。

### 2.图解kafka发送消息原理 ###

这里我们脑子里要有两个线程在不停的运行，一个是业务线程（现实中这个业务线程是有多个的，KafkaProducer也是支持多线程并发发送消息的，这里为了方便我们就说一个业务线程），它会不停的调用KafkaProducer 的send方法进行发送我们产生的消息，还有一个是就是那个sender 线程了，这个线程就是KafkaProducer 初始化的时候创建的。这里我们首先会介绍下这个业务线程在怎样发送消息的。然后再介绍下这个sender线程是怎样从RecordAccumulator 缓冲区中获取消息，然后将消息发送给broker 的。  
当我们业务线程调用KafkaProducer 的send方法发送消息的时候，先会经过一堆拦截器来过滤我们的消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]

接着就会等着获取关于这个topic 的元数据，我们这里就认为元数据已经获取到了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
接着往下走就会把你指定的key 跟value 进行序列化，这个序列化其实就是使用我们指定的那个两个序列化器来进行，就是将字符串啥的转成二进制。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
接着就是使用partitioner 来指定这个消息发送到哪个partition上去，如果你指定了这个消息的partition ，就会按照你指定的来，如果你没有指定的话，就会判断你这个key是否为null，如果你这个key不是null ，就根据key做hash 来模partition的数量，如果你这个key是null的话就使用轮询的方式来选择partition。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
接着就会将消息追加到RecordAccumulator 这个缓冲区中， 在这个内存缓冲区中，一个topic下的partition 对应着一个队列，队列中的元素是batch ，这个batch可以理解为是一堆消息，一批消息。它首先是先找到这个消息topic-partition对应的队列，如果没有的话就创建，然后将消息追加到队列队尾的那个batch中，如果队尾的那个batch不存在或者是满了（满了是由batch.size这个参数控制）就会新创建个batch ，将消息追加到这个batch中，然后添加batch到队列队尾。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
只要追加消息到batch成功，这个业务线程发送这条消息的任务就算完事了，就可以返回继续发送其他消息了。  
业务线程发送消息大体运行流程就是上面那个样子，接下来我们大体介绍下它的io线程，也就是sender是怎样运行的。

sender 线程首先会遍历RecordAccumulator 中的所有partition对应的队列，找出准备好的node（node可以理解是kafka broker），怎样找出准备好的node？其实就是遍历每个队列，找出对头的那个batch，然后看看对头这个batch满没满，或者是不是重试的，是否超过了linger.ms这个参数配置的时候还没有发送出去，只要有满足条件的认为这个队列中的batch准备好了，等待发送，然后找出这个队列对应partition对应的node。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
接着就会遍历这堆准备好的node，然后看看与这个node连接建立了没，没有建立连接就请求建立连接啥的，再就是看看与这个node连接的channel是否准备好，发送出去的请求还没响应的是否大于"max.in.flight.requests.per.connection"这个参数设置的值。  
没建立连接，没准备好channel，发送出去的请求还没响应的大于"max.in.flight.requests.per.connection"这个参数设置的值 就认为这个node没准备好，就从node 列表中移除掉这个node。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
然后带着剩余准备好的node，去RecordAccumulator 内存缓冲区中，取出这个node下面partition 对应队列的batch，也就是nodeId 对应这个batch集合。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
将获取出来的node 与batch集合对应关系的map进行遍历，封装成每个node的请求。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]  
遍历这堆请求，找到NetworkClient这个组件进行发送，这个发送其实就是放到每个node对应的channel中了，没有真实的发送，只是给channel一个OP\_WRITE事件，当这个sender线程进行poll的时候就会拿到这个事件，然后真正处理发送，将消息写到channel中，我们这里就当他发送了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
其实这个sender 干的事情非常的多，发送消息请求，发送拉取元数据的，接收响应，建立连接，断开连接，封装请求等等，我们这里就不详细展开了。

### 总结 ###

本文主要是回顾了一下KakfaProducer 消息生产者端一些核心组件，然后以图解的方式，介绍了下消息KafkaProducer是怎样发送消息的。

[kafka]: https://inetyoung.blog.csdn.net/article/details/111914662
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221120/ebc2cd4ff8a84af09cb71f3ec9dc03f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221120/060a56afabf646588ee9c5a68c790797.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221120/b90d64a84e814cd1b51e25316fa9d598.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221120/b0b6604b403d4f16a7eecb3a368a8ec4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: /images/20221120/46bb98eefa3e4bca89d91966eb53889a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: /images/20221120/8e3157a040f34140a750dbcf2872e8b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: /images/20221120/5ae301448bd64a9b8f764f1a9facfe33.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: /images/20221120/810601f158d846e0beda795586b80b16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]: /images/20221120/31d3d53fa52d495a9f19ac6f209ad081.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]: /images/20221120/d7566b82796a43ef918e9a70ce3e20b0.png