日志布局

Ka飲a 中的消息是以主題為基本單位進行歸類的， 各個主題在邏輯
上相互獨立。 每個主題又可以分為一個或多個分區(qū)， 分區(qū)的數(shù)量可以在主題創(chuàng)建的時候指定，也可以在之后修改。 每條消息在發(fā)送的時候會根據(jù)分區(qū)規(guī)則被追加到指定的分區(qū)中， 分區(qū)中的每條消息都會被分配一個唯 一的序列號， 也就是通常所說的偏移量(o?set )。

如果分區(qū)規(guī)則設置得合理， 那么所有的消息可以均勻地分布到不同的分區(qū)中， 這樣就可以實現(xiàn)水平擴展。 不考慮多副本的情況， 一個分區(qū)對應一個日志(Log)。 為了防止Log過大，Ka住a又引入了日志分段(LogSegment)的概念，將Log切分為多個LogSegment, 相當于 一個巨型文件被平均分配為多個相對較小的文件， 這樣也便于消息的維護和清理。 事實上， Log 和LogSegnient也不是純粹物理意義上的概念， Log在物理上只以文件夾的形式存儲， 而每個LogSegment對應于磁盤上的 一個日志文件 和兩個索引文件， 以及可能的其他文件（比如以" . txnindex"為后綴的事務索引文件）。如下圖簡單描繪了日志存儲結構。

向Log中追加 消息時是順序寫入的，只有最后 一 個LogSegment才能執(zhí)行寫入操作， 在此之 前所有的LogSegment都 不能寫入數(shù)據(jù)。 為了方便描述， 我們將最后 一 個LogSegment稱為"activeSegment" , 即表示當前活躍的日志分段。 隨著消息的不斷寫入 ， 當activeSegment滿足一 定 的條件 時 ， 就需要創(chuàng)建新的activeSegment, 之后追加的消息將寫入新的activeSegment。

為了便于消息的檢索， 每個LogSegment中的日志文件 （以 " .log"為文件后綴）都有對應的兩個索引文件：偏移量 索引文件（以".index"為文件后綴）和時間戳索引文件（以".timeindex"為文件后綴） 。 每個LogSegment都有 一 個基準偏移量baseO?set, 用來表示當前LogSegment中第 一 條消息的o?set。 偏移量是一 個64位的長整型數(shù)， 日志文件和兩個索引文件都是根據(jù) 基準偏移量(baseO?set)命名 的， 名稱固定為20 位數(shù)字， 沒有達到的位數(shù)則用0填充。 比如第一 個LogSegment的基準偏移量為O, 對應的日志文件為00000000000000000000.log。

日志索引

每個日志分段文件對應了兩個索引文件，主要用來提高查找消息的效率。偏移量索引文件用來建立消息偏移量（ offset ）到物理地址之間的映射關系，方便快速定位消息所在的物理文件位置；時間戳索引文件則根據(jù)指定的時間戳（ timestamp ）來查找對應的偏移量信息。

Kafka 中的索引文件以稀疏索引（ sparse index ）的方式構造消息的索引，它并不保證每個消息在索引文件中都有對應的索引 I頁 。 每當寫入一定量（由 broker 端參數(shù) log.index.interval.bytes 指定，默認值為 4096 ，即 4KB ）的消息時，偏移量索引文件和時間戳索引文件分別增加一個偏移量索引項和時間戳索引項，增大或減小 log.index . interval.bytes的值，對應地可以增加或縮小索引項的密度。

稀疏索引通過 MappedByteBuffer 將索引文件映射到內存中，以加快索引的查詢速度。偏移量索引文件中的偏移量是單調遞增的，查詢指定偏移量時，使用二分查找法來快速定位偏移量的位置，如果指定的偏移量不在索引文件中，則會返回小于指定偏移量的最大偏移量 。 時間戳索引文件中的時間戳也保持嚴格的單調遞增，查詢指定時間戳時，也根據(jù)二分查找法來查找不大于該時間戳的最大偏移量，至于要找到對應的物理文件位置還需要根據(jù)偏移量索引文件來進行再次定位。稀疏索引的方式是在磁盤空間、內存空間、查找時間等多方面之間的一個折中。

