從這篇文章開始，我將對 Kafka 專項知識進行深度剖析， 今天我就來聊聊 kafka 的存儲系統架構設計， 說到存儲系統，大家可能對 MySQL 比較熟悉，也知道 MySQL 是基于 B+ tree 來作為它的索引數據結構。
Kafka 又是基于什么機制來存儲？為什么要設計成這樣？它解決了什么問題？又是如何解決的？里面又用到了哪些高大上的技術？
帶著這些疑問，我們就來和你聊一聊 Kafka 存儲架構設計背后的深度思考和實現原理。
認真讀完這篇文章，我相信你會對 Kafka 存儲架構，有更加深刻的理解。也能有思路來觸類旁通其他存儲系統的架構。
圖1：kafka 存儲架構大綱
1
kafka 存儲場景剖析
在講解 Kafka 的存儲方案之前，我們先來看看 Kafka 官網給的定義：
Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical Applications.
翻譯成中文如下：
Apache kafka 是一個開源的分布式事件流處理平臺，由成千上萬的公司用于高性能的數據管道流分析、數據集成和關鍵任務的應用程序。
了解 Kafka 的老司機都知道它是從 Linkedin 內部孵化的項目，從一開始，Kafka 就是為了解決大數據的實時日志流而生的, 每天要處理的日志量級在千億規模。對于日志流的特點主要包括 1)、數據實時產生 2)、海量數據存儲與處理，所以它必然要面臨分布式系統遇到的高并發、高可用、高性能等三高挑戰。
通過上面的背景可以得出：一切脫離業務場景談架構設計都是耍流氓
綜上我們看對于 Kafka 的存儲需求來說，要保證以下幾點：
1. 存儲的主要是消息流（可以是簡單的文本格式也可以是其他格式，對于 Broker 存儲來說，它并不關心數據本身） 2. 要支持海量數據的高效存儲、高持久化（保證重啟后數據不丟失） 3. 要支持海量數據的高效檢索（消費的時候可以通過offset或者時間戳高效查詢并處理） 4. 要保證數據的安全性和穩定性、故障轉移容錯性
2
kafka 存儲選型
有了上面的場景需求分析后， 我們接下來分析看看 Kafka 到底基于什么機制來存儲的，能否直接用現有我們了解到的關系型數據庫來實現呢？我們接著繼續深度分析。
1
存儲基本知識
我們先來了解下存儲的基本知識或者常識, 在我們的認知中，對于各個存儲介質的速度大體同下圖所示的，層級越高代表速度越快。很顯然，磁盤處于一個比較尷尬的位置，然而，事實上磁盤可以比我們預想的要快，也可能比我們預想的要慢，這完全取決于我們如何使用它。
圖2：各存儲介質對比分布（來自網絡）
關于磁盤和內存的 IO 速度，我們可以從下圖性能測試的結果看出普通機械磁盤的順序I/O性能指標是53.2M values/s，而內存的隨機I/O性能指標是36.7M values/s。由此似乎可以得出結論：磁盤的順序I/O性能要強于內存的隨機I/O性能。
圖3：磁盤和內存的 IO 速度對比（來自網絡）
另外從整個數據讀寫性能方面，有不同的實現方式，要么提高讀速度，要么提高寫速度。
1. 提高讀速度：利用索引，來提高查詢速度，但是有了索引，大量寫操作都會維護索引，那么會降低寫入效率。常見的如關系型數據庫：mysql等 2. 提高寫速度：這種一般是采用日志存儲, 通過順序追加寫的方式來提高寫入速度，因為沒有索引，無法快速查詢，最嚴重的只能一行行遍歷讀取。常見的如大數據相關領域的基本都基于此方式來實現。
2
Kafka 存儲方案剖析
上面從存儲基礎知識，以及存儲介質 IO 速度、讀寫性能方面剖析了存儲類系統的實現方式，那么我們來看看 Kafka 的存儲到底該采用哪種方式來實現呢？
對于 Kafka 來說， 它主要用來處理海量數據流，這個場景的特點主要包括：
1. 寫操作：寫并發要求非常高，基本得達到百萬級 TPS，順序追加寫日志即可，無需考慮更新操作 2. 讀操作：相對寫操作來說，比較簡單，只要能按照一定規則高效查詢即可（offset或者時間戳）
根據上面兩點分析，對于寫操作來說，直接采用順序追加寫日志的方式就可以滿足 Kafka 對于百萬TPS寫入效率要求。但是如何解決高效查詢這些日志呢？ 直接采用 MySQL 的 B+ tree 數據結構存儲是否可以？我們來逐一分析下：
如果采用 B+ tree 索引結構來進行存儲，那么每次寫都要維護索引，還需要有額外空間來存儲索引、更會出現關系型數據庫中經常出現的“數據頁分裂”等操作， 對于 Kafka 這種高并發的系統來說，這些設計都太重了，所以并不適合用。
但是在數據庫索引中，似乎有一種索引看起來非常適合此場景，即：哈希索引【底層基于Hash Table 實現】，為了提高讀速度， 我們只需要在內存中維護一個映射關系即可，每次根據 Offset 查詢消息的時候，從哈希表中得到偏移量，再去讀文件就可以快速定位到要讀的數據位置。但是哈希索引通常是需要常駐內存的，對于Kafka 每秒寫入幾百萬消息數據來說，是非常不現實的，很容易將內存撐爆, 造成 oom。
這時候我們可以設想把消息的 Offset 設計成一個有序的字段，這樣消息在日志文件中也就有序存放了，也不需要額外引入哈希表結構， 可以直接將消息劃分成若干個塊，對于每個塊，我們只需要索引當前塊的第一條消息的 Offset ，這個是不是有點二分查找算法的意思。即先根據 Offset 大小找到對應的塊， 然后再從塊中順序查找。如下圖所示：
圖4：kafka 稀疏索引查詢示意圖
這樣就可以快速定位到要查找的消息的位置了，在 Kafka 中，我們將這種索引結構叫做 “稀疏索引”。
3
kafka 存儲架構設計
上面從 Kafka 誕生背景、 存儲場景分析、存儲介質 IO 對比、以及 Kafka 存儲方案選型等幾個方面進行深度剖析， 得出了 Kafka 最終的存儲實現方案， 即基于順序追加寫日志 + 稀疏哈希索引。
接下來我們來看看 Kafka 日志存儲結構：
圖5：kafka日志存儲結構
從上圖可以看出來，Kafka 是基于「主題 + 分區 + 副本 + 分段 + 索引 」 的結構：
1. kafka 中消息是以主題 Topic 為基本單位進行歸類的，這里的 Topic 是邏輯上的概念，實際上在磁盤存儲是根據分區 Partition 存儲的, 即每個 Topic 被分成多個 Partition，分區 Partition 的數量可以在主題 Topic 創建的時候進行指定。 2. Partition 分區主要是為了解決 Kafka 存儲的水平擴展問題而設計的， 如果一個 Topic 的所有消息都只存儲到一個 Kafka Broker上的話， 對于 Kafka 每秒寫入幾百萬消息的高并發系統來說，這個 Broker 肯定會出現瓶頸， 故障時候不好進行恢復，所以 Kafka 將 Topic 的消息劃分成多個 Partition， 然后均衡的分布到整個 Kafka Broker 集群中。 3. Partition 分區內每條消息都會被分配一個唯一的消息 id,即我們通常所說的 偏移量 Offset, 因此 kafka 只能保證每個分區內部有序性,并不能保證全局有序性。 4. 然后每個 Partition 分區又被劃分成了多個 LogSegment，這是為了防止 Log 日志過大，Kafka 又引入了日志分段(LogSegment)的概念，將 Log 切分為多個 LogSegement，相當于一個巨型文件被平均分割為一些相對較小的文件，這樣也便于消息的查找、維護和清理。這樣在做歷史數據清理的時候，直接刪除舊的 LogSegement 文件就可以了。 4. Log 日志在物理上只是以文件夾的形式存儲，而每個 LogSegement 對應磁盤上的一個日志文件和兩個索引文件，以及可能的其他文件(比如以".snapshot"為后綴的快照索引文件等)
也可以直接看之前寫的 中的存儲機制部分，也有詳細的說明。
4
kafka 日志系統架構設計
了解了 Kafka 存儲選型和存儲架構設計后， 我們接下來再深度剖析下 Kafka 日志系統的架構設計。
根據上面的存儲架構剖析，我們知道 kafka 消息是按主題 Topic 為基礎單位歸類的，各個 Topic 在邏輯上是獨立的，每個 Topic 又可以分為一個或者多個 Partition，每條消息在發送的時候會根據分區規則被追加到指定的分區中，如下圖所示：
圖6：4個分區的主題邏輯結構圖
1
日志目錄布局
那么 Kafka 消息寫入到磁盤的日志目錄布局是怎樣的？接觸過 Kafka 的老司機一般都知道 Log 對應了一個命名為-的文件夾。舉個例子，假設現在有一個名為“topic-order”的 Topic，該 Topic 中有4個 Partition，那么在實際物理存儲上表現為“topic-order-0”、“topic-order-1”、“topic-order-2”、“topic-order-3” 這4個文件夾。
看上圖我們知道首先向 Log 中寫入消息是順序寫入的。但是只有最后一個 LogSegement 才能執行寫入操作，之前的所有 LogSegement 都不能執行寫入操作。為了更好理解這個概念，我們將最后一個 LogSegement 稱為"activeSegement"，即表示當前活躍的日志分段。隨著消息的不斷寫入，當 activeSegement 滿足一定的條件時，就需要創建新的 activeSegement，之后再追加的消息會寫入新的 activeSegement。
圖7：activeSegment示意圖
為了更高效的進行消息檢索，每個 LogSegment 中的日志文件（以“.log”為文件后綴）都有對應的幾個索引文件：偏移量索引文件（以“.index”為文件后綴）、時間戳索引文件（以“.timeindex”為文件后綴）、快照索引文件 （以“.snapshot”為文件后綴）。其中每個 LogSegment 都有一個 Offset 來作為基準偏移量（baseoffset），用來表示當前 LogSegment 中第一條消息的 Offset。偏移量是一個64位的 Long 長整型數，日志文件和這幾個索引文件都是根據基準偏移量（baseOffset）命名的，名稱固定為20位數字，沒有達到的位數前面用0填充。比如第一個 LogSegment 的基準偏移量為0，對應的日志文件為00000000000000000000.log。
我們來舉例說明，向主題topic-order中寫入一定量的消息，某一時刻topic-order-0目錄中的布局如下所示：
圖8：log 目錄布局示意圖
上面例子中 LogSegment 對應的基準位移是12768089，也說明了當前 LogSegment 中的第一條消息的偏移量為12768089，同時可以說明當前 LogSegment 中共有12768089條消息（偏移量從0至12768089的消息）。
注意每個 LogSegment 中不只包含“.log”、“.index”、“.timeindex”這幾種文件，還可能包含“.snapshot”、“.txnindex”、“leader-epoch-checkpoint”等文件, 以及 “.deleted”、“.cleaned”、“.swap”等臨時文件。
另外 消費者消費的時候，會將提交的位移保存在 Kafka 內部的主題__consumer_offsets中，對它不了解的可以直接查看之前寫的 中的位移提交部分，下面我們來看一個整體的日志目錄結構圖：
圖9：log 整體目錄布局示意圖
2
日志格式演變
對于一個成熟的消息中間件來說，日志格式不僅影響功能的擴展，還關乎性能維度的優化。所以隨著 Kafka 的迅猛發展，其日志格式也在不斷升級改進中，Kafka 的日志格式總共經歷了3個大版本：V0，V1和V2版本。
我們知道在 Kafka Partition 分區內部都是由每一條消息進行組成，如果日志格式設計得不夠精巧，那么其功能和性能都會大打折扣。
V0 版本
在 Kafka 0.10.0 之前的版本都是采用這個版本的日志格式的。在這個版本中，每條消息對應一個 Offset 和 message size。Offset 用來表示它在 Partition分區中的偏移量。message size 表示消息的大小。兩者合起來總共12B，被稱為日志頭部。日志頭部跟 Record 整體被看作為一條消息。如下圖所示:
圖10：V0 版本日志格式示意圖
1. crc32（4B）：crc32校驗值。校驗范圍為magic至value之間。 2. magic（1B）：日志格式版本號，此版本的magic值為0。 3. attributes（1B）：消息的屬性。總共占1個字節，低3位表示壓縮類型：0 表示NONE、1表示GZIP、2表示SNAPPY、3表示LZ4（LZ4自Kafka 0.9.x 版本引入），其余位保留。 4. key length（4B）：表示消息的key的長度。如果為-1，則沒有設置key。 5. key：可選，如果沒有key則無此字段。 6. value length（4B）：實際消息體的長度。如果為-1，則消息為空。 7. value：消息體。
從上圖可以看出，V0 版本的消息最小為 14 字節，小于 14 字節的消息會被 Kafka 認為是非法消息。
下面我來舉個例子來計算一條消息的具體大小，消息的各個字段值依次如下：
-
CRC：對消息進行 CRC 計算后的值；
-
magic：0；
-
attribute：0x00（未使用壓縮）；
-
key 長度：5；
-
key：hello；
-
value 長度：5；
-
value：world。
那么該條消息長度為：4 + 1 + 1 + 4 + 5 + 4 + 5 = 24 字節。
V1 版本
隨著 Kafka 版本的不斷迭代發展， 用戶發現 V0 版本的日志格式由于沒有保存時間信息導致 Kafka 無法根據消息的具體時間進行判斷，在進行清理日志的時候只能使用日志文件的修改時間導致可能會被誤刪。
從 V0.10.0 開始到 V0.11.0 版本之間所使用的日志格式版本為 V1，比 V0 版本多了一個 timestamp 字段，表示消息的時間戳。如下圖所示：
圖11：V1 版本日志格式示意圖
V1 版本比 V0 版本多一個 8B 的 timestamp 字段； 那么 timestamp 字段作用： 對內：會影響日志保存、切分策略； 對外：影響消息審計、端到端延遲等功能擴展
從上圖可以看出，V1 版本的消息最小為 22 字節，小于 22 字節的消息會被 Kafka 認為是非法消息。
總的來說比 V0 版本的消息大了 8 字節，如果還是按照 V0 版本示例那條消息計算，則在 V1 版本中它的總字節數為：24 + 8 = 32 字節。
V0、V1 版本的設計缺陷
通過上面我們分析畫出的 V0、V1 版本日志格式，我們會發現它們在設計上的一定的缺陷，比如：
1. 空間使用率低：無論 key 或 value 是否存在，都需要一個固定大小 4 字節去保存它們的長度信息，當消息足夠多時，會浪費非常多的存儲空間。 2. 消息長度沒有保存：需要實時計算得出每條消息的總大小，效率低下。 3. 只保存最新消息位移。 4. 冗余的 CRC 校驗：即使是批次發送消息，每條消息也需要單獨保存 CRC。
V2 版本
針對 上面我們分析的 關于 V0、V1 版本日志格式的缺陷，Kafka 在 0.11.0.0 版本對日志格式進行了大幅度重構，使用可變長度類型解決了空間使用率低的問題，增加了消息總長度字段，使用增量的形式保存時間戳和位移，并且把一些字段統一抽取到 RecordBatch 中。
圖12：V2 版本日志格式示意圖
從以上圖可以看出，V2 版本的消息批次（RecordBatch），相比 V0、V1 版本主要有以下變動：
1. 將 CRC 值從消息中移除，被抽取到消息批次中。 2. 增加了 procuder id、producer epoch、序列號等信息主要是為了支持冪等性以及事務消息的。 3. 使用增量形式來保存時間戳和位移。 4. 消息批次最小為 61 字節，比 V0、V1 版本要大很多，但是在批量消息發送場景下，會提供發送效率，降低使用空間。
綜上可以看出V2 版本日志格式主要是通過可變長度提高了消息格式的空間使用率，并將某些字段抽取到消息批次（RecordBatch）中，同時消息批次可以存放多條消息，從而在批量發送消息時，可以大幅度地節省了磁盤空間。
3
日志清理機制
Kafka 將消息存儲到磁盤中，隨著寫入數據不斷增加，磁盤占用空間越來越大，為了控制占用空間就需要對消息做一定的清理操作。從上面 Kafka 存儲日志結構分析中每一個分區副本（Replica）都對應一個 Log，而 Log 又可以分為多個日志分段（LogSegment），這樣就便于 Kafka 對日志的清理操作。
Kafka提供了兩種日志清理策略：
1. 日志刪除（Log Retention）：按照一定的保留策略直接刪除不符合條件的日志分段（LogSegment）。 2. 日志壓縮（Log Compaction）：針對每個消息的key進行整合，對于有相同key的不同value值，只保留最后一個版本。
這里我們可以通過 Kafka Broker 端參數log.cleanup.policy來設置日志清理策略，默認值為 “delete”，即采用日志刪除的清理策略。如果要采用日志壓縮的清理策略，就需要將 log.cleanup.policy 設置為“compact”，這樣還不夠，必須還要將log.cleaner.enable（默認值為 true）設為 true。
如果想要同時支持兩種清理策略， 可以直接將 log.cleanup.policy 參數設置為“delete，compact”。
3.1 日志刪除
Kafka 的日志管理器（LogManager）中有一個專門的日志清理任務通過周期性檢測和刪除不符合條件的日志分段文件（LogSegment），這里我們可以通過 Kafka Broker 端的參數log.retention.check.interval.ms來配置，默認值為300000，即5分鐘。
在 Kafka 中一共有3種保留策略：
基于時間策略
日志刪除任務會周期檢查當前日志文件中是否有保留時間超過設定的閾值(retentionMs)來尋找可刪除的日志段文件集合(deletableSegments)。
其中retentionMs可以通過 Kafka Broker 端的這幾個參數的大小判斷的
log.retention.ms > log.retention.minutes > log.retention.hours優先級來設置，默認情況只會配置 log.retention.hours 參數，值為168即為7天。
這里需要注意：刪除過期的日志段文件，并不是簡單的根據該日志段文件的修改時間計算的，而是要根據該日志段中最大的時間戳 largestTimeStamp 來計算的，首先要查詢該日志分段所對應的時間戳索引文件，查找該時間戳索引文件的最后一條索引數據，如果時間戳值大于0，則取值，否則才會使用最近修改時間（lastModifiedTime）。
【刪除步驟】：
1. 首先從 Log 對象所維護的日志段的跳躍表中移除要刪除的日志段，用來確保已經沒有線程來讀取這些日志段。
2. 將日志段所對應的所有文件，包括索引文件都添加上“.deleted”的后綴。
3. 最后交給一個以“delete-file”命名的延遲任務來刪除這些以“ .deleted ”為后綴的文件。默認1分鐘執行一次， 可以通過 file.delete.delay.ms 來配置。
圖13：基于時間保留策略示意圖
基于日志大小策略
日志刪除任務會周期檢查當前日志大小是否超過設定的閾值(retentionSize)來尋找可刪除的日志段文件集合(deletableSegments)。
其中 retentionSize 這里我們可以通過 Kafka Broker 端的參數log.retention.bytes來設置， 默認值為-1，即無窮大。
這里需要注意的是 log.retention.bytes 設置的是Log中所有日志文件的大小，而不是單個日志段的大小。單個日志段可以通過參數 log.segment.bytes 來設置，默認大小為1G。
【刪除步驟】：
1. 首先計算日志文件的總大小Size和retentionSize的差值，即需要刪除的日志總大小。
2. 然后從日志文件中的第一個日志段開始進行查找可刪除的日志段的文件集合(deletableSegments)
3. 找到后就可以進行刪除操作了。
圖14：基于日志大小保留策略示意圖
基于日志起始偏移量
該策略判斷依據是日志段的下一個日志段的起始偏移量 baseOffset 是否小于等于 logStartOffset，如果是，則可以刪除此日志分段。
【如下圖所示 刪除步驟】：
1. 首先從頭開始遍歷每個日志段，日志段 1 的下一個日志分段的起始偏移量為20，小于logStartOffset的大小，將日志段1加入deletableSegments。
2. 日志段2的下一個日志偏移量的起始偏移量為35，也小于logStartOffset的大小，將日志分段2頁加入deletableSegments。
3. 日志段3的下一個日志偏移量的起始偏移量為50，也小于logStartOffset的大小，將日志分段3頁加入deletableSegments。
4. 日志段4的下一個日志偏移量通過對比后，在logStartOffset的右側，那么從日志段4開始的所有日志段都不會加入deletableSegments。
5. 待收集完所有的可刪除的日志集合后就可以直接刪除了。
圖15：基于日志起始偏移量保留策略示意圖
5.2 日志壓縮
日志壓縮 Log Compaction 對于有相同key的不同value值，只保留最后一個版本。如果應用只關心 key 對應的最新 value 值，則可以開啟 Kafka 相應的日志清理功能，Kafka會定期將相同 key 的消息進行合并，只保留最新的 value 值。
Log Compaction 可以類比 redis 中的 RDB 的持久化模式。我們可以想象下，如果每次消息變更都存 Kafka，在某一時刻， Kafka 異常崩潰后，如果想快速恢復，可以直接使用日志壓縮策略， 這樣在恢復的時候只需要恢復最新的數據即可，這樣可以加快恢復速度。
圖16：日志壓縮策略示意圖
4
磁盤數據存儲
我們知道 Kafka 是依賴文件系統來存儲和緩存消息，以及典型的順序追加寫日志操作，另外它使用操作系統的 PageCache 來減少對磁盤 I/O 操作，即將磁盤的數據緩存到內存中，把對磁盤的訪問轉變為對內存的訪問。
在 Kafka 中，大量使用了 PageCache， 這也是 Kafka 能實現高吞吐的重要因素之一， 當一個進程準備讀取磁盤上的文件內容時，操作系統會先查看待讀取的數據頁是否在 PageCache 中，如果命中則直接返回數據，從而避免了對磁盤的 I/O 操作；如果沒有命中，操作系統則會向磁盤發起讀取請求并將讀取的數據頁存入 PageCache 中，之后再將數據返回給進程。同樣，如果一個進程需要將數據寫入磁盤，那么操作系統也會檢查數據頁是否在頁緩存中，如果不存在，則 PageCache 中添加相應的數據頁，最后將數據寫入對應的數據頁。被修改過后的數據頁也就變成了臟頁，操作系統會在合適的時間把臟頁中的數據寫入磁盤，以保持數據的一致性。
除了消息順序追加寫日志、PageCache以外， kafka 還使用了零拷貝（Zero-Copy）技術來進一步提升系統性能， 如下圖所示：
圖17：kafka 零拷貝示意圖
這里也可以查看之前寫的 中高性能部分。
消息從生產到寫入磁盤的整體過程如下圖所示：
圖18：日志消息寫入磁盤過程示意圖
