最新新闻事件50字_建筑工程网格优化_有没有免费推广平台_百度seo是啥意思

---

前言

相信现在的javer对于Kafka应该都很熟悉了，不管是八股文还是工作中使用。虽然Kafka server是scala写的，但是client是java写的，所以咱们理解client的代码还是比较容易的，今天先来基于源码解读下Kafka Producer的主体流程

---

一、Kafka Producer是什么？

这个应该不用过多介绍了吧，我们发送消息到MQ就是通过Kafka Producer来实现的。

二、主要组件

1.Kafka Producer

1.1 partitioner

partitioner的作用是用来对发送的消息进行partition选择的，譬如msg要发送某个topic，topic里有多个partition，需要选择发往哪个partition。
如果我们发送消息时设置了partition key，那么就按照partition key进行hash，然后选择发送的partition
如果没有设置partition key，那么就默认使用轮询的方式来选择partition

1.2 keySerializer

顾名思义，就是对key进行序列化的工具类。因为我们的数据需要发送到网络里，所以数据必须序列化成二进制，所以需要进行序列化

1.3 valueSerializer

同上，和keySerializer类似的作用

1.4 accumulator

accumulator是比较重要的一个组件，主要功能包括：

1. 选择/创建需要append数据的batch
2. 把msg添加到该batch里面，获得batch里produceFuture

1.5 sender

sender是真正执行消息发送的组件，主要功能包括：

1. 把可以发送的batch，组装成clientRequest，给底层的nioSelector注册写事件
2. nioSelector处理读写事件，写事件处理时，将消息发送出去，更新batch的produceFuture让accumulator感知到消息发送结果
3. 以下详细讲讲Sender的流程

2.Sender

2.1 acks

acks是用来描述msg发送要怎样才能确认发送成功。
“-1”：全部副本应答，默认值
“0”：不需要任何应答，这种情况发送完立马认为发送成功
“1”：leader副本应该即认为发送成功

2.2 client

接口是KafkaClient，实现类是NetWorkClient

1. send：

记录当前发送中的request=》inFlightRequests，
调用Selector进行消息发送

2. poll：

调用Selector进行写事件处理：发送消息
清理一些数据：处理inFlightRequests
执行回调：handleProduceResponse：producerBatch.complete=》会更新batch里的produceFuture

2.3 inFlightBatches

记录当前当前正在被发送的batch

Map<TopicPartition, List<ProducerBatch>> inFlightBatches;

可以看到该Map的value是list类型，保证了同一个partition中的batch数据有序性

3. Selector

其实Selector的作用上面已经描述了，就是基于partition的node找到对应的channel，执行写事件注册和真实的消息发送

3.1 nioSelector

这个应该不用详细讲了，学过NIO的同学都知道，java的NIO Selector可以进行读写事件处理，就是通过selector的select方法，找到可处理的keys，然后基于不同的keys，拿到对应的channel，往channel写数据或者从channel读数据

3.2 channels

该channel是Kafka对java的channel进行的封装，得到的KafkaChannel，其实能够实现的功能就是暂存下可发送消息，以及调用java的channel 发送数据。此处是保存partition node和channel的映射关系。

4. 全局总览

5. 一点思考

为什么Producer要分为accumulator和sender这两个重要组件呢？
为啥不像我们写业务代码一样，把消息append到batch里面之后，就直接发送消息？
答案是：功能定位不同，面对的处理对象也不同，所以解耦最合适。
accumulator的功能是追加消息；Sender的功能是网络发送消息；
追加消息，面对的处理对象是消息；而网络发送消息，面对的处理对象是partition node，也就是创建网络连接的节点。
为了增加消息发送吞吐量，中间引入了batch。
所以把「追加消息」和「网络发送消息」分开让两个不同组件来处理，在功能的实现上进行解耦~

---

总结

Kafka的消息发送其实就是一个RPC的过程，有自己的网络协议、消息协议、消息序列化方式、数据批量发送（增加吞吐量）、超时处理、底层网络模型。
不过Producer对于batch的复用，其实还是有一定的厉害之处的，等下次有空分析下~