生產者客戶端由兩個執行緒協調執行。其中主執行緒建立訊息，並經過攔截器、序列化器、分割槽器作用後快取到訊息累加器; 訊息累加器中的ProducerBatch是一個雙端佇列，訊息新增時從尾部進入，Sender讀取訊息時從頭部取出。ProducerBatch包含鏈多個ProducerRecord，這樣使位元組的使用更加緊湊，同時傳送時減少了網路請求的次數以提升整體的吞吐量。

RecordAccumulator快取的大小可以通過生產者客戶端引數Buffer.memory配置，預設值為32MB。如果生產者傳送訊息的速度超過傳送到伺服器的速度，則會導致生產者空間不足，這個時候KafkaProducer的send方法要麼被阻塞，要麼丟擲異常，這個取決於max.block.ms的配置，預設為60000,即60秒。如果生產者客戶端要向很多分割槽傳送訊息，則可以適當增加buffer.memory，以提高整體的吞吐量。

在Sender執行緒將請求傳送kafka之前，還會將請求儲存到InFlightRequests中，其形式是Map<NodeId, Deque<Request>>, NodeId表示節點的id，這個map它的主要作用是快取了已經發出去但還沒有收到響應的請求，InFlightRequests提供配置引數來限制每個連線的最大快取數，超過該值後就不能再向該連線傳送請求，除非快取的請求收到了響應。通過比較Deque<Request>的size與這個引數的大小進行比較來判斷這個連線是否已經堆積了很多未響應的訊息。果真如此的話，說明這個Node節點負載較大或者網路連線有問題，再繼續向其傳送請求會增大請求超時的可能。

leastLoadedNode是所有Node中負載最小的那一個，選擇這個Node傳送請求可以使請求儘快發出，避免因網路擁塞等異常而影響傳送的進度。通常用於元資料請求、消費者組播協議的互動。

在向一個主題傳送訊息時，KafkaProducer要將此訊息追加到指定主題的某個分割槽所對應的leader副本之前，首先需要知道分割槽的數量，然後經過計算得出目標分割槽，還要知道目標分割槽的leader副本所在的broker節點的地址、埠資訊之後才能建立連線，最終才可以把訊息傳送到kafka。

元資料是指Kafka叢集的元資料，這些元資料記錄鏈叢集的有什麼主題、主題有哪些分割槽、每個分割槽的leader副本分配在那個節點，follower副本分配在哪些節點上，哪些副本在AR 、ISR等集合中，叢集中有哪些節點，控制器節點又是哪一個等等資訊。

當客戶端中沒有需要使用的元資料資訊時，比如沒有指定的主題資訊，或者超過metadata.max.age.ms 時間沒有更新元資料都會引起元資料的更新操作。客戶端引數metadata.max.age.ms的預設值為300000，即5分鐘。元資料的更新由Sender執行緒向leastLoadedNode傳送MetadataRequest請求。

                                                    生產者客戶端架構

生產者配置引數：

1.acks

acks=1。預設值即為1。生產者傳送訊息之後，只要分割槽的leader副本成功寫入訊息，那麼它就會收到來自服務端的成功響應。如果訊息無法寫入leader副本，比如在leader 副本崩潰、重新選舉新的 leader 副本的過程中，那麼生產者就會收到一個錯誤的響應，為了避免訊息丟失，生產者可以選擇重發訊息。

acks=0。生產者傳送訊息之後不需要等待任何服務端的響應。acks 設定為 0 可以達到最大的吞吐量。

acks=-1或acks=all。生產者在訊息傳送之後，需要等待ISR中的所有副本都成功寫入訊息之後才能夠收到來自服務端的成功響應。在其他配置環境相同的情況下，acks 設定為-1（all）可以達到最強的可靠性。

2.max.request.size

這個引數用來限制生產者客戶端能傳送的訊息的最大值，預設值為 1048576B，即1MB。

3.retries和retry.backoff.ms

retries引數用來配置生產者重試的次數，預設值為0，即在發生異常的時候不進行任何重試動作。訊息在從生產者發出到成功寫入伺服器之前可能發生一些臨時性的異常，比如網路抖動、leader副本的選舉等，這種異常往往是可以自行恢復的，生產者可以通過配置retries大於0的值，以此通過內部重試來恢復

retry.backoff.ms這個引數的預設值為100，它用來設定兩次重試之間的時間間隔，避免無效的頻繁重試.

4.compression.type

這個引數用來指定訊息的壓縮方式，預設值為“none”，即預設情況下，訊息不會被壓縮。該引數還可以配置為“gzip”“snappy”和“lz4”。對訊息進行壓縮可以極大地減少網路傳輸量、降低網路I/O，從而提高整體的效能。訊息壓縮是一種使用時間換空間的優化方式，如果對時延有一定的要求，則不推薦對訊息進行壓縮。

5.connections.max.idle.ms

指定在多久之後關閉限制的連線，預設值是540000（ms），即9分鐘。

6.linger.ms

指定生產者傳送 ProducerBatch 之前等待更多訊息（ProducerRecord）加入ProducerBatch 的時間，預設值為 0。生產者客戶端會在 ProducerBatch 被填滿或等待時間超過linger.ms 值時傳送出去。增大這個引數的值會增加訊息的延遲，但是同時能提升一定的吞吐量。

7.receive.buffer.bytes

設定Socket接收訊息緩衝區（SO_RECBUF）的大小，預設值為32768（B），即32KB。如果設定為-1，則使用作業系統的預設值。

8.send.buffer.bytes

設定Socket傳送訊息緩衝區（SO_SNDBUF）的大小，預設值為131072（B），即128KB。與receive.buffer.bytes引數一樣，如果設定為-1，則使用作業系統的預設值。

9.request.timeout.ms

配置Producer等待請求響應的最長時間，預設值為30000（ms）。請求超時之後可以選擇進行重試。注意這個引數需要比broker端引數replica.lag.time.max.ms的值要大，這樣可以減少因客戶端重試而引起的訊息重複的概率。