Kafka 源码分析——Producer

电玩女神 2023-10-16 11:09 2阅读 0赞

#### 文章目录 ####

*  前言
 *  Producer 整体流程
 *  Producer 初始化
 *  Producer 发送流程
 *   *  执行拦截器逻辑
     *  获取集群元数据
     *  序列化
     *  选择分区
     *  消息累加进缓存
     *  消息发送
 *  Producer缓冲区
 *  Producer 参数调优

## 前言 ##

在 Kafka 中, 把产生消息的一方称为 Producer 即 生产者，它是 Kafka 的核心组件之一， 也是消息的来源所在。它的主要功能是将客户端的请求打包封装发送到 kafka 集群的某个 Topic 的某个分区上。那么这些生产者产生的消息是怎么传到 Kafka 服务端的呢？

## Producer 整体流程 ##

Kafka一条消息发送和消费的流程

![image.png][]

站在源码的核心角度，可以把Producer分成以下几个核心部分：

1.  Producer初始化
2.  Producer发送流程
3.  Producer缓冲区
4.  Producer参数与调优

## Producer 初始化 ##

![image.png][image.png 1]

因为源码中有非常多的一些额外处理，所以解读源码没必要每行都读，只需要根据梳理的主流程找到核心代码进行解读就可以。

**设置分区器(partitioner)**，分区器是支持自定义的

![image.png][image.png 2]

**设置重试时间**

重试时间(retryBackoffMs)默认100ms。如果发送消息到broker时抛出异常，且是允许重试的异常，那么就会最大重试retries参数指定的次数，同时retryBackoffMs是重试的间隔。

![image.png][image.png 3]

**设置序列化器**

![image.png][image.png 4]

**设置拦截器(interceptors)**

![image.png][image.png 5]

拦截器一般用得不多，可以为消息统一添加字段或统计发送失败成功次数，这些逻辑会拖慢producer的消息发送效率，不推荐生产中使用。

想要实现拦截器，我们需要先实现ProducerInterceptor接口即可，然后在生产者中设置进去即可。

![image.png][image.png 6]

![image.png][image.png 7]

![image.png][image.png 8]  
上图的一些设置

1.  设置最大的消息为多大(maxRequestSize), 默认最大1M, 生产环境可以提高到10M
2.  设置缓存大小(totalMemorySize) 默认是32M
3.  设置压缩格式(compressionType)
4.  初始化RecordAccumulator也就是缓冲区指定为32M

**设置缓冲区**

![image.png][image.png 9]

**设置消息累加器**

因为生产者是通过缓冲的方式发送，所以需要一个消息累加器配合才能完成消息的发送。

![image.png][image.png 10]

**初始化集群元数据(metadata)**

![image.png][image.png 11]

**创建Sender线程**

![image.png][image.png 12]

**这里还初始化了一个重要的管理网路的组件 NetworkClient**

![image.png][image.png 13]

KafkaThread将Sender设置为守护线程并启动

![image.png][image.png 14]

## Producer 发送流程 ##

### 执行拦截器逻辑 ###

执行拦截器逻辑，预处理消息，封装 Producer Record

![image.png][image.png 15]

### 获取集群元数据 ###

从 Kafka Broker 集群获取集群元数据metadata

![image.png][image.png 16]

### 序列化 ###

调用Serializer.serialize()方法进行消息的key/value序列化

![image.png][image.png 17]

### 选择分区 ###

调用partition()选择合适的分区策略，给消息体 Producer Record 分配要发送的 topic 分区号

![image.png][image.png 18]

### 消息累加进缓存 ###

将消息缓存到RecordAccumulator 收集器中, 最后判断是否要发送。

![image.png][image.png 19]

### 消息发送 ###

真正的消息发送是Sender线程来做，并且还要结合缓冲区来处理。这里我们只需要知道发送的条件：缓冲区数据大小达到 batch.size 或者 linger.ms 达到上限。

## Producer缓冲区 ##

Kafka生产者的缓冲区，也就是内存池，可以将其类比为连接池(DB, Redis)，主要是避免不必要的创建连接的开销,。这样内存池可以对 RecordBatch 做到反复利用，防止引起Full GC问题。

核心就是这段代码：

![image.png][image.png 20]

![image.png][image.png 21]

Kafka 内存设计有两部分，可用的内存（未分配的内存，初始的时候是 32M）和已经被分配了的内存，每个小 Batch 是 16K，然后这一个个的 Batch 就可以被反复利用，不需要每次都申请内存, 两部分加起来是 32M。

**申请内存的过程**

发送流程中会把消息放入 accumulator中，即调用 accumulator.append() 追加, 然后把消息封装成一个个Batch 进行发送，然后去申请内存(free.allocate())。

![image.png][image.png 22]

1.  如果申请的内存大小超过了整个缓存池的大小，则抛异常出来。
2.  如果申请的大小是每个 recordBatch 的大小（16K），并且已分配内存不为空，则直接取出来一个返回。
3.  如果整个内存池大小比要申请的内存大小大 （this.availableMemory + freeListSize >= size），则直接从可用内存申请一块内存。

## Producer 参数调优 ##

在 Kafka 实际使用中，Producer 端既要保证吞吐量，又要确保无消息丢失，一些核心参数的配置就显得至关重要。

**acks**

参数说明：对于 Kafka Producer 来说是一个非常重要的参数，它表示指定分区中成功写入消息的副本数量，是 Kafka 生产端消息的持久性的保证。

**max.request.size**

参数说明：这个参数对于 Kafka Producer 也比较重要， **表示生产端能够发送的最大消息大小，默认值为1048576(1M)** 。

调优建议：这个配置对于生产环境来说有点小， **为了避免因消息过大导致发送失败，生产环境建议适当调大，比如可以调到10485760(10M)** 。

