---

歡迎支持筆者新作：《深入理解Kafka:核心設(shè)計(jì)與實(shí)踐原理》和《RabbitMQ實(shí)戰(zhàn)指南》，同時(shí)歡迎關(guān)注筆者的微信公眾號(hào)：朱小廝的博客。

歡迎跳轉(zhuǎn)到本文的原文鏈接：https://honeypps.com/mq/log-deletion-of-kafka-log-retention/

---

Kafka將消息存儲(chǔ)在磁盤中，為了控制磁盤占用空間的不斷增加就需要對(duì)消息做一定的清理操作。Kafka中每一個(gè)分區(qū)partition都對(duì)應(yīng)一個(gè)日志文件，而日志文件又可以分為多個(gè)日志分段文件，這樣也便于日志的清理操作。Kafka提供了兩種日志清理策略：

日志刪除（Log Deletion）：按照一定的保留策略來直接刪除不符合條件的日志分段。

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

我們可以通過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í)支持日志刪除和日志壓縮兩種策略。日志清理的粒度可以控制到topic級(jí)別，比如與log.cleanup.policy對(duì)應(yīng)的主題級(jí)別的參數(shù)為cleanup.policy，為了簡化說明，本文只采用broker端參數(shù)做陳述，如若需要topic級(jí)別的參數(shù)可以查看官方文檔。

日志刪除

Kafka日志管理器中會(huì)有一個(gè)專門的日志刪除任務(wù)來周期性檢測和刪除不符合保留條件的日志分段文件，這個(gè)周期可以通過broker端參數(shù)log.retention.check.interval.ms來配置，默認(rèn)值為300,000，即5分鐘。當(dāng)前日志分段的保留策略有3種：基于時(shí)間的保留策略、基于日志大小的保留策略以及基于日志起始偏移量的保留策略。

1. 基于時(shí)間

日志刪除任務(wù)會(huì)檢查當(dāng)前日志文件中是否有保留時(shí)間超過設(shè)定的閾值retentionMs來尋找可刪除的的日志分段文件集合deletableSegments，參考下圖所示。retentionMs可以通過broker端參數(shù)log.retention.hours、log.retention.minutes以及l(fā)og.retention.ms來配置，其中l(wèi)og.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)槿罩痉侄蔚膌astModifiedTime可以被有意或者無意的修改，比如執(zhí)行了touch操作，或者分區(qū)副本進(jìn)行了重新分配，lastModifiedTime并不能真實(shí)地反映出日志分段在磁盤的保留時(shí)間。要獲取日志分段中的最大時(shí)間戳largestTimeStamp的值，首先要查詢該日志分段所對(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ì)從日志文件對(duì)象中所維護(hù)日志分段的跳躍表中移除待刪除的日志分段，以保證沒有線程對(duì)這些日志分段進(jìn)行讀取操作。然后將日志分段文件添加上“.deleted”的后綴，當(dāng)然也包括日志分段對(duì)應(yīng)的索引文件。最后交由一個(gè)以“delete-file”命名的延遲任務(wù)來刪除這些“.deleted”為后綴的文件，這個(gè)任務(wù)的延遲執(zhí)行時(shí)間可以通過file.delete.delay.ms參數(shù)來設(shè)置，默認(rèn)值為60000，即1分鐘。

2. 基于日志大小

日志刪除任務(wù)會(huì)檢查當(dāng)前日志的大小是否超過設(shè)定的閾值retentionSize來尋找可刪除的日志分段的文件集合deletableSegments，參考下圖所示。retentionSize可以通過broker端參數(shù)log.retention.bytes來配置，默認(rèn)值為-1，表示無窮大。注意log.retention.bytes配置的是日志文件的總大小，而不是單個(gè)的日志分段的大小，一個(gè)日志文件包含多個(gè)日志分段。

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

3. 基于日志起始偏移量

一般情況下日志文件的起始偏移量logStartOffset等于第一個(gè)日志分段的baseOffset，但是這并不是絕對(duì)的，logStartOffset的值可以通過DeleteRecordsRequest請(qǐng)求、日志的清理和截?cái)嗟炔僮餍薷摹?br />

基于日志起始偏移量的刪除策略的判斷依據(jù)是某日志分段的下一個(gè)日志分段的起始偏移量baseOffset是否小于等于logStartOffset，若是則可以刪除此日志分段。參考上圖，假設(shè)logStartOffset等于25，日志分段1的起始偏移量為0，日志分段2的起始偏移量為11，日志分段3的起始偏移為23，那么我們通過如下動(dòng)作收集可刪除的日志分段的文件集合deletableSegments：

從頭開始遍歷每個(gè)日志分段，日志分段1的下一個(gè)日志分段的起始偏移量為11，小于logStartOffset的大小，將日志分段1加入到deletableSegments中；

日志分段2的下一個(gè)日志偏移量的起始偏移量為23，也小于logStartOffset的大小，將日志分段2頁加入到deletableSegments中；

日志分段3的下一個(gè)日志偏移量在logStartOffset的右側(cè)，故從日志分段3開始的所有日志分段都不會(huì)被加入到deletableSegments中。

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

歡迎跳轉(zhuǎn)到本文的原文鏈接：https://honeypps.com/mq/log-deletion-of-kafka-log-retention/

---

歡迎支持筆者新作：《深入理解Kafka:核心設(shè)計(jì)與實(shí)踐原理》和《RabbitMQ實(shí)戰(zhàn)指南》，同時(shí)歡迎關(guān)注筆者的微信公眾號(hào)：朱小廝的博客。

---

總結(jié)

以上是生活随笔為你收集整理的Kafka日志清理之Log Deletion的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。