kafka消息生产者的两个线程详解

主线程：负责消息创建，拦截器，序列化器，分区器等操作，并将消息追加到消息收集器RecoderAccumulator中；
消息收集器RecoderAccumulator为每个分区都维护了一个Deque类型的双端队列。
ProducerBatch可以理解为是ProducerRecord集合，批量发送有利于提升吞吐量，降低网络影响；
由于生产者客户端使用 java.io.ByteBuffer在发送消息前进行消息保存，并维护了一个BufferPool实现ByteBuffer的复用；该缓存池只针对特定大小(batch.size指定)的ByteBuffer进行管理，对于消息过大的缓存，不能做到重复利用。
每次追加一条ProducerRecord消息，会寻找/新建对应的双端队列，从其尾部获取一个ProducerBatch，判断当前消息大小是否可以写入该批次中。若可以写入则写入；若不可写入，则新建一个ProducerBatch，判断该消息大小是否超过客户端参数配置 batch.size值，不超过，则以batch.size建立新的ProducerBatch，这样方便进行缓存重复利用；若超过，则以计算的消息大小建立对应的ProducerBatch ，缺点就是该内存不能被复用了。

Sender线程：
该线程从消息收集器获取缓存的消息，将其处理为<Node,List>的形式， Node表示集群的broker节点。
进一步将转化为<Node,Request>的形式，此时才可以向服务端发送数据。
在发送前，Sender线程将消息以Map<NodeId,Deque> 的形式保存到InFlightRequests中进行缓存，可以通过其获取 leastLoadedNode ,即当前Node中负载压力最小的一个，以实现消息的尽快发出。

如果主线程生产消息过快,会把缓冲区占满,就会阻塞,阻塞时间可以配置,如果这个时间过后这个消息还没有被发送出去,主线程将会抛异常

最后

以上就是务实草莓为你收集整理的kafka消息生产者的两个线程详解的全部内容，希望文章能够帮你解决kafka消息生产者的两个线程详解所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。