正文

在進(jìn)行詳解之前，我想先聲明一下，本次我們進(jìn)行講解說(shuō)明的是 Kafka 消息存儲(chǔ)的信息文件內(nèi)容，不是所謂的 Kafka 服務(wù)器運(yùn)行產(chǎn)生的日志文件，這一點(diǎn)希望大家清楚。

Kafka 消息是以主題為單位進(jìn)行歸類，各個(gè)主題之間是彼此獨(dú)立的，互不影響。每個(gè)主題又可以分為一個(gè)或多個(gè)分區(qū)。每個(gè)分區(qū)各自存在一個(gè)記錄消息數(shù)據(jù)的日志文件。也就是該文要著重關(guān)注的內(nèi)容。我們根據(jù)如下的圖進(jìn)行進(jìn)一步說(shuō)明：

圖中，創(chuàng)建了一個(gè) demo-topic 主題，其存在 7 個(gè) Parition，對(duì)應(yīng)的每個(gè) Parition 下存在一個(gè) [Topic-Parition] 命名的消息日志文件。在理想情況下，數(shù)據(jù)流量分?jǐn)偟礁鱾€(gè) Parition 中，實(shí)現(xiàn)了負(fù)載均衡的效果。在分區(qū)日志文件中，你會(huì)發(fā)現(xiàn)很多類型的文件，比如：.index、.timestamp、.log、.snapshot 等，其中，文件名一致的文件集合就稱為 LogSement。我們先留有這樣的一個(gè)整體的日志結(jié)構(gòu)概念，接下來(lái)我們一一的進(jìn)行詳細(xì)的說(shuō)明其中的設(shè)計(jì)。

LogSegment

我們已經(jīng)知道分區(qū)日志文件中包含很多的 LogSegment ，Kafka 日志追加是順序?qū)懭氲?，LogSegment 可以減小日志文件的大小，進(jìn)行日志刪除的時(shí)候和數(shù)據(jù)查找的時(shí)候可以快速定位。同時(shí)，ActiveLogSegment 也就是活躍的日志分段擁有文件擁有寫(xiě)入權(quán)限，其余的 LogSegment 只有只讀的權(quán)限。

日志文件存在多種后綴文件，重點(diǎn)需要關(guān)注 .index、.timestamp、.log 三種類型。其他的日志類型功能作用，請(qǐng)查詢下面圖表：

每個(gè) LogSegment 都有一個(gè)基準(zhǔn)偏移量，用來(lái)表示當(dāng)前 LogSegment 中第一條消息的 offset。偏移量是一個(gè) 64 位的長(zhǎng)整形數(shù)，固定是20位數(shù)字，長(zhǎng)度未達(dá)到，用 0 進(jìn)行填補(bǔ)，索引文件和日志文件都由該作為文件名命名規(guī)則（00000000000000000000.index、00000000000000000000.timestamp、00000000000000000000.log）。特別說(shuō)明一下，如果日志文件名為 00000000000000000121.log ，則當(dāng)前日志文件的一條數(shù)據(jù)偏移量就是 121，偏移量是從 0 開(kāi)始的。

如果想要查看相應(yīng)文件內(nèi)容可以通過(guò) kafka-run-class.sh 腳本查看 .log ：

/data/kafka/bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.log

2.0 中可以使用 kafka-dump-log.sh 查 看.index 文件

/data/kafka/bin/kafka-dump-log.sh --files ./00000000000000000000.index

日志與索引文件

偏移量索引文件用于記錄消息偏移量與物理地址之間的映射關(guān)系。時(shí)間戳索引文件則根據(jù)時(shí)間戳查找對(duì)應(yīng)的偏移量。

Kafka 中的索引文件是以稀疏索引的方式構(gòu)造消息的索引，他并不保證每一個(gè)消息在索引文件中都有對(duì)應(yīng)的索引項(xiàng)。每當(dāng)寫(xiě)入一定量的消息時(shí)，偏移量索引文件和時(shí)間戳索引文件分別增加一個(gè)偏移量索引項(xiàng)和時(shí)間戳索引項(xiàng)，通過(guò)修改 log.index.interval.bytes 的值，改變索引項(xiàng)的密度。

切分文件

從上文中可知，日志文件和索引文件都會(huì)存在多個(gè)文件，組成多個(gè) SegmentLog，那么其切分的規(guī)則是怎樣的呢？

當(dāng)滿足如下幾個(gè)條件中的其中之一，就會(huì)觸發(fā)文件的切分：

