kafka存儲(chǔ)結(jié)構(gòu)以及Log清理機(jī)制

本文主要聚焦 kafka 的日志存儲(chǔ)以及日志清理相關(guān)。

日志存儲(chǔ)結(jié)構(gòu)

首先我們來看一張 kafak 的存儲(chǔ)結(jié)構(gòu)圖。

如上圖所示、kafka 中消息是以主題 topic 為基本單位進(jìn)行歸類的，這里的 topic 是邏輯上的概念，實(shí)際上在磁盤存儲(chǔ)是根據(jù)分區(qū)存儲(chǔ)的，每個(gè)主題可以分為多個(gè)分區(qū)、分區(qū)的數(shù)量可以在主題創(chuàng)建的時(shí)候進(jìn)行指定。例如下面 kafka 命令創(chuàng)建了一個(gè) topic 為 test 的主題、該主題下有 4 個(gè)分區(qū)、每個(gè)分區(qū)有兩個(gè)副本保證高可用。

./bin/kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 2 --partitions 4 --topic test

復(fù)制代碼

分區(qū)的修改除了在創(chuàng)建的時(shí)候指定。還可以動(dòng)態(tài)的修改。如下將 kafka 的 test 主題分區(qū)數(shù)修改為 12 個(gè)

./kafka-topics.sh --alter --zookeeper 127.0.0.1:2181 --topic test --partitions 12

復(fù)制代碼

分區(qū)內(nèi)每條消息都會(huì)被分配一個(gè)唯一的消息 id,也就是我們通常所說的 offset,  因此 kafak 只能保證每一個(gè)分區(qū)內(nèi)部有序性,不能保證全局有序性。

如果分區(qū)設(shè)置的合理，那么所有的消息都可以均勻的分布到不同的分區(qū)中去，這樣可以實(shí)現(xiàn)水平擴(kuò)展。不考慮多副本的情況下，一個(gè)分區(qū)對(duì)應(yīng)一個(gè) log 日志、如上圖所示。為了防止 log 日志過大，kafka 又引入了日志分段(LogSegment)的概念，將 log 切分為多個(gè) LogSegement，相當(dāng)于一個(gè)巨型文件被平均分配為相對(duì)較小的文件，這樣也便于消息的維護(hù)和清理。事實(shí)上，Log 和 LogSegement 也不是純粹物理意義上的概念，Log 在物理上只是以文件夾的形式存儲(chǔ)，而每個(gè) LogSegement 對(duì)應(yīng)于磁盤上的一個(gè)日志文件和兩個(gè)索引文件，以及可能的其他文件(比如以".txindex"為后綴的事務(wù)索引文件)。

kafak 中的 Log 對(duì)應(yīng)了一個(gè)命名為<topic>-<partition> 的文件夾。舉個(gè)例子、假如有一個(gè) test 主題，此主題下游 3 個(gè)分區(qū)，那么在實(shí)際物理上的存儲(chǔ)就是 "test-0","test-1","test-2" 這三個(gè)文件夾。

向 Log 中寫入消息是順序?qū)懭氲?。只有最后一個(gè) LogSegement 才能執(zhí)行寫入操作，在此之前的所有 LogSegement 都不能執(zhí)行寫入操作。為了方便描述，我們將最后一個(gè) LogSegement 成為"ActiveSegement"，即表示當(dāng)前活躍的日志分段。隨著消息的不斷寫入，當(dāng) ActiveSegement 滿足一定的條件時(shí)，就需要?jiǎng)?chuàng)建新的 activeSegement，之后在追加的消息寫入新的 activeSegement。

為了便于消息的檢索，每個(gè) LogSegement 中的日志文件(以".log" 為文件后綴)都有對(duì)應(yīng)的兩個(gè)文件索引：偏移量索引文件(以".index" 為文件后綴)和時(shí)間戳索引文件(以".timeindex"為文件后綴)。每個(gè) LogSegement 都有一個(gè)“基準(zhǔn)偏移量” baseOffset，用來標(biāo)識(shí)當(dāng)前 LogSegement 中第一條消息的 offset。偏移量是一個(gè) 64 位的長整形。日志文件和兩個(gè)索引文件都是根據(jù)基準(zhǔn)偏移量（baseOffset）命名的，名稱固定為 20 位數(shù)字，沒有達(dá)到的位數(shù)則用 0 填充。比如第一個(gè) LogSegment 的基準(zhǔn)偏移量為 0，對(duì)應(yīng)的日志文件為 00000000000000000000.log

示例中第 2 個(gè) LogSegment 對(duì)應(yīng)的基準(zhǔn)位移是 256，也說明了該 LogSegment 中的第一條消息的偏移量為 256，同時(shí)可以反映出第一個(gè) LogSegment 中共有 256 條消息（偏移量從 0 至 254 的消息）。

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

日志清理機(jī)制

由于 kafak 是把消息存儲(chǔ) 在磁盤上，為了控制消息的不斷增加我們就必須對(duì)消息做一定的清理和壓縮。kakfa 中的每一個(gè)分區(qū)副本都對(duì)應(yīng)的一個(gè) log 日志文件。而 Log 又分為多個(gè) LogSegement 日志分段。這樣也便于日志清理。kafka 內(nèi)部提供了兩種日志清理策略。

