kafka生产者的缓存机制介绍

​

 ​

前言

今天咱们来看下kafka生产者客户端缓存架构的设计是怎样的？咱们先总体看下发送消息到java资料分服务端的完整过程和架构：

​

消息累加器

其实，整个生产者客户端是由两个线程协调运行的，一个是主线程Producer线程，一个是Sender线程。由主线程生产消息，而后缓存到消息累加器（RecordAccumulator）；而Sender线程则负责从消息累加器中不断获取消息，而后发送到kakfa broker。这个过程大体以下图：

​

 

那为何须要消息累加器呢？直接发送不是更直接？其实主要是用来缓存消息以便Sender线程能够每次批量发送，从而减小网络传输的资源消耗；而数据到服务端，服务端也能够批量写操做，从而减小磁盘I/O资源消耗，能够提高性能。这个设计，也能够运用于咱们平时的业务需求场景开发中。

 

那么消息累加器中的结构是怎样的呢？消息累加器的内部为每一个区分维护一个双端队列Deque，队列的内容是ProducerBatch；而ProducerBatch包含了一个或者多个ProducerRecord消息。

ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches;

具体结构以下图：

​

 

消息Record写入到消息累加器时，会被追加到双端队列的尾部。而尾部的ProducerBatch剩余可用空间也意味着可否再写入本次的这条消息ProducerRecord。ProducerBatch的大小和batch.size参数有着密切的关系。batch.size的参数大小控制着ProducerBatch的大小，默认是16KB大小。

 

​

 

整个过程是这样的：

 

1. 消息写入到消息累加器时，先寻找其对应分区的双端队列（若是不存在则新建队列）
2. 从双端队列的尾部获取一个ProducerBatch对象（若是不存在则新建Batch对象）
3. 判断ProducerBatch是否还有剩余空间写入本次的消息Record，若是能够则写入（若是不能够则须要建立一个新的ProducerBatch对象）
4. 新建ProducerBatch对象时会评估这条消息的大小是否超过batch.size参数的大小，若是不超过则以batch.size参数大小建立Batch对象（若是超过，就以消息的评估大小建立新Batch对象）。而用batch.size建立的内存空间是会被BufferPool管理的，能够进行空间复用；而超过batch.size建立的内存空间则不会被复用，使用完成则被回收。

而整个消息累加器的缓存空间与buffer.memory参数有关。默认是32MB大小。若是生产者客户端须要向不少区分发送比较多的消息，则能够根据实际状况将此参数适当调大以增长总体的吞吐量。

​

 

Sender线程

 

Sender线程则异步从消息累加器中获取缓存的消息，而后将其转为指定格式的ProducerRequest对象，将Request对象请求发往各个broker了。不过，请求在从Sender线程发往broker以前还会被保存到InFlightRequests中，其主要做用是缓存了已经发出去但尚未收到服务端响应的请求。

InFlightRequests相关的有一种重要的配置参数是max.in.flight.requests.per.connection。该参数表示每一个链接（客户端与broker node节点之间的网络链接）的最多缓存请求数，默认值为5个。当超过该数值以后，消费者客户端便不能再向这个链接发送更多的请求了。另外也得注意，当该参数配置大于1时，因为由于失败重试缘由，可能会存在消息乱序的风险。

​

 

总结

 

本次咱们了解了生产者客户端的两个线程，主线程Producer线程和Sender线程的各自的职责和做用；同时也了解了kafka的Producer客户端发送缓存机制。对于优秀的设计机制，咱们也能够思考借鉴运用于其余相似的业务开发中。

最后

最近我整理了整套《JAVA核心知识点总结》,说实话 ,做为一名Java程序员,不论你需不须要面试都应该好好看下这份资料。拿到手老是不亏的~个人很多粉丝也所以拿到腾讯字节快手等公司的Offer

进【Java进阶之路群】，找管理员获取哦-！

​

​

 

​