• 當(dāng)前日志分段文件的大小超過(guò)了 broker 端參數(shù) log.segment.bytes 配置的值。log.segment.bytes 參數(shù)的默認(rèn)值為 1073741824，即 1GB。
• 當(dāng)前日志分段中消息的最大時(shí)間戳與當(dāng)前系統(tǒng)的時(shí)間戳的差值大于 log.roll.ms 或 log.roll.hours 參數(shù)配置的值。如果同時(shí)配置了 log.roll.ms 和 log.roll.hours 參數(shù)，那么 log.roll.ms 的優(yōu)先級(jí)高。默認(rèn)情況下，只配置了 log.roll.hours 參數(shù)，其值為168，即 7 天。
• 偏移量索引文件或時(shí)間戳索引文件的大小達(dá)到 broker 端參數(shù) log.index.size.max.bytes 配置的值。log.index.size.max.bytes 的默認(rèn)值為 10485760，即 10MB。
• 追加的消息的偏移量與當(dāng)前日志分段的偏移量之間的差值大于 Integer.MAX_VALUE，即要追加的消息的偏移量不能轉(zhuǎn)變?yōu)橄鄬?duì)偏移量。

為什么是 Integer.MAX_VALUE ？

在偏移量索引文件中，每個(gè)索引項(xiàng)共占用 8 個(gè)字節(jié)，并分為兩部分。相對(duì)偏移量和物理地址。

相對(duì)偏移量：表示消息相對(duì)與基準(zhǔn)偏移量的偏移量，占 4 個(gè)字節(jié)

物理地址：消息在日志分段文件中對(duì)應(yīng)的物理位置，也占 4 個(gè)字節(jié)

4 個(gè)字節(jié)剛好對(duì)應(yīng) Integer.MAX_VALUE ，如果大于 Integer.MAX_VALUE ，則不能用 4 個(gè)字節(jié)進(jìn)行表示了。

索引文件切分過(guò)程

索引文件會(huì)根據(jù) log.index.size.max.bytes 值進(jìn)行預(yù)先分配空間，即文件創(chuàng)建的時(shí)候就是最大值，當(dāng)真正的進(jìn)行索引文件切分的時(shí)候，才會(huì)將其裁剪到實(shí)際數(shù)據(jù)大小的文件。這一點(diǎn)是跟日志文件有所區(qū)別的地方。其意義降低了代碼邏輯的復(fù)雜性。

查找消息

offset 查詢

偏移量索引由相對(duì)偏移量和物理地址組成。

可以通過(guò)如下命令解析.index 文件

/data/kafka/bin/kafka-dump-log.sh --files ./00000000000000000000.index

offset:0 position:0
offset:20 position:320
offset:43 position:1220

注意：offset 與 position 沒(méi)有直接關(guān)系哦，由于存在數(shù)據(jù)刪除和日志清理。

e.g. 如何查看 偏移量為 23 的消息？

Kafka 中存在一個(gè) ConcurrentSkipListMap 來(lái)保存在每個(gè)日志分段，通過(guò)跳躍表方式，定位到在 00000000000000000000.index ，通過(guò)二分法在偏移量索引文件中找到不大于 23 的最大索引項(xiàng)，即 offset 20 那欄，然后從日志分段文件中的物理位置為320 開(kāi)始順序查找偏移量為 23 的消息。

時(shí)間戳方式查詢

在上文已經(jīng)有所提及，通過(guò)時(shí)間戳方式進(jìn)行查找消息，需要通過(guò)查找時(shí)間戳索引和偏移量索引兩個(gè)文件。

時(shí)間戳索引索引格式

e.g. 查找時(shí)間戳為 1557554753430 開(kāi)始的消息？

• 將 1557554753430 和每個(gè)日志分段中最大時(shí)間戳 largestTimeStamp 逐一對(duì)比，直到找到不小于 1557554753430 所對(duì)應(yīng)的日志分段。日志分段中的 largestTimeStamp 的計(jì)算是先查詢?cè)撊罩痉侄嗡鶎?duì)應(yīng)時(shí)間戳索引文件，找到最后一條索引項(xiàng)，若最后一條索引項(xiàng)的時(shí)間戳字段值大于 0 ，則取該值，否則去該日志分段的最近修改時(shí)間。
• 找到相應(yīng)日志分段之后，使用二分法進(jìn)行定位，與偏移量索引方式類似，找到不大于 1557554753430 最大索引項(xiàng)，也就是 [1557554753420 430]。
• 拿著偏移量為 430 到偏移量索引文件中使用二分法找到不大于 430 最大索引項(xiàng)，即 [20，320] 。
• 日志文件中從 320 的物理位置開(kāi)始查找不小于 1557554753430 數(shù)據(jù)。

注意：timestamp文件中的 offset 與 index 文件中的 relativeOffset 不是一一對(duì)應(yīng)的哦。因?yàn)閿?shù)據(jù)的寫(xiě)入是各自追加。

在偏移量索引文件中，索引數(shù)據(jù)都是順序記錄 offset ，但時(shí)間戳索引文件中每個(gè)追加的索引時(shí)間戳必須大于之前追加的索引項(xiàng)，否則不予追加。在 Kafka 0.11.0.0 以后，消息信息中存在若干的時(shí)間戳信息。如果 broker 端參數(shù) log.message.timestamp.type 設(shè)置為 LogAppendTIme ，那么時(shí)間戳必定能保持單調(diào)增長(zhǎng)。反之如果是 CreateTime 則無(wú)法保證順序。

