聊聊 Kafka：Producer 原始碼決議

一、前言

前面幾篇我們講了關于 Kafka 的基礎架構以及搭建，從這篇開始我們就來原始碼分析一波，我們這用的 Kafka 版本是 2.7.0，其 Client 端是由 Java 實作，Server 端是由 Scala 來實作的，在使用 Kafka 時，Client 是用戶最先接觸到的部分，因此，我們從 Client 端開始，會先從 Producer 端開始，今天我們就來對 Producer 原始碼決議一番，

二、Producer 使用

首先我們先通過一段代碼來展示 KafkaProducer 的使用方法，在下面的示例中，我們使用 KafkaProducer 實作向 Kafka 發送訊息的功能，在示例程式中，首先將 KafkaProduce 使用的配置寫入
到 Properties 中，每項配置的具體含義在注釋中進行解釋，之后以此 Properties 物件為引數構造 KafkaProducer 物件，最后通過 send 方法完成發送，代碼中包含同步發送、異步發送兩種情況，

從上面的代碼可以看出 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);
}

四、訊息發送程序

用戶是直接使用 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 攔截器代碼

4.1.2 攔截器核心邏輯

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() 的具體實作：

在 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 內部提供的序列化和反序列化演算法如下圖所示：

當然我們也是可以自定義序列化的具體實作，不過一般情況下，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;
}

上面這個默認演算法核心就是粘著磁區快取

5.4 向 RecordAccmulator 中追加 record 資料

我們講 RecordAccumulator 之前先看這張圖，這樣的話會對整個發送流程有個大局觀，

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，畫張圖來看下：

上面的代碼不知道大家有沒有疑問？分配記憶體的代碼為啥不在 synchronized 同步塊中分配？導致下面的 synchronized 同步塊中還要 tryAppend 一下，

因為這時候可能其他執行緒已經創建好 RecordBatch 了，造成多余的記憶體申請，

如果把分配記憶體放在 synchronized 同步塊會有什么問題？

記憶體申請不到執行緒會一直等待，如果放在同步塊中會造成一直不釋放 Deque 佇列的鎖，那其他執行緒將無法對 Deque 佇列進行執行緒安全的同步操作，

再跟下 tryAppend() 方法，這就比較簡單了，

以上代碼見圖解：

5.5 喚醒 sender 執行緒發送 RecordBatch

當 record 寫入成功后，如果發現 RecordBatch 已滿足發送的條件（通常是 queue 中有多個 batch，那么最先添加的那些 batch 肯定是可以發送了），那么就會喚醒 sender 執行緒，發送 RecordBatch，

sender 執行緒對 RecordBatch 的處理是在 run() 方法中進行的，該方法具體實作如下：

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

其中 pollTimeout 意思是最長阻塞到至少有一個通道在你注冊的事件就緒了，回傳 0 則表示走起發車了，

我們繼續跟下：org.apache.kafka.clients.producer.internals.RecordAccumulator#ready

最后再來看下里面這個方法 org.apache.kafka.clients.producer.internals.RecordAccumulator#drain，從accumulator 緩沖區獲取要發送的資料，最大一次性發 max.request.size 大小的資料，

六、總結

最后為了讓你對 Kafka Producer 有個宏觀的架構理解，請看下圖：

簡要說明：

• 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 的更新機制，敬請期待~