對非當前活躍的日志分段而言，其對應的索引文件內容己經固定而不需要再寫入索引項，所以會被設定為只讀 。 而對當前活躍的日志分段 C activeSegment ）而言，索引文件還會追加更多的索引項，所以被設定為可讀寫。在索引文件切分的時候， Kafka 會關閉當前正在寫入的索引文件并置為只讀模式，同時以可讀寫的模式創(chuàng)建新的索引文件，索引文件的大小由 broker 端參數(shù) log . index.size . max.bytes 配置。 Kafka 在創(chuàng)建索引文件的時候會為其預分配log.index . size .m ax . bytes 大小的空間，注意這一點與日志分段文件不同，只有當索引文件進行切分的時候， Kafka 才會把該索引文件裁剪到實際的數(shù)據(jù)大小 。 也就是說，與當前活躍的日志分段對應的索引文件的大小固定為 log . index.s 工 ze.max.bytes ，而其余日志分段對應的索引文件的大小為實際的占用空間。

日志清理

Kafka 將 消息存儲在磁盤中，為了 控制磁盤占用空間的不斷增加就需要對消息做一 定的清理操作。 Kafka 中 每 一個分區(qū)副本都對應 一個 Log, 而Log又可以分為多個日志分段， 這樣也便于日志的清理操作。 Kafka提供了兩種日志清理策略。

• (1)日志刪除(LogRetention): 按照 一 定的保留策略直接刪除不符合條件的日志分段。
• (2)日志壓縮 (Log Compaction): 針對每個消息的key進行整合， 對千有相同 key的不 同value 值， 只保留最后 一個版本。

我們可以通過broker端參數(shù)log.cleanup.police 來設置 日志清理策略，此參數(shù)的默認
值為"delete " , 即采用日志刪除的清理策略。 如果要采用日志壓縮的清理策略， 就需要將log.cleanup.police 設置為"compact" , 并且還需要將log.cleaner.enable (默認值
為true )設定為true。 通過將log.cleanup.police 參數(shù) 設置為"del ete,compact" , 還可以
同時支持日志刪除和日志壓縮兩種策略。 日志清理的粒度可以控制到主題級別， 比如與log.cleanup.police 對應的主題級別的參數(shù)為cleanup.police。

日志刪除

在Kafka 的日志管理器中會有一個專門的日志刪除任務來周期性地檢測和刪除不符合 保留條件的日志分段文件，這個周期可以通過broker端參數(shù)log.retention.check.interval.ms來配置 ，默認值為300000, 即5分鐘。 當前日志分段的保留策略有3 種：基于時間 的保留策略、基于日志大小的保留策略 和基于日志起始偏移量的保留策略 。實際應用中這些策略只要命中一個就可以觸發(fā)日志刪除。

基于時間

日志刪除任務會檢查當前日志文件中是否有保留時間超過設定的闕值(retentionMs)來尋找可刪除的日志分段文件集合(deletableSegments)。etentionMs可以通過broker
端參數(shù)log.retention.hours、log.retention.minutes和log.retention.ms來配
置 ， 其 中 log.retention.ms 的優(yōu)先級最高， log.retention.minutes 次之，
log.retention.hours最低。 默認情況下只配置 了log.retention.hours參數(shù)， 其值為
168, 故默認情況下日志分段文件的保留時間為7天。

查找過期的日志分段文件，并不是簡單地根據(jù)日志分段的最近修改時間lastModifiedTime來計算的， 而是根據(jù)日志分段中最大的時間戳largestTimeStamp來計算的。 因為日志分段的lastModifiedTime可以被有意或無意地修改， 比如執(zhí)行了touch操作 ， 或者分區(qū)副本進行了重新分配，lastModifiedTime并不能真實地反映出日志分段在磁盤的保留時間。要獲取日志分段中的最大時間戳largestTimeStamp的值， 首先要查詢該日志分段所對應的時間戳索引文件，查找 時間戳索引文件中最后 一條索引項， 若最后 一 條索引項的時間戳字段值 大于O,則取其值， 否則才設置為最近修改時間 lastModifedTime 。

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

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

