RocketMQ broker文件清理源碼解析
1. broker 清理文件介紹
本系列RocketMQ4.8註釋github地址，希望對大傢有所幫助，要是覺得可以的話麻煩給點一下Star哈
1.1 哪些文件需要清理
首先我們需要介紹下在RocketMQ中哪些文件需要清理，其實可以想一想，在RocketMQ中哪些文件是一直在往裡面寫入東西的，最容易想到的就是commitlog 瞭，因為在一個broker 進程中，所有的普通消息，事務消息，系統消息啥的都往這個commitlog中寫，隨著時間的越來越長，然後commitlog就會越積攢越多，肯定會有磁盤放不下的那一天，而且我們消息消費完成後，那些被消費完成後的消息其實作用就很小瞭，可能會有這麼一個場景，比如說我線上出現瞭某個問題，我想看下關於這個問題的消息有沒有被消費到，可能你會用到這個消息，但是這種問題一般就是比較緊急的，最近實效的，之前那些消息其實作用就基本沒有瞭，所以就需要清理掉之前的消息。其實不光commitlog需要清理，還需要清理一下ConsumeQueue 與indexFile ， 因為你commitlog裡面的消息都被清理瞭，ConsumeQueue 與indexFile 再保存著之前的一些數據，就是純粹浪費空間瞭。
所以說 broker 文件清理主要是清理commitlog ， ConsumeQueue ， indexFile。
1.2 RocketMQ文件清理的機制
我們介紹下RocketMQ文件清理的機制，RocketMQ默認是清理72小時之前的消息，然後它有幾個觸發條件， 默認是凌晨4點觸發清理， 除非你你這個磁盤空間占用到75% 以上瞭。在清理commitlog 的時候，並不是一條消息一條消息的清理，拿到所有的MappedFile（拋去現在還在用著的，也就是最後一個） ，然後比對每個MappedFile的最後一條消息的時間，如果是72小時之前的就把MappedFile對應的文件刪除瞭，銷毀對應MappedFile，這種情況的話隻要你MappedFile 最後一條消息還在存活實效內的話，它就不會清理你這個MappedFile，就算你這個MappedFile 靠前的消息過期瞭。但是有一種情況它不管你消息超沒超過72小時，直接就是刪，那就是磁盤空間不足的時候，也就是占瞭85%以上瞭，就會立即清理。
清理完成commitlog 之後，就會拿到commitlog中最小的offset ，然後去ConsumeQueue與indexFile中把小於offset 的記錄刪除掉。清理ConsumeQueue 的時候也是遍歷MappedFile ，然後它的最後一條消息（unit）小於commitlog中最小的offset 的話，就說明這個MappedFile都小於offset ，因為他們是順序追加寫的，這個MappedFile 就會清理掉，如果你MappedFile 最後一個unit不是小於offset 的話，這個MappedFile 就不刪瞭。
2. 源碼解析
我們來看下源碼是怎樣實現的： 在broker 存儲器DefaultMessageStore 啟動（start）的時候，會添加幾個任務調度，其中有一個就是文件清理的:
private void addScheduleTask() {
// todo 清理過期文件 每隔10s
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
// todo
DefaultMessageStore.this.cleanFilesPeriodically();
}
}, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS);
...
}
默認是10s執行一次，可以看到它調用瞭DefaultMessageStore 的cleanFilesPeriodically方法：
private void cleanFilesPeriodically() {
// todo 清除CommitLog文件
this.cleanCommitLogService.run();
// todo 清除ConsumeQueue文件
this.cleanConsumeQueueService.run();
}
2.1 清理commitlog
我們先來看下關於commitlog的清理工作:
public void run() {
try {
// todo 刪除過期文件
this.deleteExpiredFiles();
this.redeleteHangedFile();
} catch (Throwable e) {
DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
}
}
我們看下deleteExpiredFiles 方法的實現:
private void deleteExpiredFiles() {
int deleteCount = 0;
// 文件保留時間，如果超過瞭該時間，則認為是過期文件，可以被刪除
long fileReservedTime = DefaultMessageStore.this.getMessageStoreConfig().getFileReservedTime();
// 刪除物理文件的間隔時間，在一次清除過程中，可能需要被刪除的文件不止一個，該值指定兩次刪除文件的間隔時間
int deletePhysicFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteCommitLogFilesInterval();
// 在清除過期文件時，如
//果該文件被其他線程占用（引用次數大於0，比如讀取消息），此時會
//阻止此次刪除任務，同時在第一次試圖刪除該文件時記錄當前時間
//戳，destroyMapedFileIntervalForcibly表示第一次拒絕刪除之後能
//保留文件的最大時間，在此時間內，同樣可以被拒絕刪除，超過該時
//間後，會將引用次數設置為負數，文件將被強制刪除
int destroyMapedFileIntervalForcibly = DefaultMessageStore.this.getMessageStoreConfig().getDestroyMapedFileIntervalForcibly();
// 指定刪除文件的時間點，RocketMQ通過deleteWhen設置每天在
//固定時間執行一次刪除過期文件操作，默認凌晨4點
boolean timeup = this.isTimeToDelete();
// todo 檢查磁盤空間是否充足，如果磁盤空間不充足，則返回true，表示應該觸發過期文件刪除操作
boolean spacefull = this.isSpaceToDelete();
// 預留手工觸發機制，可以通過調用excuteDeleteFilesManualy
//方法手工觸發刪除過期文件的操作，目前RocketMQ暫未封裝手工觸發
//文件刪除的命令
boolean manualDelete = this.manualDeleteFileSeveralTimes > 0;
if (timeup || spacefull || manualDelete) {
if (manualDelete)
this.manualDeleteFileSeveralTimes--;
boolean cleanAtOnce = DefaultMessageStore.this.getMessageStoreConfig().isCleanFileForciblyEnable() && this.cleanImmediately;
log.info("begin to delete before {} hours file. timeup: {} spacefull: {} manualDeleteFileSeveralTimes: {} cleanAtOnce: {}",
fileReservedTime,
timeup,
spacefull,
manualDeleteFileSeveralTimes,
cleanAtOnce);
fileReservedTime *= 60 * 60 * 1000;
// todo 文件的銷毀和刪除
deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(fileReservedTime, deletePhysicFilesInterval,
destroyMapedFileIntervalForcibly, cleanAtOnce);
if (deleteCount > 0) {
} else if (spacefull) {
log.warn("disk space will be full soon, but delete file failed.");
}
}
}
開始幾個參數，一個是文件保留實效默認是72小時，你可以使用fileReservedTime來配置，一個是刪除文件的間隔100ms，再就是強行銷毀MappedFile的120s（這個為啥要強行銷毀，因為它還害怕還有地方用著這個MappedFile，它有個專門的引用計數器，比如說我還有地方要讀它的消息，這個時候計數器就是+1的）。
接著就是判斷到沒到刪除的那個時間，它默認是凌晨4點才能刪除：
private boolean isTimeToDelete() {
// 什麼時候刪除，默認是凌晨4點 -> 04
String when = DefaultMessageStore.this.getMessageStoreConfig().getDeleteWhen();
// 判斷是不是到點瞭 就是判斷的當前小時 是不是等於 默認的刪除時間
if (UtilAll.isItTimeToDo(when)) {
DefaultMessageStore.log.info("it's time to reclaim disk space, " + when);
return true;
}
return false;
}
再接著就是看看空間是不是充足，看看磁盤空間使用占比是什麼樣子的:
private boolean isSpaceToDelete() {
// 表示CommitLog文件、ConsumeQueue文件所在磁盤分區的最大使用量，如果超過該值，則需要立即清除過期文件
double ratio = DefaultMessageStore.this.getMessageStoreConfig().getDiskMaxUsedSpaceRatio() / 100.0;
// 表示是否需要立即執行清除過期文件的操作
cleanImmediately = false;
{
// 當前CommitLog目錄所在的磁盤分區的磁盤使用率，通過File#getTotalSpace方法獲取文件所在磁盤分區的總容量，
//通過File#getFreeSpace方法獲取文件所在磁盤分區的剩餘容量
double physicRatio = UtilAll.getDiskPartitionSpaceUsedPercent(getStorePathPhysic());
// diskSpaceWarningLevelRatio:默認0.90。如果磁盤分區使用率超過該閾值，將設置磁盤為不可寫，此時會拒絕寫入新消息
// 如果當前磁盤分區使用率大於diskSpaceWarningLevelRatio，應該立即啟動過期文件刪除操作
if (physicRatio > diskSpaceWarningLevelRatio) {
// 設置 磁盤不可寫
boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskFull();
if (diskok) {
DefaultMessageStore.log.error("physic disk maybe full soon " + physicRatio + ", so mark disk full");
}
cleanImmediately = true;
//diskSpaceCleanForciblyRatio：默認0.85 如果磁盤分區使用超過該閾值，建議立即執行過期文件刪除，但不會拒絕寫入新消息
// 如果當前磁盤分區使用率大於diskSpaceCleanForciblyRatio，建議立即執行過期文件清除
} else if (physicRatio > diskSpaceCleanForciblyRatio) {
cleanImmediately = true;
} else {
// 設置 磁盤可以寫入
boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskOK();
if (!diskok) {
DefaultMessageStore.log.info("physic disk space OK " + physicRatio + ", so mark disk ok");
}
}
// 如果當前磁盤使用率小於diskMaxUsedSpaceRatio，則返回false，表示磁盤使用率正常，
// 否則返回true，需要執行刪除過期文件
if (physicRatio < 0 || physicRatio > ratio) {
DefaultMessageStore.log.info("physic disk maybe full soon, so reclaim space, " + physicRatio);
return true;
}
}
/**
* 對consumeQueue 做同樣的判斷
*/
...
return false;
}
這裡其實不光是判斷 commitlog的存儲區域，後面還有段判斷ConsumeQueue的存儲區域的，然後與這塊邏輯一樣，就沒有放上。這裡就是獲取默認的最大使用占比 就是75% ，接著就是看看commitlog 存儲的那地方使用瞭多少瞭，如果是使用90% 瞭，就設置runningFlag 說磁盤滿瞭，立即清理設置成true，這個參數設置成true之後，就不會管你消息有沒有超過72小時，如果你使用瞭85% 以上瞭，也是設置立即清理，如果超過75% 返回true。好瞭，磁盤占用空間這塊我們就看完瞭。
接著看上面deleteExpiredFiles方法實現，還有一個手動清除的，這塊我沒有找到哪裡有用到的，如果後續找到，會補充上， 判斷 到瞭清理的點 或者是磁盤空間滿瞭 或者是手動刪除瞭，滿足一個條件就ok瞭，如果是立即清除是個true，它這裡這個cleanAtOnce 變量就是true瞭，因為前面那個強制清理是默認開啟的。
接著計算瞭一下fileReservedTime 就是將小時轉成瞭毫秒，為瞭後面好比對，最後就是調用commitlog的deleteExpiredFile 方法清理瞭:
/**
* 刪除過期的文件
* @param expiredTime 過期時間 默認72小時
* @param deleteFilesInterval 刪除文件的間隔 100ms
* @param intervalForcibly 強制刪除 1000 * 120
* @param cleanImmediately 是不是要一次性清理瞭
* @return
*/
public int deleteExpiredFile(
final long expiredTime,
final int deleteFilesInterval,
final long intervalForcibly,
final boolean cleanImmediately
) {
// todo
return this.mappedFileQueue.deleteExpiredFileByTime(expiredTime, deleteFilesInterval, intervalForcibly, cleanImmediately);
}
可以看到commitlog 對象調用mappedFileQueue 的deleteExpiredFileByTime 方法來處理的，這個mappedFileQueue 就是管理瞭一堆MappedFile:
/**
* 刪除文件
*
* 從倒數第二個文件開始遍歷，計算文件的最大存活時間，即文件的最後一次更新時間+文件存活時間（默認
* 72小時），如果當前時間大於文件的最大存活時間或需要強制刪除文
* 件（當磁盤使用超過設定的閾值）時，執行MappedFile#destory方
* 法，清除MappedFile占有的相關資源，如果執行成功，將該文件加入
* 待刪除文件列表中，最後統一執行File#delete方法將文件從物理磁盤
* 中刪除。
*/
public int deleteExpiredFileByTime(final long expiredTime,
final int deleteFilesInterval,
final long intervalForcibly,
final boolean cleanImmediately) {
// 拿到mappedFile的引用
Object[] mfs = this.copyMappedFiles(0);
if (null == mfs)
return 0;
int mfsLength = mfs.length - 1;
int deleteCount = 0;
List<MappedFile> files = new ArrayList<MappedFile>();
if (null != mfs) {
for (int i = 0; i < mfsLength; i++) {
MappedFile mappedFile = (MappedFile) mfs[i];
// 計算文件的最大存活時間，即文件的最後一次更新時間+文件存活時間（默認72小時）
long liveMaxTimestamp = mappedFile.getLastModifiedTimestamp() + expiredTime;
// 如果當前時間大於文件的最大存活時間 或 需要強制刪除文件（當磁盤使用超過設定的閾值）時
if (System.currentTimeMillis() >= liveMaxTimestamp || cleanImmediately) {
// todo 執行destroy方法
if (mappedFile.destroy(intervalForcibly)) {
files.add(mappedFile);
deleteCount++;
// 一批 最多刪除10 個
if (files.size() >= DELETE_FILES_BATCH_MAX) {
break;
}
// 刪除間隔
if (deleteFilesInterval > 0 && (i + 1) < mfsLength) {
try {
Thread.sleep(deleteFilesInterval);
} catch (InterruptedException e) {
}
}
} else {
break;
}
} else {
//avoid deleting files in the middle
break;
}
}
}
// todo 統一執行File#delete方法將文件從物理磁盤中刪除
deleteExpiredFile(files);
return deleteCount;
}
這裡首先是拿到所有MappedFile的引用，然後就是遍歷瞭，可以看到它這個length是-1的，也就是最後一個MappedFile 是遍歷不到的，這個是肯定的，因為最後一個MappedFile肯定是在用著的，如果你來個強制清理，一下清理瞭，就沒法提供服務瞭。
遍歷的時候，拿到對應MappedFile 裡面最後一條消息，看看它的寫入時間是不是已經過瞭這個過期時間瞭，或者直接強制刪除，就會執行MappedFile的銷毀方法，而且帶著銷毀時間：
/**
* 銷毀方法
* @param intervalForcibly 表示拒絕被銷毀的最大存活時間
* @return
*/
public boolean destroy(final long intervalForcibly) {
// todo
this.shutdown(intervalForcibly);
// 清理結束
if (this.isCleanupOver()) {
try {
// 關閉文件通道，
this.fileChannel.close();
log.info("close file channel " + this.fileName + " OK");
long beginTime = System.currentTimeMillis();
// 刪除物理文件
boolean result = this.file.delete();
log.info("delete file[REF:" + this.getRefCount() + "] " + this.fileName
+ (result ? " OK, " : " Failed, ") + "W:" + this.getWrotePosition() + " M:"
+ this.getFlushedPosition() + ", "
+ UtilAll.computeElapsedTimeMilliseconds(beginTime));
} catch (Exception e) {
log.warn("close file channel " + this.fileName + " Failed. ", e);
}
return true;
} else {
log.warn("destroy mapped file[REF:" + this.getRefCount() + "] " + this.fileName
+ " Failed. cleanupOver: " + this.cleanupOver);
}
return false;
}
public void shutdown(final long intervalForcibly) {
// 關閉MappedFile
if (this.available) {
this.available = false;
// 初次關閉的時間戳
this.firstShutdownTimestamp = System.currentTimeMillis();
// todo 嘗試釋放資源
this.release();
} else if (this.getRefCount() > 0) {
if ((System.currentTimeMillis() - this.firstShutdownTimestamp) >= intervalForcibly) {
this.refCount.set(-1000 - this.getRefCount());
this.release();
}
}
}
這裡就不詳細說瞭，其實就是shutdown，然後過瞭120s後強制把引用清瞭，之後就是關閉channel，刪除對應文件。
接著往下說，就是銷毀成功瞭，會記錄刪除數量，判斷刪瞭多少瞭，一批是最多刪10個的，這塊應該是怕影響性能的，你一直刪的的話，這東西很消耗磁盤性能，容易影響其他寫入，讀取功能，如果你銷毀失敗，直接就停瞭。最後就是將刪除的這些MappedFile從MappedFileQueue中刪除掉。再回到commitlog clean service 的run方法:
public void run() {
try {
// todo 刪除過期文件
this.deleteExpiredFiles();
// todo
this.redeleteHangedFile();
} catch (Throwable e) {
DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
}
}
我們deleteExpiredFiles 方法已經介紹完瞭，然後再來看看第二個方法是幹嘛的，這個其實就是判斷第一個MappedFile 還可不可用瞭，如果不可用的話，就刪瞭，這塊有可能是上面 deleteExpiredFiles 方法MappedFile銷毀失敗，然後設置瞭不可用，但是沒有清理掉，所以這塊再來善後下:
private void redeleteHangedFile() {
// redeleteHangedFileInterval間隔 默認1000*120
int interval = DefaultMessageStore.this.getMessageStoreConfig().getRedeleteHangedFileInterval();
// 當前時間戳
long currentTimestamp = System.currentTimeMillis();
if ((currentTimestamp - this.lastRedeleteTimestamp) > interval) {
this.lastRedeleteTimestamp = currentTimestamp;
// 獲取強制銷毀Mapped文件間隔
int destroyMapedFileIntervalForcibly =
DefaultMessageStore.this.getMessageStoreConfig().getDestroyMapedFileIntervalForcibly();
// todo 重新刪除第一個MappedFile
if (DefaultMessageStore.this.commitLog.retryDeleteFirstFile(destroyMapedFileIntervalForcibly)) {
}
}
}
public boolean retryDeleteFirstFile(final long intervalForcibly) {
// 獲取到 第一個mappedFile
MappedFile mappedFile = this.getFirstMappedFile();
if (mappedFile != null) {
// 不可用
if (!mappedFile.isAvailable()) {
log.warn("the mappedFile was destroyed once, but still alive, " + mappedFile.getFileName());
// 銷毀
boolean result = mappedFile.destroy(intervalForcibly);
if (result) {
log.info("the mappedFile re delete OK, " + mappedFile.getFileName());
List<MappedFile> tmpFiles = new ArrayList<MappedFile>();
tmpFiles.add(mappedFile);
this.deleteExpiredFile(tmpFiles);
} else {
log.warn("the mappedFile re delete failed, " + mappedFile.getFileName());
}
return result;
}
}
return false;
}
這塊就是看第一個MappedFile 還可不可用，不可用的話，就銷毀掉。好瞭commitlog 文件清理源碼就解析完成瞭。接下來看下這個ConsumeQueue與indexFile的清理。
2.2 ConsumeQueue 清理
private void cleanFilesPeriodically() {
// todo 清除CommitLog文件
this.cleanCommitLogService.run();
// todo 清除ConsumeQueue文件
this.cleanConsumeQueueService.run();
}
DefaultMessageStore.CleanConsumeQueueService#run:
public void run() {
try {
// 刪除 過期的file
this.deleteExpiredFiles();
} catch (Throwable e) {
DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
}
}
接下來DefaultMessageStore.CleanConsumeQueueService#deleteExpiredFiles：
private void deleteExpiredFiles() {
// 刪除間隔 100
int deleteLogicsFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteConsumeQueueFilesInterval();
// 獲取 commitLog 的最小offset
long minOffset = DefaultMessageStore.this.commitLog.getMinOffset();
if (minOffset > this.lastPhysicalMinOffset) {
// 上次 清理 到哪瞭
this.lastPhysicalMinOffset = minOffset;
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> tables = DefaultMessageStore.this.consumeQueueTable;
// 遍歷刪除
for (ConcurrentMap<Integer, ConsumeQueue> maps : tables.values()) {
for (ConsumeQueue logic : maps.values()) {
// 進行刪除
int deleteCount = logic.deleteExpiredFile(minOffset);
// 間隔
if (deleteCount > 0 && deleteLogicsFilesInterval > 0) {
try {
Thread.sleep(deleteLogicsFilesInterval);
} catch (InterruptedException ignored) {
}
}
}
}
// todo 刪除 過期的 indexFile
DefaultMessageStore.this.indexService.deleteExpiredFile(minOffset);
}
}
首先是獲取刪除間隔，然後拿到commitlog中最小的那個offset ，接著就是判斷上次清理位置與最小offset 比較，如果offset 大於它上次清理的位置的話，就說明 它得把最小offset之前的清理掉。先是記錄最後一次清理的offset是最小offset ， 接著就是遍歷所有的ConsumeQueue ，調用每個ConsumeQueue 的 deleteExpiredFile 方法來清理，我們來看下這個方法：
public int deleteExpiredFile(long offset) {
// 進行銷毀 然後得到銷毀個數
int cnt = this.mappedFileQueue.deleteExpiredFileByOffset(offset, CQ_STORE_UNIT_SIZE);
// 糾正最小偏移量
this.correctMinOffset(offset);
return cnt;
}
CQ_STORE_UNIT_SIZE 這個就是每個unit 占20個字節，見。
/**
* 刪除過期的file
* @param offset 最小offset
* @param unitSize 大小為20字節
* @return
*/
public int deleteExpiredFileByOffset(long offset, int unitSize) {
Object[] mfs = this.copyMappedFiles(0);
List<MappedFile> files = new ArrayList<MappedFile>();
int deleteCount = 0;
if (null != mfs) {
int mfsLength = mfs.length - 1;
for (int i = 0; i < mfsLength; i++) {
boolean destroy;
MappedFile mappedFile = (MappedFile) mfs[i];
// 最後一個單元位置到這個MappedFile結束，其實就是獲取最後一個單元
SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize);
if (result != null) {
// 獲取最大的offset
long maxOffsetInLogicQueue = result.getByteBuffer().getLong();
result.release();
// 判斷是否銷毀 如果小於offset 就要銷毀
destroy = maxOffsetInLogicQueue < offset;
if (destroy) {
log.info("physic min offset " + offset + ", logics in current mappedFile max offset "
+ maxOffsetInLogicQueue + ", delete it");
}
} else if (!mappedFile.isAvailable()) { // Handle hanged file.
log.warn("Found a hanged consume queue file, attempting to delete it.");
destroy = true;
} else {
log.warn("this being not executed forever.");
break;
}
// 進行銷毀
if (destroy && mappedFile.destroy(1000 * 60)) {
files.add(mappedFile);
deleteCount++;
} else {
break;
}
}
}
// 刪除引用
deleteExpiredFile(files);
return deleteCount;
}
它的刪除跟commitlog 的差不多，隻不過commitlog 是根據時間來判斷的，它是根據commitlog 的offset 來判斷的，判斷要不要刪除這個MappedFile，如果這個MappedFile最後一個unit 存儲的offset 小於 commitlog 最小的offset 的話就要銷毀瞭。接著就是銷毀，超時時間是1分鐘，最後是刪除引用。
2.3 indexFile 清理
最後我們來看下 indexFile的清理工作: DefaultMessageStore.CleanConsumeQueueService#deleteExpiredFiles:
private void deleteExpiredFiles() {
// 刪除間隔 100
int deleteLogicsFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteConsumeQueueFilesInterval();
// 獲取 commitLog 的最小offset
long minOffset = DefaultMessageStore.this.commitLog.getMinOffset();
if (minOffset > this.lastPhysicalMinOffset) {
...
// todo 刪除 過期的 indexFile
DefaultMessageStore.this.indexService.deleteExpiredFile(minOffset);
}
}
/**
* 刪除 過期文件
* @param offset 最小的offset 小於這個offset都要刪除
*/
public void deleteExpiredFile(long offset) {
Object[] files = null;
try {
// 獲取讀鎖
this.readWriteLock.readLock().lock();
if (this.indexFileList.isEmpty()) {
return;
}
// 獲取第一個indexFile 的一個offset
long endPhyOffset = this.indexFileList.get(0).getEndPhyOffset();
if (endPhyOffset < offset) {
files = this.indexFileList.toArray();
}
} catch (Exception e) {
log.error("destroy exception", e);
} finally {
this.readWriteLock.readLock().unlock();
}
if (files != null) {
// 找到需要刪除的indexFile
List<IndexFile> fileList = new ArrayList<IndexFile>();
for (int i = 0; i < (files.length - 1); i++) {
IndexFile f = (IndexFile) files[i];
if (f.getEndPhyOffset() < offset) {
fileList.add(f);
} else {
break;
}
}
// 刪除
this.deleteExpiredFile(fileList);
}
}
可以看到，先是拿第一個indexFile 看看有沒有小於commitlog 最小offset 的情況發生，這裡也是拿的indexFile最後一個offset 做的對比，因為這塊也是按照offset大小 前後順序處理的，最後一個的offest 肯定是這個indexFile中最大的瞭，如果第一個indexFile滿足瞭的話，就會拿到所有引用，然後遍歷找出符合條件的indexFile， 調用deleteExpiredFile方法遍歷銷毀:
private void deleteExpiredFile(List<IndexFile> files) {
if (!files.isEmpty()) {
try {
this.readWriteLock.writeLock().lock();
for (IndexFile file : files) {
// 銷毀
boolean destroyed = file.destroy(3000);
// 從index 集合中移除
destroyed = destroyed && this.indexFileList.remove(file);
if (!destroyed) {
log.error("deleteExpiredFile remove failed.");
break;
}
}
} catch (Exception e) {
log.error("deleteExpiredFile has exception.", e);
} finally {
this.readWriteLock.writeLock().unlock();
}
}
}
這裡就是遍歷銷毀，然後移除對這個indexFile管理。
3. 總結
本文主要是介紹瞭RocketMQ broker 消息清理機制，介紹瞭主要清理哪些文件 ：commitlog ，ConsumeQueue，indexFile
接著就是介紹瞭什麼時候觸發清理，比如說凌晨4點 ，磁盤沒滿85% 以上的話，就是清理72小時之前的，如果是滿瞭85%就除瞭還在用著的那個先清10個看看， 還有就是磁盤使用空間75% 以上也是會觸發的， 低於85 % 清理72小時之前的，高於85% 先清理10個文件看看，這是commitlog的清理機制，關於ConsumeQueue與indexFile的話，就是與commitlog中最小的那個offset 有關瞭，小於commitlog中最小offset 的那些還是要清理掉的。 最後就是分別解析瞭一下commitlog 文件清理，ConsumeQueue 文件清理與indexFile 文件清理。
參考
RocketMQ源碼分析專欄
以上就是RocketMQ broker文件清理源碼解析的詳細內容，更多關於RocketMQ broker文件清理的資料請關註WalkonNet其它相關文章！
推薦閱讀：
- RocketMQ消息存儲文件的加載與恢復機制源碼分析
- 一文徹底掌握RocketMQ 的存儲模型
- RocketMQ ConsumeQueue與IndexFile實時更新機制源碼解析
- RocketMQ broker 消息投遞流程處理PULL_MESSAGE請求解析
- RocketMQ特性Broker存儲事務消息實現