原創(chuàng)不易，轉(zhuǎn)載請注明出處

文章目錄

• • • 1. broker 清理文件介紹
    • • 1.1 哪些文件需要清理
      • 1.2 RocketMQ文件清理的機制
    • 2.源碼解析
    • • 2.1 清理commitlog
      • 2.2 ConsumeQueue 與indexFile 清理
    • 總結(jié)

---

1. broker 清理文件介紹

1.1 哪些文件需要清理

首先我們需要介紹下在RocketMQ中哪些文件需要清理，其實可以想一想，在RocketMQ中哪些文件是一直在往里面寫入東西的，最容易想到的就是commitlog 了，因為在一個broker 進程中，所有的普通消息，事務(wù)消息，系統(tǒng)消息啥的都往這個commitlog中寫，隨著時間的越來越長，然后commitlog就會越積攢越多，肯定會有磁盤放不下的那一天，而且我們消息消費完成后，那些被消費完成后的消息其實作用就很小了，可能會有這么一個場景，比如說我線上出現(xiàn)了某個問題，我想看下關(guān)于這個問題的消息有沒有被消費到，可能你會用到這個消息，但是這種問題一般就是比較緊急的，最近實效的，之前那些消息其實作用就基本沒有了，所以就需要清理掉之前的消息。其實不光commitlog需要清理，還需要清理一下ConsumeQueue 與indexFile ， 因為你commitlog里面的消息都被清理了，ConsumeQueue 與indexFile 再保存著之前的一些數(shù)據(jù)，就是純粹浪費空間了。
所以說 broker 文件清理主要是清理commitlog ， ConsumeQueue ， indexFile

1.2 RocketMQ文件清理的機制

我們介紹下RocketMQ文件清理的機制，RocketMQ默認(rèn)是清理72小時之前的消息，然后它有幾個觸發(fā)條件， 默認(rèn)是凌晨4點觸發(fā)清理， 除非你你這個磁盤空間占用到75% 以上了。在清理commitlog 的時候，并不是一條消息一條消息的清理，拿到所有的MappedFile（拋去現(xiàn)在還在用著的，也就是最后一個） ，然后比對每個MappedFile的最后一條消息的時間，如果是72小時之前的就把MappedFile對應(yīng)的文件刪除了，銷毀對應(yīng)MappedFile，這種情況的話只要你MappedFile 最后一條消息還在存活實效內(nèi)的話，它就不會清理你這個MappedFile，就算你這個MappedFile 靠前的消息過期了。但是有一種情況它不管你消息超沒超過72小時，直接就是刪，那就是磁盤空間不足的時候，也就是占了85%以上了，就會立即清理。
清理完成commitlog 之后，就會拿到commitlog中最小的offset ，然后去ConsumeQueue 與indexFile 中把小于offset 的記錄刪除掉。清理ConsumeQueue 的時候也是遍歷MappedFile ，然后它的最后一條消息（unit）小于commitlog中最小的offset 的話，就說明這個MappedFile都小于offset ，因為他們是順序追加寫的，這個MappedFile 就會清理掉，如果你MappedFile 最后一個unit不是小于offset 的話，這個MappedFile 就不刪了。

2.源碼解析

我們來看下源碼是怎樣實現(xiàn)的：
在broker 存儲器DefaultMessageStore 啟動（start）的時候，會添加幾個任務(wù)調(diào)度，其中有一個就是文件清理的

默認(rèn)是10s執(zhí)行一次，可以看到它調(diào)用了DefaultMessageStore 的cleanFilesPeriodically方法

2.1 清理commitlog

我們先來看下關(guān)于commitlog的清理工作

我們看下deleteExpiredFiles 方法的實現(xiàn)

開始幾個參數(shù)，一個是文件保留實效默認(rèn)是72小時，你可以使用fileReservedTime來配置，一個是刪除文件的間隔100ms，再就是強行銷毀MappedFile的120s（這個為啥要強行銷毀，因為它還害怕還有地方用著這個MappedFile，它有個專門的引用計數(shù)器，比如說我還有地方要讀它的消息，這個時候計數(shù)器就是+1的），接著就是判斷到?jīng)]到刪除的那個時間，它默認(rèn)是凌晨4點才能刪除

再接著就是看看空間是不是充足，看看磁盤空間使用占比是什么樣子的

這里其實不光是判斷 commitlog的存儲區(qū)域，后面還有段判斷ConsumeQueue的存儲區(qū)域的，然后與這塊邏輯一樣，就沒有放上。這里就是獲取默認(rèn)的最大使用占比 就是75% ，接著就是看看commitlog 存儲的那地方使用了多少了，如果是使用90% 了，就設(shè)置runningFlag 說磁盤滿了，立即清理設(shè)置成true，這個參數(shù)設(shè)置成true之后，就不會管你消息有沒有超過72小時，如果你使用了85% 以上了，也是設(shè)置立即清理，如果超過75 % 返回true。好了，磁盤占用空間這塊我們就看完了，接著看上面deleteExpiredFiles方法實現(xiàn)，還有一個手動清除的，這塊我沒有找到哪里有用到的，如果后續(xù)找到，會補充上， 判斷 到了清理的點 或者是磁盤空間滿了 或者是手動刪除了，滿足一個條件就ok了，如果是立即清除是個true，它這里這個cleanAtOnce 變量就是true了，因為前面那個強制清理是默認(rèn)開啟的。
接著計算了一下fileReservedTime 就是將小時轉(zhuǎn)成了毫秒，為了后面好比對，最后就是調(diào)用commitlog的deleteExpiredFile 方法清理了

