搞透Kafka的存儲(chǔ)架構(gòu)，看這篇就夠了

1. 提高讀速度：利用索引，來(lái)提高查詢速度，但是有了索引，大量寫操作都會(huì)維護(hù)索引，那么會(huì)降低寫入效率。常見的如關(guān)系型數(shù)據(jù)庫(kù)：mysql等

2. 提高寫速度：這種一般是采用日志存儲(chǔ), 通過(guò)順序追加寫的方式來(lái)提高寫入速度，因?yàn)闆](méi)有索引，無(wú)法快速查詢，最嚴(yán)重的只能一行行遍歷讀取。常見的如大數(shù)據(jù)相關(guān)領(lǐng)域的基本都基于此方式來(lái)實(shí)現(xiàn)。

1. 寫操作：寫并發(fā)要求非常高，基本得達(dá)到百萬(wàn)級(jí) TPS，順序追加寫日志即可，無(wú)需考慮更新操作

2.?讀操作：相對(duì)寫操作來(lái)說(shuō)，比較簡(jiǎn)單，只要能按照一定規(guī)則高效查詢即可（offset或者時(shí)間戳）

1.? kafka 中消息是以主題 Topic 為基本單位進(jìn)行歸類的，這里的 Topic 是邏輯上的概念，實(shí)際上在磁盤存儲(chǔ)是根據(jù)分區(qū) Partition?存儲(chǔ)的, 即每個(gè) Topic 被分成多個(gè) Partition，分區(qū) Partition?的數(shù)量可以在主題 Topic 創(chuàng)建的時(shí)候進(jìn)行指定。

2. ?Partition 分區(qū)主要是為了解決 Kafka 存儲(chǔ)的水平擴(kuò)展問(wèn)題而設(shè)計(jì)的， 如果一個(gè) Topic 的所有消息都只存儲(chǔ)到一個(gè) Kafka Broker上的話，?對(duì)于 Kafka 每秒寫入幾百萬(wàn)消息的高并發(fā)系統(tǒng)來(lái)說(shuō)，這個(gè) Broker 肯定會(huì)出現(xiàn)瓶頸， 故障時(shí)候不好進(jìn)行恢復(fù)，所以 Kafka 將 Topic 的消息劃分成多個(gè) Partition，?然后均衡的分布到整個(gè) Kafka Broker 集群中。

3.??Partition?分區(qū)內(nèi)每條消息都會(huì)被分配一個(gè)唯一的消息 id,即我們通常所說(shuō)的 偏移量 Offset, ?因此 kafka 只能保證每個(gè)分區(qū)內(nèi)部有序性,并不能保證全局有序性。

4.??然后每個(gè) Partition 分區(qū)又被劃分成了多個(gè) LogSegment，這是為了防止 Log 日志過(guò)大，Kafka 又引入了日志分段(LogSegment)的概念，將 Log 切分為多個(gè) LogSegement，相當(dāng)于一個(gè)巨型文件被平均分割為一些相對(duì)較小的文件，這樣也便于消息的查找、維護(hù)和清理。這樣在做歷史數(shù)據(jù)清理的時(shí)候，直接刪除舊的?LogSegement?文件就可以了。

4.? Log 日志在物理上只是以文件夾的形式存儲(chǔ)，而每個(gè) LogSegement 對(duì)應(yīng)磁盤上的一個(gè)日志文件和兩個(gè)索引文件，以及可能的其他文件(比如以".snapshot"為后綴的快照索引文件等)

? ? ??看上圖我們知道首先向 Log 中寫入消息是順序?qū)懭氲摹?/span>但是只有最后一個(gè) LogSegement 才能執(zhí)行寫入操作，之前的所有 LogSegement 都不能執(zhí)行寫入操作。為了更好理解這個(gè)概念，我們將最后一個(gè) LogSegement 稱為"activeSegement"，即表示當(dāng)前活躍的日志分段。隨著消息的不斷寫入，當(dāng) activeSegement 滿足一定的條件時(shí)，就需要?jiǎng)?chuàng)建新的 activeSegement，之后再追加的消息會(huì)寫入新的 activeSegement。

? ? ?

? ? ? ? ? ? ?