日志刪除

按照一定的保留策略直接刪除不符合條件的日志分段。

基于時(shí)間

我們可以通過 broker 端參數(shù) log.cleanup.policy 來設(shè)置日志清理策略，此參數(shù)的默認(rèn)值為“delete”，即采用日志刪除的清理策略。如果要采用日志壓縮的清理策略，就需要將 log.cleanup.policy 設(shè)置為“compact”，并且還需要將 log.cleaner.enable（默認(rèn)值為 true）設(shè)定為 true。通過將 log.cleanup.policy 參數(shù)設(shè)置為“delete，compact”，還可以同時(shí)支持日志刪除和日志壓縮兩種策略。日志清理的粒度可以控制到主題級(jí)別，比如與 log.cleanup.policy 對(duì)應(yīng)的主題級(jí)別的參數(shù)為 cleanup.policy，為了簡化說明，本文只采用 broker 端參數(shù)做陳述。

日志刪除任務(wù)會(huì)檢查當(dāng)前日志文件中是否有保留時(shí)間超過設(shè)定的閾值（retentionMs）來尋找可刪除的日志分段文件集合（deletableSegments），如圖下圖所示。retentionMs 可以通過 broker 端參數(shù) log.retention.hours、log.retention.minutes 和 log.retention.ms 來配置，其中 log.retention.ms 的優(yōu)先級(jí)最高，log.retention.minutes 次之，log.retention.hours 最低。默認(rèn)情況下只配置了 log.retention.hours 參數(shù)，其值為 168，故默認(rèn)情況下日志分段文件的保留時(shí)間為 7 天。

查找過期的日志分段文件，并不是簡單地根據(jù)日志分段的最近修改時(shí)間 lastModifiedTime 來計(jì)算的，而是根據(jù)日志分段中最大的時(shí)間戳 largestTimeStamp 來計(jì)算的。因?yàn)槿罩痉侄蔚?lastModifiedTime 可以被有意或無意地修改，比如執(zhí)行了 touch 操作，或者分區(qū)副本進(jìn)行了重新分配，lastModifiedTime 并不能真實(shí)地反映出日志分段在磁盤的保留時(shí)間。要獲取日志分段中的最大時(shí)間戳 largestTimeStamp 的值，首先要查詢?cè)撊罩痉侄嗡鶎?duì)應(yīng)的時(shí)間戳索引文件，查找時(shí)間戳索引文件中最后一條索引項(xiàng)，若最后一條索引項(xiàng)的時(shí)間戳字段值大于 0，則取其值，否則才設(shè)置為最近修改時(shí)間 lastModifiedTime.

若待刪除的日志分段的總數(shù)等于該日志文件中所有的日志分段的數(shù)量，那么說明所有的日志分段都已過期，但該日志文件中還要有一個(gè)日志分段用于接收消息的寫入，即必須要保證有一個(gè)活躍的日志分段 activeSegment，在此種情況下，會(huì)先切分出一個(gè)新的日志分段作為 activeSegment，然后執(zhí)行刪除操作。

刪除日志分段時(shí)，首先會(huì)從 Log 對(duì)象中所維護(hù)日志分段的跳躍表中移除待刪除的日志分段，以保證沒有線程對(duì)這些日志分段進(jìn)行讀取操作。然后將日志分段所對(duì)應(yīng)的所有文件添加上“.deleted”的后綴（當(dāng)然也包括對(duì)應(yīng)的索引文件）。最后交由一個(gè)以“delete-file”命名的延遲任務(wù)來刪除這些以“.deleted”為后綴的文件，這個(gè)任務(wù)的延遲執(zhí)行時(shí)間可以通過 file.delete.delay.ms 參數(shù)來調(diào)配，此參數(shù)的默認(rèn)值為 60000，即 1 分鐘。

基于日志大小

日志刪除任務(wù)會(huì)檢查當(dāng)前日志的大小是否超過設(shè)定的閾值（retentionSize）來尋找可刪除的日志分段的文件集合（deletableSegments），如下圖所示。retentionSize 可以通過 broker 端參數(shù) log.retention.bytes 來配置，默認(rèn)值為-1，表示無窮大。注意 log.retention.bytes 配置的是 Log 中所有日志文件的總大小，而不是單個(gè)日志分段（確切地說應(yīng)該為.log 日志文件）的大小。單個(gè)日志分段的大小由 broker 端參數(shù) log.segment.bytes 來限制，默認(rèn)值為 1073741824，即 1GB。

基于日志大小的保留策略與基于時(shí)間的保留策略類似，首先計(jì)算日志文件的總大小 size 和 retentionSize 的差值 diff，即計(jì)算需要?jiǎng)h除的日志總大小，然后從日志文件中的第一個(gè)日志分段開始進(jìn)行查找可刪除的日志分段的文件集合 deletableSegments。查找出 deletableSegments 之后就執(zhí)行刪除操作，這個(gè)刪除操作和基于時(shí)間的保留策略的刪除操作相同