可以看到commitlog 對象調(diào)用mappedFileQueue 的deleteExpiredFileByTime 方法來處理的，這個mappedFileQueue 就是管理了一堆MappedFile

這里首先是拿到所有MappedFile的引用，然后就是遍歷了，可以看到它這個length是-1的，也就是最后一個MappedFile 是遍歷不到的，這個是肯定的，因為最后一個MappedFile肯定是在用著的，如果你來個強制清理，一下清理了，就沒法提供服務(wù)了。
遍歷的時候，拿到對應(yīng)MappedFile 里面最后一條消息，看看它的寫入時間是不是已經(jīng)過了這個過期時間了，或者直接強制刪除，就會執(zhí)行MappedFile的銷毀方法，而且?guī)еN毀時間


這里就不詳細(xì)說了，其實就是shutdown，然后過了120s后強制把引用清了，之后就是關(guān)閉channel，刪除對應(yīng)文件。
接著往下說，就是銷毀成功了，會記錄刪除數(shù)量，判斷刪了多少了，一批是最多刪10個的，這塊應(yīng)該是怕影響性能的，你一直刪的的話，這東西很消耗磁盤性能，容易影響其他寫入，讀取功能，如果你銷毀失敗，直接就停了。最后就是將刪除的這寫MappedFile從MappedFileQueue中刪除掉。再回到commitlog clean service 的run方法

我們deleteExpiredFiles 方法已經(jīng)介紹完了，然后再來看看第二個方法是干嘛的，這個其實就是判斷第一個MappedFile 還可不可用了，如果不可用的話，就刪了，這塊有可能是上面 deleteExpiredFiles 方法MappedFile銷毀失敗，然后設(shè)置了不可用，但是沒有清理掉，所以這塊再來善后下


這塊就是看第一個MappedFile 還可不可用，不可用的話，就銷毀掉。好了commitlog 文件清理源碼就解析完成了。接下來看下這個ConsumeQueue與indexFile的清理

2.2 ConsumeQueue 與indexFile 清理

這個藍色框里面是關(guān)于ConsumeQueue刪除的，首先是獲取刪除間隔，然后拿到commitlog中最小的那個offset ，接著就是判斷上次清理位置與最小offset 比較，如果offset 大于它上次清理的位置的話，就說明 它得把最小offset之前的清理掉。先是記錄最后一次清理的offset是最小offset ， 接著就是遍歷所有的ConsumeQueue ，調(diào)用每個ConsumeQueue 的 deleteExpiredFile 方法來清理，我們來看下這個方法

CQ_STORE_UNIT_SIZE 這個就是每個unit 占20個字節(jié)，這塊與它存儲有關(guān)，推薦看下《RocketMQ源碼解析之broker消息存儲流程（Reput ConsumeQueue）》來詳細(xì)了解下它是怎么存的。

它的刪除跟commitlog 的差不多，只不過commitlog 是根據(jù)時間來判斷的，它是根據(jù)commitlog 的offset 來判斷的，判斷要不要刪除這個MappedFile，如果這個MappedFile最后一個unit 存儲的offset 小于 commitlog 最小的offset 的話就要銷毀了。接著就是銷毀，超時時間是1分鐘，最后是刪除引用。
最后我們來看下 indexFile的清理工作


可以看到，先是拿第一個indexFile 看看有沒有小于commitlog 最小offset 的情況發(fā)生，這里也是拿的indexFile最后一個offset 做的對比，因為這塊也是按照offset大小 前后順序處理的，最后一個的offest 肯定是這個indexFile中最大的了，如果第一個indexFile滿足了的話，就會拿到所有引用，然后遍歷找出符合條件的indexFile， 調(diào)用deleteExpiredFile方法遍歷銷毀

這里就是遍歷銷毀，然后移除對這個indexFile管理。好了，到這我們consumeQueue 與indexFile的清理工作就完事了。

總結(jié)

本文主要是介紹了RocketMQ broker 消息清理機制，介紹了主要清理哪些文件 ：commitlog ，ConsumeQueue，indexFile
接著就是介紹了什么時候觸發(fā)清理，比如說凌晨4點 ，磁盤沒滿85% 以上的話，就是清理72小時之前的，如果是滿了85%就除了還在用著的那個先清10個看看， 還有就是磁盤使用空間75% 以上也是會觸發(fā)的， 低于85 % 清理72小時之前的，高于85% 先清理10個文件看看，這是commitlog的清理機制，關(guān)于ConsumeQueue與indexFile的話，就是與commitlog中最小的那個offset 有關(guān)了，小于commitlog中最小offset 的那些還是要清理掉的。
最后就是分別解析了一下commitlog 文件清理，ConsumeQueue 文件清理與indexFile 文件清理。

總結(jié)

以上是生活随笔為你收集整理的RocketMQ源码解析之broker文件清理的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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