圖7：activeSegment示意圖

? ? ? ?為了更高效的進(jìn)行消息檢索，每個(gè) LogSegment 中的日志文件（以“.log”為文件后綴）都有對(duì)應(yīng)的幾個(gè)索引文件：偏移量索引文件（以“.index”為文件后綴）、時(shí)間戳索引文件（以“.timeindex”為文件后綴）、快照索引文件 （以“.snapshot”為文件后綴）。其中每個(gè) LogSegment 都有一個(gè) Offset 來(lái)作為基準(zhǔn)偏移量（baseOffset），用來(lái)表示當(dāng)前 LogSegment 中第一條消息的 Offset。偏移量是一個(gè)64位的 Long 長(zhǎng)整型數(shù)，日志文件和這幾個(gè)索引文件都是根據(jù)基準(zhǔn)偏移量（baseOffset）命名的，名稱固定為20位數(shù)字，沒(méi)有達(dá)到的位數(shù)前面用0填充。比如第一個(gè) LogSegment 的基準(zhǔn)偏移量為0，對(duì)應(yīng)的日志文件為00000000000000000000.log。

? ? ?我們來(lái)舉例說(shuō)明，向主題topic-order中寫入一定量的消息，某一時(shí)刻topic-order-0目錄中的布局如下所示：

? ? ? ? ??? ?

圖8：log 目錄布局示意圖? ? ? ?

? ? ? ? 上面例子中?LogSegment 對(duì)應(yīng)的基準(zhǔn)位移是12768089，也說(shuō)明了當(dāng)前?LogSegment 中的第一條消息的偏移量為12768089，同時(shí)可以說(shuō)明當(dāng)前?LogSegment 中共有12768089條消息（偏移量從0至12768089的消息）。

注意每個(gè) LogSegment 中不只包含“.log”、“.index”、“.timeindex”這幾種文件，還可能包含“.snapshot”、“.txnindex”、“l(fā)eader-epoch-checkpoint”等文件, 以及 “.deleted”、“.cleaned”、“.swap”等臨時(shí)文件。

??? ? ?另外 消費(fèi)者消費(fèi)的時(shí)候，會(huì)將提交的位移保存在 Kafka 內(nèi)部的主題__consumer_offsets中，對(duì)它不了解的可以直接查看之前寫的??聊聊 Kafka Consumer 那點(diǎn)事?中的位移提交部分，下面我們來(lái)看一個(gè)整體的日志目錄結(jié)構(gòu)圖：

? ? ? ? ? ? ?

圖9：log 整體目錄布局示意圖

2

日志格式演變

? ? ??對(duì)于一個(gè)成熟的消息中間件來(lái)說(shuō)，日志格式不僅影響功能的擴(kuò)展，還關(guān)乎性能維度的優(yōu)化。所以隨著 Kafka 的迅猛發(fā)展，其日志格式也在不斷升級(jí)改進(jìn)中，Kafka 的日志格式總共經(jīng)歷了3個(gè)大版本：V0，V1和V2版本。

? ??

? ? ? 我們知道在 Kafka Partition 分區(qū)內(nèi)部都是由每一條消息進(jìn)行組成，如果日志格式設(shè)計(jì)得不夠精巧，那么其功能和性能都會(huì)大打折扣。

? ? ??

? ? ??

V0 版本

? ? ??在 Kafka 0.10.0 之前的版本都是采用這個(gè)版本的日志格式的。在這個(gè)版本中，每條消息對(duì)應(yīng)一個(gè) Offset 和 message size。Offset 用來(lái)表示它在 Partition分區(qū)中的偏移量。message size 表示消息的大小。兩者合起來(lái)總共12B，被稱為日志頭部。日志頭部跟 Record 整體被看作為一條消息。如下圖所示:

? ? ? ? ? ? ?

圖10：V0 版本日志格式示意圖

1. crc32（4B）：crc32校驗(yàn)值。校驗(yàn)范圍為magic至value之間。

2. magic（1B）：日志格式版本號(hào)，此版本的magic值為0。

