聊聊 Kafka：Producer 源码解析

我就是我 2022-09-10 05:24 149阅读 0赞

## 一、前言 ##

前面几篇我们讲了关于 Kafka 的基础架构以及搭建，从这篇开始我们就来源码分析一波。我们这用的 Kafka 版本是 2.7.0，其 Client 端是由 Java 实现，Server 端是由 Scala 来实现的，在使用 Kafka 时，Client 是用户最先接触到的部分，因此，我们从 Client 端开始，会先从 Producer 端开始，今天我们就来对 Producer 源码解析一番。

## 二、Producer 使用 ##

首先我们先通过一段代码来展示 KafkaProducer 的使用方法。在下面的示例中，我们使用 KafkaProducer 实现向 Kafka 发送消息的功能。在示例程序中，首先将 KafkaProduce 使用的配置写入  
到 Properties 中，每项配置的具体含义在注释中进行解释。之后以此 Properties 对象为参数构造 KafkaProducer 对象，最后通过 send 方法完成发送，代码中包含同步发送、异步发送两种情况。

![在这里插入图片描述][355609a63e9c44dda922fdda3a4ef60d.png]  
从上面的代码可以看出 Kafka 为用户提供了非常简洁方便的 API，在使用时，只需要如下两步：

*  初始化 KafkaProducer 实例
 *  调用 send 接口发送数据

本文主要是围绕着初始化 KafkaProducer 实例与如何实现 send 接口发送数据而展开的。

## 三、KafkaProducer 实例化 ##

了解了 KafkaProducer 的基本使用，然后我们来深入了解下方法核心逻辑：

public KafkaProducer(Properties properties) { 
        this(Utils.propsToMap(properties), (Serializer)null, (Serializer)null, (ProducerMetadata)null, (KafkaClient)null, (ProducerInterceptors)null, Time.SYSTEM);
    }

![在这里插入图片描述][5d27a1573c264fd6ac142320360ffc40.png]

## 四、消息发送过程 ##

用户是直接使用 producer.send() 发送的数据，先看一下 send() 接口的实现

// 异步向一个 topic 发送数据
    public Future<RecordMetadata> send(ProducerRecord<K, V> record) { 
        return this.send(record, (Callback)null);
    }
    
    // 向 topic 异步地发送数据，当发送确认后唤起回调函数
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) { 
        ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
        return this.doSend(interceptedRecord, callback);
    }

数据发送的最终实现还是调用了 Producer 的 doSend() 接口。

**4.1 拦截器**

首先方法会先进入拦截器集合 ProducerInterceptors ， onSend 方法是遍历拦截器 onSend 方 法，拦截器的目的是将数据处理加工， Kafka 本身并没有给出默认的拦截器的实现。如果需要使用拦截器功能，必须自己实现接口。

**4.1.1 拦截器代码**

![在这里插入图片描述][ffadc4f4823c4105b2c3dec1ebf8f2d8.png]  
**4.1.2 拦截器核心逻辑**  
![在这里插入图片描述][620b94435c7e4268a1e115f6d9dec433.png]  
ProducerInterceptor 接口包括三个方法：

*  `onSend(ProducerRecord<K, V> var1)`：该方法封装进 KafkaProducer.send 方法中，即它运行在用户主线程中的。 确保在消息被序列化以计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的 topic 和分区，否则会影响目标分区的计算。
 *  `onAcknowledgement(RecordMetadata var1, Exception var2)`：该方法会在消息被应答之前或消息发送失败时调用，并且通常都是在 producer 回调逻辑触发之前。onAcknowledgement 运行在 producer 的 IO 线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢 producer 的消息发送效率。
 *  `close()`：关闭 interceptor，主要用于执行一些资源清理工作。

拦截器可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个 interceptor，则 producer 将按照指定顺序调用它们，并仅仅是捕获每个 interceptor 可能抛出的异常记录到错误日志中而非在向上传递。

**4.2 Producer 的 doSend 实现**

下面是 doSend() 的具体实现：

![在这里插入图片描述][658f41570d7944d18d2bc93bdb744d55.png]  
在 doSend() 方法的实现上，一条 Record 数据的发送，主要分为以下五步：

*  确认数据要发送到的 topic 的 metadata 是可用的（如果该 partition 的 leader 存在则是可用的，如果开启权限时，client 有相应的权限），如果没有 topic 的 metadata 信息，就需要获取相应的 metadata；
 *  序列化 record 的 key 和 value；
 *  获取该 record 要发送到的 partition（可以指定，也可以根据算法计算）；
 *  向 accumulator 中追加 record 数据，数据会先进行缓存；
 *  如果追加完数据后，对应的 RecordBatch 已经达到了 batch.size 的大小（或者 batch 的剩余空间不足以添加下一条 Record），则唤醒 sender 线程发送数据。

