整體架構

消息在真正發(fā)往Kafka之前，有可能需要經歷攔截器(lnterceptor）、序列化器（Serializer）和分區(qū)器（Partitioner）等一系列的作用，生產者客戶端的整體架構，如圖所示。

整個生產者客戶端由兩個線程協調運行，這兩個線程分別為:主線程和Sender 線程（發(fā)送線程）。

在主線程中由K afkaProducer 創(chuàng)建消息，然后通過可能的攔截器、序列化器和分區(qū)器的作用之后緩存到消息累加器（ RecordAccumulator ，也稱為消息收集器〉中。

Sender 線程負責從RecordAccumulator 中獲取消息并將其發(fā)送到Kafka 中。

RecordAccumulator主要用來緩存消息以便Sender線程可以批量發(fā)送，進而減少網絡傳輸的資源消耗以提升性能。???????? RecordAccumulator緩存的大小可以通過生產者客戶端參數buffer.memory配置，默認值為33554432B，即32MB。如果生產者發(fā)送消息的速度超過發(fā)送到服務器的速度，則會導致生產者空間不足，這個時候KafkaProducer的send（）方法調用要么被阻塞，要么拋出異常，這個取決于參數max.block.ms的配置，此參數的默認值為60000,即60秒。

主線程中發(fā)送過來的消息都會被迫加到RecordAccumulator的某個雙端隊列（Deque）中，在RecordAccumulator的內部為每個分區(qū)都維護了一個雙端隊列，隊列中的內容就是ProducerBatch，即Deque<ProducerBatch＞。消息寫入緩存時，追加到雙端隊列的尾部：Sender讀取消息時，從雙端隊列的頭部讀取。注意ProducerBatch不是ProducerRecord,ProducerBatch中可以包含一至多個ProducerRecord。通俗地說，ProducerRecord是生產者中創(chuàng)建的消息，而ProducerBatch是指一個消息批次，ProducerRecord會被包含在ProducerBatch中，這樣可以使字節(jié)的使用更加緊湊。與此同時，將較小的ProducerRecord拼湊成一個較大的ProducerBatch，也可以減少網絡請求的次數以提升整體的吞吐量。ProducerBatch和消息的具體格式有關。如果生產者客戶端需要向很多分區(qū)發(fā)送消息，則可以將buffer.memory參數適當調大以增加整體的吞吐量。

消息在網絡上都是以字節(jié)(Byte）的形式傳輸的，在發(fā)送之前需要創(chuàng)建一塊內存區(qū)域來保存對應的消息。在Kafka生產者客戶端中，通過java.io.ByteBuffer實現消息內存的創(chuàng)建和釋放。不過頻繁的創(chuàng)建和釋放是比較耗費資源的，在RecordAccumulator的內部還有一個BufferPool,它主要用來實現ByteBuffer的復用，以實現緩存的高效利用。不過BufferPool只針對特定大小的ByteBuffer進行管理，而其他大小的ByteBuffer不會緩存進BufferPool中，這個特定的大小由batch.size參數來指定，默認值為16384B，即16KB。可以適當地調大batch.size參數以便多緩存一些消息。

?

?

?

超強干貨來襲 云風專訪：近40年碼齡，通宵達旦的技術人生

總結

以上是生活随笔為你收集整理的深入理解Kafka（2）-Producer的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。