日志清理

日志清理，不是日志刪除哦，這還是有所區(qū)別的，日志刪除會(huì)在下文進(jìn)行說(shuō)明。

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

日志刪除：按照一定的刪除策略，將不滿足條件的數(shù)據(jù)進(jìn)行數(shù)據(jù)刪除

日志壓縮：針對(duì)每個(gè)消息的 Key 進(jìn)行整合，對(duì)于有相同 Key 的不同 Value 值，只保留最后一個(gè)版本。

Kafka 提供 log.cleanup.policy 參數(shù)進(jìn)行相應(yīng)配置，默認(rèn)值：delete，還可以選擇 compact。

是否支持針對(duì)具體的 Topic 進(jìn)行配置？

答案是肯定的，主題級(jí)別的配置項(xiàng)是 cleanup.policy 。

日志刪除

Kafka 會(huì)周期性根據(jù)相應(yīng)規(guī)則進(jìn)行日志數(shù)據(jù)刪除，保留策略有 3 種：基于時(shí)間的保留策略、基于日志大小的保留策略和基于日志其實(shí)偏移量的保留策略。

基于時(shí)間

日志刪除任務(wù)會(huì)根據(jù) log.retention.hours/log.retention.minutes/log.retention.ms 設(shè)定日志保留的時(shí)間節(jié)點(diǎn)。如果超過(guò)該設(shè)定值，就需要進(jìn)行刪除。默認(rèn)是 7 天，log.retention.ms 優(yōu)先級(jí)最高。

如何查找日志分段文件中已經(jīng)過(guò)去的數(shù)據(jù)呢？

Kafka 依據(jù)日志分段中最大的時(shí)間戳進(jìn)行定位，首先要查詢?cè)撊罩痉侄嗡鶎?duì)應(yīng)的時(shí)間戳索引文件，查找時(shí)間戳索引文件中最后一條索引項(xiàng)，若最后一條索引項(xiàng)的時(shí)間戳字段值大于 0，則取該值，否則取最近修改時(shí)間。

為什么不直接選最近修改時(shí)間呢？

因?yàn)槿罩疚募梢杂幸鉄o(wú)意的被修改，并不能真實(shí)的反應(yīng)日志分段的最大時(shí)間信息。

刪除過(guò)程

• 從日志對(duì)象中所維護(hù)日志分段的跳躍表中移除待刪除的日志分段，保證沒(méi)有線程對(duì)這些日志分段進(jìn)行讀取操作。
• 這些日志分段所有文件添加 上 .delete 后綴。
• 交由一個(gè)以 "delete-file" 命名的延遲任務(wù)來(lái)刪除這些 .delete 為后綴的文件。延遲執(zhí)行時(shí)間可以通過(guò) file.delete.delay.ms 進(jìn)行設(shè)置

如果活躍的日志分段中也存在需要?jiǎng)h除的數(shù)據(jù)時(shí)？

Kafka 會(huì)先切分出一個(gè)新的日志分段作為活躍日志分段，然后執(zhí)行刪除操作。

基于日志大小

日志刪除任務(wù)會(huì)檢查當(dāng)前日志的大小是否超過(guò)設(shè)定值。設(shè)定項(xiàng)為 log.retention.bytes ，單個(gè)日志分段的大小由 log.regment.bytes 進(jìn)行設(shè)定。

刪除過(guò)程

• 計(jì)算需要被刪除的日志總大小 (當(dāng)前日志文件大小-retention值)。
• 從日志文件第一個(gè) LogSegment 開(kāi)始查找可刪除的日志分段的文件集合。
• 執(zhí)行刪除。

基于日志起始偏移量

基于日志起始偏移量的保留策略的判斷依據(jù)是某日志分段的下一個(gè)日志分段的起始偏移量是否大于等于日志文件的起始偏移量，若是，則可以刪除此日志分段。

注意：日志文件的起始偏移量并不一定等于第一個(gè)日志分段的基準(zhǔn)偏移量，存在數(shù)據(jù)刪除，可能與之相等的那條數(shù)據(jù)已經(jīng)被刪除了。

刪除過(guò)程

• 從頭開(kāi)始遍歷每一個(gè)日志分段，日志分段 1 的下一個(gè)日志分段的起始偏移量為 11，小于 logStartOffset，將 日志分段 1 加入到刪除隊(duì)列中
• 日志分段 2 的下一個(gè)日志分段的起始偏移量為 23，小于 logStartOffset，將 日志分段 2 加入到刪除隊(duì)列中
• 日志分段 3 的下一個(gè)日志分段的起始偏移量為 30，大于 logStartOffset，則不進(jìn)行刪除。

以上就是Kafka 日志存儲(chǔ)實(shí)現(xiàn)過(guò)程的詳細(xì)內(nèi)容，更多關(guān)于Kafka 日志存儲(chǔ)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論