基千日志大小

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

基于日志大小的保留策略與基于時間的保留策略類似， 首先計算日志文件的總大小 size 和retentionSize的差值 di?, 即計算需要刪除的日志總大小， 然后從日志文件中的第 一個日志分段開始進行查找可刪除的日志分段的文件集合deletableSegments。 查找出deletableSegments之后就執(zhí)行刪除操作， 這個刪除操作和基千時間的保留策略的刪除操作相同， 這里不再贅述。

基于日志起始偏移量

一般情況下， 日志文件的起始偏移量 logStartO?set 等于第 一個日志分段的baseO?set , 但這并不 是絕對的， logStartO?set 的值 可 以 通 過DeleteRecordsRequest請求、 日志的清理和截斷等操作 進行修改。

基于日志起始偏移量的保留策略的判斷依據(jù)是某日志分段的下一個日志分段的起始偏移量baseO?set 是否小于等于logStartO?set, 若是， 則可以刪除此日志分段 。 如圖所示， 假設logStartO?set等于25, 日志分段 l 的起始偏移量為O, 日志分段2的起始偏移量為11, 日志分段 3的起始偏移量為23, 通過如下動作收集可刪除的日志分段的文件集合deletableSegments:

• (1 )從頭開始遍歷每個日志分段 ， 日志分段 l 的下一個日志分段的起始偏移量為11, 小 于logStartO?set的大小，將日志分段 l加入deletableSegments。
• (2)日志分段2的下 一個日志偏移量的起始偏移量為23 , 也小于logStartO?set的大小，將日志分段2頁加入deletableSegments。
• (3)日志分段3的下 一個日志偏移量在logStartO?set的右側， 故從日志分段3開始的所有日志分段都不會加入deletableSegments。

收集完可刪除的日志分段的文件集合之后的刪除操作同基于日志大小的保留策略和基千時間的保留策略相同， 這里不再贅述。

日志壓縮

Kafa中的Log Compaction是指在默認的日志刪除(Log Retention)規(guī)則之外提供的 一種
清理過時數(shù)據(jù)的方式。 如圖所示， Log Compaction對于有相同key的不同value值， 只保留最后 一個版本。如果應用只關心key對應的最新value值，則可以開啟Kafka的日志清理功能，Kafa會定期將相同key的消息進行合并， 只保留最新的value值。

Log Compaction執(zhí)行前后， 日志分段中的每條消息的偏移量和寫入時的偏移量保待 一致。Log Compaction 會生成新的日志分段 文件， 日志分段中每條消息的物理位置會重新按照新文件來組織。 Log Compaction執(zhí)行過后的偏移量不再是連續(xù)的， 不過這并不影響日志的查詢。

Log Compaction會保留key相應的最新value值 ，那么當需要刪除 一個key時怎么辦? Kafka提供了 一個墓碑消息(tombstone)的概念， 如果 一 條消息的key不為null, 但是其value為null,那么此消息就是墓碑消息 。 日志清理線程發(fā)現(xiàn)墓碑消息時會先進行常規(guī)的清理， 并保留墓碑消息 一 段 時間。

Log Compaction執(zhí)行過后的日志分段的大小會比原先的日志分段的要小， 為了防止出現(xiàn)太多的小 文件， Kafa在實際清理過程 中并不對單個的日志分段進行單獨清理，而是將日志文件中o?set從0至frstUncleanableOffset的所有日志分段進行分組，每個日志分段只屬于一 組，分組策略為：按照日志分段的順序遍歷， 每 組中日志分段的占用空間大小之和不超過segmentSize（可以通過broker端參數(shù)log.segment.bytes設置， 默認值為1GB) , 且對應的索引文件占用大小之和不超過maxindexSize(可以通過broker端參數(shù)log.index.interval.bytes設置， 默認值為10MB) 。 同 一個組的多個日志分段清理過后， 只會生成 一個新的日志分段 。

總結

注意將 日志壓縮和日志刪除區(qū)分開， 日志刪除是指清除整個日志分段， 而日志壓縮是針對相同key的消息的合并清理 。