kafka生產者的緩存機制介紹

​

 ​

前言

今天咱們來看下kafka生產者客戶端緩存架構的設計是怎樣的？咱們先總體看下發送消息到java資料分服務端的完整過程和架構：

​

消息累加器

其實，整個生產者客戶端是由兩個線程協調運行的，一個是主線程Producer線程，一個是Sender線程。由主線程生產消息，而後緩存到消息累加器（RecordAccumulator）；而Sender線程則負責從消息累加器中不斷獲取消息，而後發送到kakfa broker。這個過程大體以下圖：

​

 

那爲何須要消息累加器呢？直接發送不是更直接？其實主要是用來緩存消息以便Sender線程能夠每次批量發送，從而減小網絡傳輸的資源消耗；而數據到服務端，服務端也能夠批量寫操做，從而減小磁盤I/O資源消耗，能夠提高性能。這個設計，也能夠運用於咱們平時的業務需求場景開發中。

 

那麼消息累加器中的結構是怎樣的呢？消息累加器的內部爲每一個區分維護一個雙端隊列Deque，隊列的內容是ProducerBatch；而ProducerBatch包含了一個或者多個ProducerRecord消息。

ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches;

具體結構以下圖：

​

 

消息Record寫入到消息累加器時，會被追加到雙端隊列的尾部。而尾部的ProducerBatch剩餘可用空間也意味着可否再寫入本次的這條消息ProducerRecord。ProducerBatch的大小和batch.size參數有着密切的關係。batch.size的參數大小控制着ProducerBatch的大小，默認是16KB大小。

 

​

 

整個過程是這樣的：

 

1. 消息寫入到消息累加器時，先尋找其對應分區的雙端隊列（若是不存在則新建隊列）
2. 從雙端隊列的尾部獲取一個ProducerBatch對象（若是不存在則新建Batch對象）
3. 判斷ProducerBatch是否還有剩餘空間寫入本次的消息Record，若是能夠則寫入（若是不能夠則須要建立一個新的ProducerBatch對象）
4. 新建ProducerBatch對象時會評估這條消息的大小是否超過batch.size參數的大小，若是不超過則以batch.size參數大小建立Batch對象（若是超過，就以消息的評估大小建立新Batch對象）。而用batch.size建立的內存空間是會被BufferPool管理的，能夠進行空間複用；而超過batch.size建立的內存空間則不會被複用，使用完成則被回收。

而整個消息累加器的緩存空間與buffer.memory參數有關。默認是32MB大小。若是生產者客戶端須要向不少區分發送比較多的消息，則能夠根據實際狀況將此參數適當調大以增長總體的吞吐量。

​

 

Sender線程

 

Sender線程則異步從消息累加器中獲取緩存的消息，而後將其轉爲指定格式的ProducerRequest對象，將Request對象請求發往各個broker了。不過，請求在從Sender線程發往broker以前還會被保存到InFlightRequests中，其主要做用是緩存了已經發出去但尚未收到服務端響應的請求。

InFlightRequests相關的有一種重要的配置參數是max.in.flight.requests.per.connection。該參數表示每一個鏈接（客戶端與broker node節點之間的網絡鏈接）的最多緩存請求數，默認值爲5個。當超過該數值以後，消費者客戶端便不能再向這個鏈接發送更多的請求了。另外也得注意，當該參數配置大於1時，因爲由於失敗重試緣由，可能會存在消息亂序的風險。

​

 

總結

 

本次咱們瞭解了生產者客戶端的兩個線程，主線程Producer線程和Sender線程的各自的職責和做用；同時也瞭解了kafka的Producer客戶端發送緩存機制。對於優秀的設計機制，咱們也能夠思考借鑑運用於其餘相似的業務開發中。

最後

最近我整理了整套《JAVA核心知識點總結》,說實話 ,做爲一名Java程序員,不論你需不須要面試都應該好好看下這份資料。拿到手老是不虧的~個人很多粉絲也所以拿到騰訊字節快手等公司的Offer

進【Java進階之路羣】，找管理員獲取哦-！

​

​

 

​