一、Kafka生产者 如何将消息发送到 kafka集群？

将下图纵向分为4列：

1）生产者的主线程

Producer对象：生成一个该对象，然后调用send方法

拦截器：不是必须的，可选

序列化器：kafka自己的更轻便，大部分都是数据，保证安全校验的只是小部分；而Java数据只占一小部分，大部分都是安全校验。因此在大数据的情况下，不能用Java自带的序列化功能。

分区器：这里的分区器，只是分recordAccumulator的区域，分到哪个DQueue

 

2）RecordAccumulator

该部分都在内存中，默认大小是32M

该部分有两个参数要注意 batch.size 和 linger.ms，具体含义见图。

这部分还有个内存池：

在生产者发送数据时，会从内存池拿内存用来储存数据；

待kafka集群成功接收到数据后，数据从DQueue删除释放内存，再归还给内存池

 

3）生产者的sender线程

sender线程会主动拉取数据。拉取数据的条件是batch.size 或 linger.ms，任何一个符合要求了就能拉取。

sender能容忍某个broker不返回应答的个数，是5。如果发了5个数据，都没有答复，那就不会再给该broker发送。

如果成功收到应答acks,就会清理掉DQueue里的数据；

如果发送不成功，会重试。该值默认设置的是Integer.MAX_VALUE

 

4）kafka集群

返回的应答种类，见图。

 

 

标签：sender,生产者,kafka,内存,DQueue,Kafka,数据
来源： https://www.cnblogs.com/frankcui/p/16065365.html