数据的发送过程，可以简单总结为以上五点，下面会这几部分的具体实现进行详细分析。

## 五、消息发送过程 ##

**5.1 获取 topic 的 metadata 信息**

Producer 通过 waitOnMetadata() 方法来获取对应 topic 的 metadata 信息，这块内容我下一篇再来讲。

**5.2 key 和 value 的序列化**

Producer 端对 record 的 key 和 value 值进行序列化操作，在 Consumer 端再进行相应的反序列化，Kafka 内部提供的序列化和反序列化算法如下图所示：  
![在这里插入图片描述][cd642f58f821453d948f4ba5d14d2c88.png]  
当然我们也是可以自定义序列化的具体实现，不过一般情况下，Kafka 内部提供的这些方法已经足够使用。

**5.3 获取该 record 要发送到的 partition**

获取 partition 值，具体分为下面三种情况：

*  指明 partition 的情况下，直接将指明的值直接作为 partiton 值；
 *  没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值；
 *  既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的 round-robin 算法。

具体实现如下：

// 当 record 中有 partition 值时，直接返回，没有的情况下调用 partitioner 的类的 partition 方法去计算（KafkaProducer.class）
    private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) { 
        Integer partition = record.partition();
        return partition != null ? partition : this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
    }

Producer 默认使用的 partitioner 是 org.apache.kafka.clients.producer.internals.DefaultPartitioner，用户也可以自定义 partition 的策略，下面是默认分区策略具体实现：

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
        return this.partition(topic, key, keyBytes, value, valueBytes, cluster, cluster.partitionsForTopic(topic).size());
    }
    
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster, int numPartitions) { 
        return keyBytes == null ? this.stickyPartitionCache.partition(topic, cluster) : Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

![在这里插入图片描述][ba6855ed0be145c4a8bc072ead827b26.png]  
上面这个默认算法核心就是粘着分区缓存

**5.4 向 RecordAccmulator 中追加 record 数据**

我们讲 RecordAccumulator 之前先看这张图，这样的话会对整个发送流程有个大局观。

![在这里插入图片描述][379f39fc98b3426daa82aaf2f01a3e47.png]  
RecordAccmulator 承担了缓冲区的角色。默认是 32 MB。

在 Kafka Producer 中，消息不是一条一条发给 broker 的，而是多条消息组成一个 ProducerBatch，然后由 Sender 一次性发出去，这里的 batch.size 并不是消息的条数（凑满多少条即发送），而是一个大小。默认是 16 KB，可以根据具体情况来进行优化。

在 RecordAccumulator 中，最核心的参数就是：

private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;

它是一个 ConcurrentMap，key 是 TopicPartition 类，代表一个 topic 的一个 partition。value 是一个包含 ProducerBatch 的双端队列。等待 Sender 线程发送给 broker。画张图来看下：  
![在这里插入图片描述][1373f217babb45f3b1bd9f2235417bf5.png]

![在这里插入图片描述][7335f80be8004203ac180b94566cb4b0.png]  
上面的代码不知道大家有没有疑问？分配内存的代码为啥不在 synchronized 同步块中分配？导致下面的 synchronized 同步块中还要 tryAppend 一下。

因为这时候可能其他线程已经创建好 RecordBatch 了，造成多余的内存申请。

如果把分配内存放在 synchronized 同步块会有什么问题？

内存申请不到线程会一直等待，如果放在同步块中会造成一直不释放 Deque 队列的锁，那其他线程将无法对 Deque 队列进行线程安全的同步操作。

再跟下 tryAppend() 方法，这就比较简单了。

![在这里插入图片描述][eb17cb4626d54910a8c2c485c95bac26.png]  
以上代码见图解：

![在这里插入图片描述][5e32bae5f9b64f5d99dc5f860c0c864f.png]  
**5.5 唤醒 sender 线程发送 RecordBatch**

当 record 写入成功后，如果发现 RecordBatch 已满足发送的条件（通常是 queue 中有多个 batch，那么最先添加的那些 batch 肯定是可以发送了），那么就会唤醒 sender 线程，发送 RecordBatch。

sender 线程对 RecordBatch 的处理是在 run() 方法中进行的，该方法具体实现如下：  
![在这里插入图片描述][fd0c1a75b6f949709838fc2ac5224428.png]  
![在这里插入图片描述][045c5e158c20477b96991e8c91318c26.png]

其中比较核心的方法是 run() 方法中的 org.apache.kafka.clients.producer.internals.Sender\#sendProducerData

其中 pollTimeout 意思是最长阻塞到至少有一个通道在你注册的事件就绪了。返回 0 则表示走起发车了。

![在这里插入图片描述][d67cb7e1a61e4cb7b617ace006ae88ae.png]  
我们继续跟下：org.apache.kafka.clients.producer.internals.RecordAccumulator\#ready  
![在这里插入图片描述][d98add806edb40d3a54042beceda7782.png]  
最后再来看下里面这个方法 org.apache.kafka.clients.producer.internals.RecordAccumulator\#drain，从accumulator 缓冲区获取要发送的数据，最大一次性发 max.request.size 大小的数据。

![在这里插入图片描述][6d5434a4ad6c4ff386a74c73f7f438b0.png]  
![在这里插入图片描述][53f7995ecb0f40ddbc067bd4e9525af5.png]

## 六、总结 ##

最后为了让你对 Kafka Producer 有个宏观的架构理解，请看下图：

![在这里插入图片描述][dca7c48af8e946d989ce241b3545ce1e.png]  
简要说明：

*  new KafkaProducer() 后创建一个后台线程 KafkaThread (实际运行线程是 Sender，KafkaThread 是对 Sender 的封装) 扫描 RecordAccumulator 中是否有消息。
 *  调用 KafkaProducer.send() 发送消息，实际是将消息保存到 RecordAccumulator 中，实际上就是保存到一个 Map 中 (ConcurrentMap<TopicPartition, Deque>)，这条消息会被记录到同一个记录批次 (相同主题相同分区算同一个批次) 里面，这个批次的所有消息会被发送到相同的主题和分区上。
 *  后台的独立线程扫描到 RecordAccumulator 中有消息后，会将消息发送到 Kafka 集群中 (不是一有消息就发送，而是要看消息是否 ready)
 *  如果发送成功 (消息成功写入 Kafka), 就返回一个 RecordMetaData 对象，它包括了主题和分区信息，以及记录在分区里的偏移量。
 *  如果写入失败，就会返回一个错误，生产者在收到错误之后会尝试重新发送消息 (如果允许的话，此时会将消息在保存到 RecordAccumulator 中)，几次之后如果还是失败就返回错误消息。

好了，本文对 Kafka Producer 源码进行了解析，下一篇文章将会详细介绍 metadata 的内容以及在 Producer 端 metadata 的更新机制。敬请期待~

[355609a63e9c44dda922fdda3a4ef60d.png]: /images/20220829/caa1909395f040ce9acd5fcf28e00710.png
[5d27a1573c264fd6ac142320360ffc40.png]: /images/20220829/2a2a0f8a6bce4d6b820ad99bcd033f7a.png
[ffadc4f4823c4105b2c3dec1ebf8f2d8.png]: /images/20220829/93eb2d364cfb4cb3981bf0bccfd9dc52.png
[620b94435c7e4268a1e115f6d9dec433.png]: /images/20220829/430c627d8ea04f6fb1604914a5fcf503.png
[658f41570d7944d18d2bc93bdb744d55.png]: /images/20220829/8ef1c5b5caf44f8a972773736cc78aae.png
[cd642f58f821453d948f4ba5d14d2c88.png]: /images/20220829/2769057fb9854a8f8ce5e976527644d4.png
[ba6855ed0be145c4a8bc072ead827b26.png]: /images/20220829/79da8bd9c39149e0835028799361bf36.png
[379f39fc98b3426daa82aaf2f01a3e47.png]: /images/20220829/d1a9d357e0b94a8a81a6cfbbf43be409.png
[1373f217babb45f3b1bd9f2235417bf5.png]: /images/20220829/9b996a5b81624e049c4915ba5a2b3e49.png
[7335f80be8004203ac180b94566cb4b0.png]: /images/20220829/98e4dc1a9c714e07903ad5c62aef7b13.png
[eb17cb4626d54910a8c2c485c95bac26.png]: /images/20220829/f318a9cf705f4833ae88d9daffaea4b5.png
[5e32bae5f9b64f5d99dc5f860c0c864f.png]: /images/20220829/02f8c37d79a44510bb201f141c19942b.png
[fd0c1a75b6f949709838fc2ac5224428.png]: /images/20220829/54e6e1c354024152ad5b2ce39c8cc078.png
[045c5e158c20477b96991e8c91318c26.png]: /images/20220829/ee409a7575c0489482a4a506a1948f40.png
[d67cb7e1a61e4cb7b617ace006ae88ae.png]: /images/20220829/262213dee386451eb109155c462362e6.png
[d98add806edb40d3a54042beceda7782.png]: /images/20220829/4dbe676537dc4a28bb822238f3cfe213.png
[6d5434a4ad6c4ff386a74c73f7f438b0.png]: /images/20220829/4df95500052046589100da0ff2945e1e.png
[53f7995ecb0f40ddbc067bd4e9525af5.png]: /images/20220829/b60709a1285e4f799ca7d2cb51957103.png
[dca7c48af8e946d989ce241b3545ce1e.png]: /images/20220829/b02d74f401264c9cab7255de22f383b5.png