3. attributes（1B）：消息的屬性?？偣舱?個(gè)字節(jié)，低3位表示壓縮類型：0? ? ? 表示NONE、1表示GZIP、2表示SNAPPY、3表示LZ4（LZ4自Kafka 0.9.x ?? ? 版本引入），其余位保留。

4. key length（4B）：表示消息的key的長(zhǎng)度。如果為-1，則沒(méi)有設(shè)置key。

5. key：可選，如果沒(méi)有key則無(wú)此字段。

6. value length（4B）：實(shí)際消息體的長(zhǎng)度。如果為-1，則消息為空。

7. value：消息體。

? ? ?

? ? ? 從上圖可以看出，V0 版本的消息最小為 14 字節(jié)，小于 14 字節(jié)的消息會(huì)被 Kafka 認(rèn)為是非法消息。

? ? ? 下面我來(lái)舉個(gè)例子來(lái)計(jì)算一條消息的具體大小，消息的各個(gè)字段值依次如下：

• CRC：對(duì)消息進(jìn)行 CRC 計(jì)算后的值；
• magic：0；
• attribute：0x00（未使用壓縮）；
• key 長(zhǎng)度：5；
• key：hello；
• value 長(zhǎng)度：5；
• value：world。
  

? ? 那么該條消息長(zhǎng)度為：4 1 1 4 5 4 5 = 24 字節(jié)。

V1?版本

? ? ? ?隨著 Kafka 版本的不斷迭代發(fā)展，?用戶發(fā)現(xiàn) V0 版本的日志格式由于沒(méi)有保存時(shí)間信息導(dǎo)致 Kafka 無(wú)法根據(jù)消息的具體時(shí)間進(jìn)行判斷，在進(jìn)行清理日志的時(shí)候只能使用日志文件的修改時(shí)間導(dǎo)致可能會(huì)被誤刪。

? ? ? ?從 V0.10.0 開始到 V0.11.0 版本之間所使用的日志格式版本為 V1，比 V0 版本多了一個(gè) timestamp 字段，表示消息的時(shí)間戳。如下圖所示：

? ? ? ? ? ? ?

圖11：V1 版本日志格式示意圖

V1 版本比 V0 版本多一個(gè) 8B 的 timestamp 字段；

那么 timestamp 字段作用：

? ? 對(duì)內(nèi)：會(huì)影響日志保存、切分策略；

? ? 對(duì)外：影響消息審計(jì)、端到端延遲等功能擴(kuò)展

? ??

? ? ? ?從上圖可以看出，V1 版本的消息最小為 22 字節(jié)，小于 22 字節(jié)的消息會(huì)被 Kafka 認(rèn)為是非法消息。

? ? ?

? ? ? ?總的來(lái)說(shuō)比 V0 版本的消息大了 8 字節(jié)，如果還是按照 V0 版本示例那條消息計(jì)算，則在 V1 版本中它的總字節(jié)數(shù)為：24 8 = 32 字節(jié)。

V0、V1?版本的設(shè)計(jì)缺陷

? ? ? ?通過(guò)上面我們分析畫出的 V0、V1 版本日志格式，我們會(huì)發(fā)現(xiàn)它們?cè)谠O(shè)計(jì)上的一定的缺陷，比如：

1. ?空間使用率低：無(wú)論 key 或 value 是否存在，都需要一個(gè)固定大小 4 字節(jié)去保存它們的長(zhǎng)度信息，當(dāng)消息足夠多時(shí)，會(huì)浪費(fèi)非常多的存儲(chǔ)空間。

2. ?消息長(zhǎng)度沒(méi)有保存：需要實(shí)時(shí)計(jì)算得出每條消息的總大小，效率低下。
3.??只保存最新消息位移。
4. ?冗余的 CRC 校驗(yàn)：即使是批次發(fā)送消息，每條消息也需要單獨(dú)保存 CRC。

V2?版本

? ? ? ?針對(duì) 上面我們分析的 關(guān)于 V0、V1 版本日志格式的缺陷，Kafka 在 0.11.0.0 版本對(duì)日志格式進(jìn)行了大幅度重構(gòu)，使用可變長(zhǎng)度類型解決了空間使用率低的問(wèn)題，增加了消息總長(zhǎng)度字段，使用增量的形式保存時(shí)間戳和位移，并且把一些字段統(tǒng)一抽取到 RecordBatch 中。

