更新時(shí)間:2023-04-14 來(lái)源:黑馬程序員 瀏覽量:
Kafka的消息存儲(chǔ)在磁盤(pán)中,為了控制磁盤(pán)占用空間,Kafka需要不斷地對(duì)過(guò)去的一些消息進(jìn)行清理工作。Kafka的每個(gè)分區(qū)都有很多的日志文件,這樣也是為了方便進(jìn)行日志的清理。在Kafka中,提供日志清理和日志壓縮兩種方式:
日志刪除(Log Deletion):按照指定的策略直接刪除不符合條件的日志。
日志壓縮(Log Compaction):按照消息的key進(jìn)行整合,有相同key的但有不同value值,只保留最后一個(gè)版本。
在Kafka的broker或topic配置中:
以段(segment日志)為單位來(lái)進(jìn)行定期清理的。本節(jié)我們來(lái)詳細(xì)介紹Kafka數(shù)據(jù)處理中的日志刪除和日志壓縮。
定時(shí)日志刪除任務(wù)
Kafka日志管理器中會(huì)有一個(gè)專門(mén)的日志刪除任務(wù)來(lái)定期檢測(cè)和刪除不符合保留條件的日志分段文件,這個(gè)周期可以通過(guò)broker端參數(shù)log.retention.check.interval.ms來(lái)配置,默認(rèn)值為300,000,即5分鐘。當(dāng)前日志分段的保留策略有3種:1. 基于時(shí)間的保留策略
2. 基于日志大小的保留策略
3. 基于日志起始偏移量的保留策略
基于時(shí)間的保留策略
以下三種配置可以指定如果Kafka中的消息超過(guò)指定的閾值,就會(huì)將日志進(jìn)行自動(dòng)清理:
- log.retention.hours
- log.retention.minutes
- log.retention.ms
其中,優(yōu)先級(jí)為 log.retention.ms > log.retention.minutes > log.retention.hours。默認(rèn)情況,在broker中,配置如下:
log.retention.hours=168
也就是,默認(rèn)日志的保留時(shí)間為168小時(shí),相當(dāng)于保留7天。
刪除日志分段時(shí):
1. 從日志文件對(duì)象中所維護(hù)日志分段的跳躍表中移除待刪除的日志分段,以保證沒(méi)有線程對(duì)這些日志分段進(jìn)行讀取操作
2. 將日志分段文件添加上“.deleted”的后綴(也包括日志分段對(duì)應(yīng)的索引文件)
3. Kafka的后臺(tái)定時(shí)任務(wù)會(huì)定期刪除這些“.deleted”為后綴的文件,這個(gè)任務(wù)的延遲執(zhí)行時(shí)間可以通過(guò)file.delete.delay.ms參數(shù)來(lái)設(shè)置,默認(rèn)值為60000,即1分鐘。
設(shè)置topic 5秒刪除一次
1. 為了方便觀察,設(shè)置段文件的大小為1M。
key: segment.bytes
value: 1048576
1. 設(shè)置topic的刪除策略
key: retention.ms
value: 5000
嘗試往topic中添加一些數(shù)據(jù),等待一會(huì),觀察日志的刪除情況。我們發(fā)現(xiàn),日志會(huì)定期被標(biāo)記為刪除,然后被刪除。
基于日志大小的保留策略
日志刪除任務(wù)會(huì)檢查當(dāng)前日志的大小是否超過(guò)設(shè)定的閾值來(lái)尋找可刪除的日志分段的文件集合??梢酝ㄟ^(guò)broker端參數(shù) log.retention.bytes 來(lái)配置,默認(rèn)值為-1,表示無(wú)窮大。如果超過(guò)該大小,會(huì)自動(dòng)將超出部分刪除。
注意:
log.retention.bytes 配置的是日志文件的總大小,而不是單個(gè)的日志分段的大小,一個(gè)日志文件包含多個(gè)日志分段。
基于日志起始偏移量保留策略
每個(gè)segment日志都有它的起始偏移量,如果起始偏移量小于 logStartOffset,那么這些日志文件將會(huì)標(biāo)記為刪
日志壓縮(Log Compaction)
Log Compaction是默認(rèn)的日志刪除之外的清理過(guò)時(shí)數(shù)據(jù)的方式。它會(huì)將相同的key對(duì)應(yīng)的數(shù)據(jù)只保留一個(gè)版本。
Log Compaction執(zhí)行后,offset將不再連續(xù),但依然可以查詢Segment。
Log Compaction執(zhí)行前后,日志分段中的每條消息偏移量保持不變。Log Compaction會(huì)生成一個(gè)新的Segment文件。
Log Compaction是針對(duì)key的,在使用的時(shí)候注意每個(gè)消息的key不為空。
基于Log Compaction可以保留key的最新更新,可以基于Log Compaction來(lái)恢復(fù)消費(fèi)者的最新?tīng)顟B(tài)。