**retries**

参数说明：表示生产端消息发送失败时的重试次数，默认值为0，即不重试。 **这个参数一般是为了解决因系统瞬时故障导致的消息发送失败，比如网络抖动、Leader 选举及重选举，其中瞬时的 Leader 重选举是比较常见的。因此这个参数的设置对于 Kafka Producer 就显得非常重要** 。

调优建议：这里建议设置为一个大于0的值，比如3次。

**retry.backoff.ms**

参数说明：\*\*设定两次重试之间的时间间隔，避免无效的频繁重试，默认值为100, \*\*主要跟 retries 配合使用， 在配置 retries 和 retry.backoff.ms 之前，最好先估算一下可能的异常恢复时间，需要设定总的重试时间要大于异常恢复时间，避免生产者过早的放弃重试。

**connections.max.idele.ms**

参数说明：主要用来判断多久之后关闭空闲的链接，默认值540000（ms）即9分钟。

**compression.type**

参数说明： **该参数表示生产端是否要对消息进行压缩，默认值为不压缩(none)。** 压缩可以显著减少网络IO传输、磁盘IO以及磁盘空间，从而提升整体吞吐量，但也是以牺牲CPU开销为代价的。

调优建议：出于提升吞吐量的考虑，建议在生产端对消息进行压缩。**对于Kafka来说，综合考虑吞吐量与压缩比，建议选择lz4压缩。如果追求最高的压缩比则推荐zstd压缩。**

**buffer.memory**

参数说明： **该参数表示生产端消息缓冲池或缓冲区的大小，默认值为即33554432(32M)** 。这个参数基本可以认为是 Producer 程序所使用的内存大小。

调优建议：通常我们应尽量保证生产端整体吞吐量，建议适当调大该参数，也意味着生产客户端会占用更多的内存。

**batch.size**

参数说明： **该参数表示发送到缓冲区中的消息会被封装成一个一个的Batch，分批次的发送到 Broker 端，默认值为16KB。** 因此减小 batch 大小有利于降低消息延时，增加 batch 大小有利于提升吞吐量。

调优建议：通常合理调大该参数值，能够显著提升生产端吞吐量，比如可以调整到32KB，调大也意味着消息会有相对较大的延时。

**linger.ms**

参数说明： **该参数表示用来控制 Batch 最大的空闲时间，超过该时间的 Batch 也会自动被发送到 Broker 端。** 实际情况中, 这是吞吐量与延时之间的权衡。默认值为0，表示消息需要被立即发送，无需关系 batch 是否被填满。

调优建议：通常为了减少请求次数、提升整体吞吐量，建议设置一个大于0的值，比如设置为100，此时会在负载低的情况下带来100ms的延时。

[image.png]: https://img-blog.csdnimg.cn/img_convert/219d8b6131b02119e792d2446d858b76.png
[image.png 1]: https://img-blog.csdnimg.cn/img_convert/4298e5f6a13ee0f4a05e763017a2b9e5.png
[image.png 2]: https://img-blog.csdnimg.cn/img_convert/a67359b2015e3e1e09c7090da80c4b17.png
[image.png 3]: https://img-blog.csdnimg.cn/img_convert/62483ccb244ae0f2c8d8cbbdafff7e17.png
[image.png 4]: https://img-blog.csdnimg.cn/img_convert/0034388664e92e310e51c74f9655d18e.png
[image.png 5]: https://img-blog.csdnimg.cn/img_convert/d7b28e6c3a019551000bb32183aff417.png
[image.png 6]: https://img-blog.csdnimg.cn/img_convert/ff718ef1995725c98a58d3e764db0051.png
[image.png 7]: https://img-blog.csdnimg.cn/img_convert/3570535e37334a6400dbb223c5f38579.png
[image.png 8]: https://img-blog.csdnimg.cn/img_convert/fda122db698c08a2357bfcd10f6b6d55.png
[image.png 9]: https://img-blog.csdnimg.cn/img_convert/6029bc29e97695e03e0d52fbb1028fc0.png
[image.png 10]: https://img-blog.csdnimg.cn/img_convert/b114ed7909e694a6c17cc8a07118608c.png
[image.png 11]: https://img-blog.csdnimg.cn/img_convert/ede45cf585264f52f54a96eac822d35c.png
[image.png 12]: https://img-blog.csdnimg.cn/img_convert/bad211e56fcef5fb5df19f5a1085cced.png
[image.png 13]: https://img-blog.csdnimg.cn/img_convert/58e9d83907641a69ba8c3eec2895aad5.png
[image.png 14]: https://img-blog.csdnimg.cn/img_convert/29652de6eb8184b2cee13e6999d4e81b.png
[image.png 15]: https://img-blog.csdnimg.cn/img_convert/8b07cb29c614f60469563b58a4a7387f.png
[image.png 16]: https://img-blog.csdnimg.cn/img_convert/075a1ea2ead3ca58f542d15b8c2708c4.png
[image.png 17]: https://img-blog.csdnimg.cn/img_convert/5abcd8e313da7d2239a4d18f7fad3485.png
[image.png 18]: https://img-blog.csdnimg.cn/img_convert/fa09b012400d2326f77172458fff73bf.png
[image.png 19]: https://img-blog.csdnimg.cn/img_convert/4666f3572907a08c01541eb9ab465e4f.png
[image.png 20]: https://img-blog.csdnimg.cn/img_convert/6db0a25e3e890486fd2e56a23f2b716c.png
[image.png 21]: https://img-blog.csdnimg.cn/img_convert/c0f90e98efd269b8dbe1960051c74ce9.png
[image.png 22]: https://img-blog.csdnimg.cn/img_convert/a5ae0cd47ade23da5d081fad5196c79a.png