? ? ?

? ? ? ? ? ? ?

圖12：V2 版本日志格式示意圖? ? ? ??

? ? ? ? 從以上圖可以看出，V2 版本的消息批次（RecordBatch），相比 V0、V1 版本主要有以下變動(dòng)：

1.? 將?CRC 值從消息中移除，被抽取到消息批次中。

2.??增加了 procuder id、producer epoch、序列號(hào)等信息主要是為了支持冪等性以及事務(wù)消息的。

3. ?使用增量形式來(lái)保存時(shí)間戳和位移。

4.??消息批次最小為 61 字節(jié)，比 V0、V1 版本要大很多，但是在批量消息發(fā)送場(chǎng)景下，會(huì)提供發(fā)送效率，降低使用空間。

? ? ?

? ? ? ?綜上可以看出 V2 版本日志格式主要是通過(guò)可變長(zhǎng)度提高了消息格式的空間使用率，并將某些字段抽取到消息批次（RecordBatch）中，同時(shí)消息批次可以存放多條消息，從而在批量發(fā)送消息時(shí)，可以大幅度地節(jié)省了磁盤空間。

3

日志清理機(jī)制

? ? ? ?Kafka 將消息存儲(chǔ)到磁盤中，隨著寫入數(shù)據(jù)不斷增加，磁盤占用空間越來(lái)越大，為了控制占用空間就需要對(duì)消息做一定的清理操作。從上面 Kafka 存儲(chǔ)日志結(jié)構(gòu)分析中每一個(gè)分區(qū)副本（Replica）都對(duì)應(yīng)一個(gè) Log，而 Log 又可以分為多個(gè)日志分段（LogSegment），這樣就便于 Kafka 對(duì)日志的清理操作。

? ? ? ?Kafka提供了兩種日志清理策略：

1.??日志刪除（Log Retention）：按照一定的保留策略直接刪除不符合條件的日志分段（LogSegment）。

2. ?日志壓縮（Log Compaction）：針對(duì)每個(gè)消息的key進(jìn)行整合，對(duì)于有相同key的不同value值，只保留最后一個(gè)版本。

? ? ? ?這里我們可以通過(guò) Kafka Broker 端參數(shù) log.cleanup.policy 來(lái)設(shè)置日志清理策略，默認(rèn)值為 “delete”，即采用日志刪除的清理策略。如果要采用日志壓縮的清理策略，就需要將 log.cleanup.policy 設(shè)置為 “compact”，這樣還不夠，必須還要將log.cleaner.enable（默認(rèn)值為 true）設(shè)為 true。

? ? ? ?如果想要同時(shí)支持兩種清理策略， 可以直接將 log.cleanup.policy 參數(shù)設(shè)置為“delete，compact”。

? ? ? ? ? ? ? ??

3.1?日志刪除

? ? ???

? ? ? ??Kafka 的日志管理器（LogManager）中有一個(gè)專門的日志清理任務(wù)通過(guò)周期性檢測(cè)和刪除不符合條件的日志分段文件（LogSegment），這里我們可以通過(guò)?Kafka Broker 端的參數(shù)?log.retention.check.interval.ms 來(lái)配置，默認(rèn)值為300000，即5分鐘。

? ? ??

? ? ?? 在 Kafka 中一共有3種保留策略：

基于時(shí)間策略? ? ??

? ? ? ?日志刪除任務(wù)會(huì)周期檢查當(dāng)前日志文件中是否有保留時(shí)間超過(guò)設(shè)定的閾值(retentionMs)?來(lái)尋找可刪除的日志段文件集合(deletableSegments)。

? ? ??其中retentionMs可以通過(guò)?Kafka Broker 端的這幾個(gè)參數(shù)的大小判斷的

log.retention.ms > log.retention.minutes > log.retention.hours優(yōu)先級(jí)來(lái)設(shè)置，默認(rèn)情況只會(huì)配置 log.retention.hours 參數(shù)，值為168即為7天。

