最近看kafka源碼，著實被它的客戶端緩沖池技術優(yōu)雅到了。

注：用到的源碼來自kafka2.2.2版本。

背景

當我們應用程序調用kafka客戶端 producer發(fā)送消息的時候，在kafka客戶端內部，會把屬于同一個topic分區(qū)的消息先匯總起來，形成一個batch。真正發(fā)往kafka服務器的消息都是以batch為單位的。如下圖所示：

這么做的好處顯而易見?？蛻舳撕头斩送ㄟ^網(wǎng)絡通信，這樣批量發(fā)送可以減少網(wǎng)絡帶來的性能開銷，提高吞吐量。

這個Batch的管理就非常值得探討了。可能有人會說，這不簡單嗎？用的時候分配一個塊內存，發(fā)送完了釋放不就行了嗎。

kafka是用java語言編寫的（新版本大部分都是用java實現(xiàn)的了），用上面的方案就是使用的時候new一個空間然后賦值給一個引用，釋放的時候把引用置為null等JVM GC處理就可以了。

看起來似乎也沒啥問題。但是在并發(fā)量比較高的時候就會頻繁的進行GC。我們都知道GC的時候有個stop the world，盡管最新的GC技術這個時間已經(jīng)非常短，依然有可能成為生產(chǎn)環(huán)境的性能瓶頸。

kafka的設計者當然能考慮到這一層。下面我們就來學習下kafka是如何對batch進行管理的。

緩沖池技術原理解析

kafka客戶端使用了緩沖池的概念，預先分配好真實的內存塊，放在池子里。

每個batch其實都對應了緩沖池中的一個內存空間，發(fā)送完消息之后，batch不再使用了，就把內存塊歸還給緩沖池。

聽起來是不是很耳熟啊？不錯，數(shù)據(jù)庫連接池，線程池等池化技術其實差不多都是這樣的原理。通過池化技術降低創(chuàng)建和銷毀帶來的開銷，提升執(zhí)行效率。

代碼是最好的文檔，，下面我們就來擼下源碼。

我們擼代碼的步驟采用的是從上往下的原則，先帶你看看緩沖池在哪里使用，然后再深入到緩存池內部深入分析。

下面的代碼做了一些刪減，值保留了跟本文相關的部分便于分析。

RecordAccumulator其實就是管理一個batch隊列，我們看到append方法實現(xiàn)其實是調用BufferPool的free方法申請（allocate）了一塊內存空間(ByteBuffer)， 然后把這個內存空空間包裝成batch添加到隊列后面。

當消息發(fā)送完成不在使用batch的時候，RecordAccumulator會調用deallocate方法歸還內存，內部其實是調用BufferPool的deallocate方法。

很明顯，BufferPool就是緩沖池管理的類，也是我們今天要討論的重點。我們先來看看分配內存塊的方法。

首先整個方法是加鎖操作的，所以支持并發(fā)分配內存。

邏輯是這樣的，當申請的內存大小等于poolableSize，則從緩存池中獲取。這個poolableSize可以理解成是緩沖池的頁大小，作為緩沖池分配的基本單位。從緩存池獲取其實就是從ByteBuffer隊列取出一個元素返回。

如果申請的內存不等于特定的數(shù)值，則向非緩存池申請。同時會從緩沖池中取一些內存并入到非緩沖池中。這個nonPooledAvailableMemory指的就是非緩沖池的可用內存大小。非緩沖池分配內存，其實就是調用ByteBuffer.allocat分配真實的JVM內存。

緩存池的內存一般都很少回收。而非緩存池的內存是使用后丟棄，然后等待GC回收。

繼續(xù)來看看batch釋放的代碼，

很簡單，也是分為兩種情況。要么直接歸還緩沖池，要么就是更新非緩沖池部分的可以內存。然后通知等待隊列里的第一個元素。