深度解析kafka生产者启动流程

ゝ一世哀愁。 2022-12-31 10:30 224阅读 0赞

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

### 文章目录 ###

*   *   *  前言
         *  1.生产者demo介绍
         *  2.启动流程分析
         *  总结

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

### 前言 ###

从本文开始，我们将开启kafka源码解析的征程，都知道kafka在消息中间件中以超高吞吐量闻名，我们会在接下来的kafka源码解析文章中慢慢揭秘kafka超高吞吐背后的技术点，包括顺序追加写，reactor网络模型，零拷贝等等。 本文主要从kafka生产者demo开始，然后深度解析下kafka生产者启动流程，看看生产者初始化的时候都启动了哪些组件，然后为生产者发送消息打下基础。

### 1.生产者demo介绍 ###

我这里直接贴上一段kafka 生产者发送消息demo 代码

final Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("client.id", "DemoProducer");
    properties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    // 主题
    String  topic = "xxxx";
    // final Metric m = new Metric(60*1000, 60);
    KafkaProducer producer = new KafkaProducer(properties);
    int i =0;
    while (true) { 
        i++;
        ProducerRecord producerRecord = new ProducerRecord(topic, i, " produce record，" + i);
        producer.send(producerRecord, new Callback() { 
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) { 
                System.out.println(metadata);
            }
        });
    }

我们主要是看下前半部分，首先是创建一个properties ，里面就是定义一堆配置，`bootstrap.servers` 是kafka broker 地址，你多个的话就用英文逗号隔开就行了，`client.id` 客户端id ，你不指定也行，kafka会自己给你生成，`key.serializer` key序列化器，就是序列化key的， `value.serializer`序列化value的。  
接着就是创建KafkaProducer 对象了，然后将这个properties 传进去。  
接着就是构造消息ProducerRecord 对象，需要指定topic ， key ，value ， topic 这个不用说，就是你可以理解为一类消息，key这个可以用来选择partition，如果你消息里面没有指定发送到哪个partition上的话，就会根据你这key 来选择partition，如果key是null的话，就使用轮询的方式选择partition ， key不是null的话，就hash模partition数量，这样你相同的key 就会发送到同一个partition上去了。 value就是你具体要发送的消息了  
最后就是调用producer的send 方法来发送消息了，这里采取的异步方式，需要传进去异步回调对象。

### 2.启动流程分析 ###

我们这里就分析一下`KafkaProducer producer = new KafkaProducer(properties);`这行创建KafkaProducer 对象的代码，看看它都干了啥。  
![在这里插入图片描述][20201230233459670.png]  
创建一个ProducerConfig 配置对象，将我们properties 作为构造参数传了进去，关于配置类我们这里就不看了，其实就是解析咱们这个配置啥的。直接进去KafkaProducer 的重载方法中去。  
由于这个构造方法贼长，我们分段来看看。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
这里重要一点的就是获取了一下clientId ，如果没有配置的话，就是用它默认的这个，这里我们配置了，下面这堆都是统计的东西，我们略过，接着看下一块代码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
这里其实就是获取配置项， partitioner 是选择这个partition用的，这里重要的是创建了个Metadata 对象，这个是用来管理一些元数据的，包括cluster 一些东西，接着往下看吧  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
先是看看你配置这个超时时间了没，如果你配置了，就使用你配置的这个，如果你没有配置就使用它默认的一个配置是40\*1000。 下面accumulator 这个东西是个非常重要的一个组件，它是个累加器（缓冲区），意思就是它会将你要发送的消息给你在这里面攒攒，攒成一批数据。  
再接着往下看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
这里先是验证一下这些broker 地址能不能用，创建一个Cluster 对象，里面其实就是维护了几个node ，这个时候是没有topic，partition信息的，关于每个node 里面partition，topic信息会在发送消息的时候更新上， matedata 这个对象会更新一下cluster，将cluster 维护进去。  
然后就是创建了一个ChannelBuilder，一看到这个builder 其实就能想到建造者模式，channel就是由它构造出来的，最后一个重要的组件，NetworkClient，里面还有一个Selector ，一看到这个Selector你就应该想到jdk nio 里面的selector组件，没错，里面封装了这个jdk的selector。这个NetworkClient一看就是网络通信的，很重要。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
这个sender 组件也是很重要的，可以看到后面创建了一个线程，将sender 作为任务传进去了，并启动了线程，这个sender 其实就是个Runnable实现类，这个也是个非常重要的组件，其实到这一步，kafka生产者重要的组件都已经出来了，在总结的时候我们会介绍下这些重要组件的关系。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
这个就是获取配置里面的key序列化组件与value的序列化组件，没啥好说的这个。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
这个就是从配置中获取拦截器，根据你配置的这堆拦截器生成一个生产者拦截器对象，可以理解为拦截器链 ，你如果不配置的话，默认是没有的，这里我们也就不过多的关注了。

### 总结 ###

好了，到这我们KafkaProducer 对象初始化流程就结束了，这里我们并没KafkaProducer 下面组件的初始化源码做剖析，所以我们这里细数一下它重要的组件作用

1.  metadata 一看就知道存放原数据的，然后它里面维护了一个cluster，这cluster里面维护了node节点 ， topic，partition的元数据信息。
2.  partitioner 这个组件就是用来选择消息往哪个partition发送的，比如说你一个topic 有多个partition ，然后partitioner就会帮你选择消息发送到哪个partition上去。
3.  accumulator 这个累加器（缓冲区），它的作用就是按照topic + partition 给消息分组，每个（topic + partition ）下面有一个队列，分过来的消息就追加到后面的batch里面去，batch 你可以理解为一堆消息，发送消息的时候也是一批一批发送的。
4.  NetworkClient 一看就知道是网络客户端的一个组件，它里面有个Selector 组件是非常重要的，这个Selector 封装了jdk的selector
5.  sender 其实发送消息的线程将消息追加到 accumulator 的队列之后，就会返回了，继续去处理其他消息的发送了，剩下的具体发送工作都是这个sender 线程来做的，它控制着Selector 建立连接，接收响应，发送消息各种事件的处理，包括向broker 获取topic partition元数据信息，发送消息，接收broker的响应等等。

[20201230233459670.png]: /images/20221120/eb6340307eb2486bbcde5858ab7227d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20201230233853192.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 1]: https://img-blog.csdnimg.cn/20201230234038868.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 2]: https://img-blog.csdnimg.cn/20201230234400472.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 3]: https://img-blog.csdnimg.cn/20201230234935450.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/20201230235707614.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/20201230235923614.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/20201231000217440.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70