? ? ? ?這里需要注意：刪除過(guò)期的日志段文件，并不是簡(jiǎn)單的根據(jù)該日志段文件的修改時(shí)間計(jì)算的，而是要根據(jù)該日志段中最大的時(shí)間戳 largestTimeStamp 來(lái)計(jì)算的，首先要查詢?cè)撊罩痉侄嗡鶎?duì)應(yīng)的時(shí)間戳索引文件，查找該時(shí)間戳索引文件的最后一條索引數(shù)據(jù)，如果時(shí)間戳值大于0，則取值，否則才會(huì)使用最近修改時(shí)間（lastModifiedTime）。

? ? ? ?

? ? ?【刪除步驟】：

? ? ? 1. ?首先從 Log 對(duì)象所維護(hù)的日志段的跳躍表中移除要?jiǎng)h除的日志段，用來(lái)確保已經(jīng)沒(méi)有線程來(lái)讀取這些日志段。

? ? ? 2. ?將日志段所對(duì)應(yīng)的所有文件，包括索引文件都添加上“.deleted”的后綴。

? ??

? ? ? 3. ?最后交給一個(gè)以“delete-file”命名的延遲任務(wù)來(lái)刪除這些以“ .deleted ”為后綴的文件。默認(rèn)1分鐘執(zhí)行一次， 可以通過(guò) file.delete.delay.ms 來(lái)配置。

? ? ? ? ? ? ?

圖13：基于時(shí)間保留策略示意圖

基于日志大小策略? ? ?

? ? ? 日志刪除任務(wù)會(huì)周期檢查當(dāng)前日志大小是否超過(guò)設(shè)定的閾值(retentionSize)?來(lái)尋找可刪除的日志段文件集合(deletableSegments)。

? ? 其中?retentionSize 這里我們可以通過(guò)?Kafka Broker 端的參數(shù)log.retention.bytes來(lái)設(shè)置， 默認(rèn)值為-1，即無(wú)窮大。

? ? ?

? ? ? ?這里需要注意的是 log.retention.bytes 設(shè)置的是Log中所有日志文件的大小，而不是單個(gè)日志段的大小。單個(gè)日志段可以通過(guò)參數(shù) log.segment.bytes 來(lái)設(shè)置，默認(rèn)大小為1G。

? ? ? 【刪除步驟】：

? ? ? ?1. ?首先計(jì)算日志文件的總大小Size和retentionSize的差值，即需要?jiǎng)h除的日志總大小。

? ? ? ? 2.? 然后從日志文件中的第一個(gè)日志段開始進(jìn)行查找可刪除的日志段的文件集合(deletableSegments)

? ? ? ? 3. ?找到后就可以進(jìn)行刪除操作了。

? ? ? ? ?

? ??

圖14：基于日志大小保留策略示意圖

基于日志起始偏移量? ? ??

? ? ? 該策略判斷依據(jù)是日志段的下一個(gè)日志段的起始偏移量 baseOffset 是否小于等于 logStartOffset，如果是，則可以刪除此日志分段。

? ? ??

? ? ? 【如下圖所示 刪除步驟】：

? ? ? ? 1. ? 首先從頭開始遍歷每個(gè)日志段，日志段 1 的下一個(gè)日志分段的起始偏移量為20，小于logStartOffset的大小，將日志段1加入deletableSegments。

? ? ?

? ? ? ?2. ?日志段2的下一個(gè)日志偏移量的起始偏移量為35，也小于logStartOffset的大小，將日志分段2頁(yè)加入deletableSegments。

? ? ? ?3. ?日志段3的下一個(gè)日志偏移量的起始偏移量為50，也小于logStartOffset的大小，將日志分段3頁(yè)加入deletableSegments。

? ? ? ?4. ?日志段4的下一個(gè)日志偏移量通過(guò)對(duì)比后，在logStartOffset的右側(cè)，那么從日志段4開始的所有日志段都不會(huì)加入deletableSegments。

? ? ? ?5. ?待收集完所有的可刪除的日志集合后就可以直接刪除了。

? ? ??

? ??

圖15：基于日志起始偏移量保留策略示意圖? ??

5.2?日志壓縮

