Kafka 源码分析—

Kafka 源码分析——Producer

前言

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

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

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

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

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

设置重试时间

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

设置序列化器

设置拦截器(interceptors)

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

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

上图的一些设置

设置缓冲区

设置消息累加器

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

初始化集群元数据(metadata)

创建Sender线程

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

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

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

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

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

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

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

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

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

核心就是这段代码：

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

申请内存的过程

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

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