? ? ??
? ? ??日志壓縮 Log Compaction 對(duì)于有相同key的不同value值，只保留最后一個(gè)版本。如果應(yīng)用只關(guān)心 key 對(duì)應(yīng)的最新 value 值，則可以開啟 Kafka 相應(yīng)的日志清理功能，Kafka會(huì)定期將相同 key 的消息進(jìn)行合并，只保留最新的 value 值。

? ? ?

? ? ??Log Compaction 可以類比 Redis 中的 RDB 的持久化模式。我們可以想象下，如果每次消息變更都存 Kafka，在某一時(shí)刻， Kafka 異常崩潰后，如果想快速恢復(fù)，可以直接使用日志壓縮策略， 這樣在恢復(fù)的時(shí)候只需要恢復(fù)最新的數(shù)據(jù)即可，這樣可以加快恢復(fù)速度。

? ? ? ?

? ??

圖16：日志壓縮策略示意圖4

磁盤數(shù)據(jù)存儲(chǔ)

? ? ? ?我們知道 Kafka 是依賴文件系統(tǒng)來(lái)存儲(chǔ)和緩存消息，以及典型的順序追加寫日志操作，另外它使用操作系統(tǒng)的 PageCache 來(lái)減少對(duì)磁盤 I/O 操作，即將磁盤的數(shù)據(jù)緩存到內(nèi)存中，把對(duì)磁盤的訪問(wèn)轉(zhuǎn)變?yōu)閷?duì)內(nèi)存的訪問(wèn)。

? ? ??

? ? ? 在 Kafka 中，大量使用了 PageCache， 這也是 Kafka 能實(shí)現(xiàn)高吞吐的重要因素之一，?當(dāng)一個(gè)進(jìn)程準(zhǔn)備讀取磁盤上的文件內(nèi)容時(shí)，操作系統(tǒng)會(huì)先查看待讀取的數(shù)據(jù)頁(yè)是否在 PageCache 中，如果命中則直接返回?cái)?shù)據(jù)，從而避免了對(duì)磁盤的 I/O 操作；如果沒(méi)有命中，操作系統(tǒng)則會(huì)向磁盤發(fā)起讀取請(qǐng)求并將讀取的數(shù)據(jù)頁(yè)存入 PageCache 中，之后再將數(shù)據(jù)返回給進(jìn)程。同樣，如果一個(gè)進(jìn)程需要將數(shù)據(jù)寫入磁盤，那么操作系統(tǒng)也會(huì)檢查數(shù)據(jù)頁(yè)是否在頁(yè)緩存中，如果不存在，則 PageCache 中添加相應(yīng)的數(shù)據(jù)頁(yè)，最后將數(shù)據(jù)寫入對(duì)應(yīng)的數(shù)據(jù)頁(yè)。被修改過(guò)后的數(shù)據(jù)頁(yè)也就變成了臟頁(yè)，操作系統(tǒng)會(huì)在合適的時(shí)間把臟頁(yè)中的數(shù)據(jù)寫入磁盤，以保持?jǐn)?shù)據(jù)的一致性。

? ? ? 除了消息順序追加寫日志、PageCache以外， kafka?還使用了零拷貝（Zero-Copy）技術(shù)來(lái)進(jìn)一步提升系統(tǒng)性能， 如下圖所示：

? ? ? ? ? ? ?

圖17：kafka 零拷貝示意圖

? ? ? ?這里也可以查看之前寫的???Kafka 三高架構(gòu)設(shè)計(jì)剖析? 中高性能部分。

? ? ? ?

? ? ? ? 消息從生產(chǎn)到寫入磁盤的整體過(guò)程如下圖所示：

? ??

圖18：日志消息寫入磁盤過(guò)程示意圖

5總結(jié)

? ? ? ? 本文從 Kafka 存儲(chǔ)的場(chǎng)景剖析出發(fā)、kafka 存儲(chǔ)選型分析對(duì)比、再到?Kafka 存儲(chǔ)架構(gòu)設(shè)計(jì)剖析、以及 Kafka 日志系統(tǒng)架構(gòu)設(shè)計(jì)細(xì)節(jié)深度剖析，一步步帶你揭開了 Kafka 存儲(chǔ)架構(gòu)的神秘面紗。