1000w云上開發者 全棧云產品0元試用：點擊

作者：張森澤

(相關資料圖)

隨著 RocketMQ 5.1.0 的正式發布，多級存儲作為 RocketMQ 一個新的獨立模塊到達了 Technical Preview 里程碑：允許用戶將消息從本地磁盤卸載到其他更便宜的存儲介質，可以用較低的成本延長消息保留時間。本文詳細介紹 RocketMQ 多級存儲設計與實現。

設計總覽

RocketMQ 多級存儲旨在 不影響熱數據讀寫的前提下 將數據卸載到其他存儲介質中，適用于兩種場景：

冷熱數據分離：RocketMQ 新近產生的消息會緩存在 page cache 中，我們稱之為 熱數據 ；當緩存超過了內存的容量就會有熱數據被換出成為 冷數據 。如果有少許消費者嘗試消費冷數據就會從硬盤中重新加載冷數據到 page cache，這會導致讀寫 IO 競爭并擠壓 page cache 的空間。而將冷數據的讀取鏈路切換為多級存儲就可以避免這個問題； 延長消息保留時間：將消息卸載到更大更便宜的存儲介質中，可以用較低的成本實現更長的消息保存時間。同時多級存儲支持為 topic 指定不同的消息保留時間，可以根據業務需要靈活配置消息 TTL。

RocketMQ 多級存儲對比 Kafka 和 Pulsar 的實現最大的不同是我們使用準實時的方式上傳消息，而不是等一個 CommitLog 寫滿后再上傳，主要基于以下幾點考慮：

均攤成本：RocketMQ 多級存儲需要將全局 CommitLog 轉換為 topic 維度并重新構建消息索引，一次性處理整個 CommitLog 文件會帶來性能毛刺； 對小規格實例更友好：小規格實例往往配置較小的內存，這意味著熱數據會更快換出成為冷數據，等待 CommitLog 寫滿再上傳本身就有冷讀風險。采取準實時上傳的方式既能規避消息上傳時的冷讀風險，又能盡快使得冷數據可以從多級存儲讀取。

Quick Start

多級存儲在設計上希望降低用戶心智負擔：用戶無需變更客戶端就能實現無感切換冷熱數據讀寫鏈路，通過簡單的修改服務端配置即可具備多級存儲的能力，只需以下兩步：

修改 Broker 配置，指定使用 org.apache.rocketmq.tieredstore.TieredMessageStore 作為 messageStorePlugIn 配置你想使用的儲存介質，以卸載消息到其他硬盤為例：配置 tieredBackendServiceProvider 為 org.apache.rocketmq.tieredstore.provider.posix.PosixFileSegment，同時指定新儲存的文件路徑：tieredStoreFilepath

可選項：支持修改 tieredMetadataServiceProvider 切換元數據存儲的實現，默認是基于 json 的文件存儲

更多使用說明和配置項可以在 GitHub 上查看多級存儲的 README[1]

技術架構

architecture

接入層 ：TieredMessageStore/TieredDispatcher/TieredMessageFetcher

接入層實現 MessageStore 中的部分讀寫接口，并為他們增加了異步語意。TieredDispatcher 和 TieredMessageFetcher 分別實現了多級存儲的上傳/下載邏輯，相比于底層接口這里做了較多的性能優化：包括使用獨立的線程池，避免慢 IO 阻塞訪問熱數據；使用預讀緩存優化性能等。

容器層 ：TieredCommitLog/TieredConsumeQueue/TieredIndexFile/TieredFileQueue

容器層實現了和 DefaultMessageStore 類似的邏輯文件抽象，同樣將文件劃分為 CommitLog、ConsumeQueue、IndexFile，并且每種邏輯文件類型都通過 FileQueue 持有底層物理文件的引用。有所不同的是多級存儲的 CommitLog 改為 queue 維度。

驅動層 ：TieredFileSegment

驅動層負責維護邏輯文件到物理文件的映射，通過實現 TieredStoreProvider 對接底層文件系統讀寫接口(Posix、S3、OSS、MinIO 等)。目前提供了 PosixFileSegment 的實現，可以將數據轉移到其他硬盤或通過 fuse 掛載的對象存儲上。

消息上傳

RocketMQ 多級存儲的消息上傳是由 dispatch 機制觸發的：初始化多級存儲時會將 TieredDispatcher 注冊為 CommitLog 的 dispacher。這樣每當有消息發送到 Broker 會調用 TieredDispatcher 進行消息分發，TieredDispatcher 將該消息寫入到 upload buffer 后立即返回成功。整個 dispatch 流程中不會有任何阻塞邏輯，確保不會影響本地 ConsumeQueue 的構建。

TieredDispatcher

TieredDispatcher 寫入 upload buffer 的內容僅為消息的引用，不會將消息的 body 讀入內存。因為多級儲存以 queue 維度構建 CommitLog，此時需要重新生成 commitLog offset 字段。

upload buffer

觸發 upload buffer 上傳時讀取到每條消息的 commitLog offset 字段時采用拼接的方式將新的 offset 嵌入到原消息中。

上傳進度控制

每個隊列都會有兩個關鍵位點控制上傳進度：

dispatch offset：已經寫入緩存但是未上傳的消息位點 commit offset：已上傳的消息位點

upload progress

類比消費者，dispatch offset 相當于拉取消息的位點，commit offset 相當于確認消費的位點。commit offset 到 dispatch offset 之間的部分相當于已拉取未消費的消息。

消息讀取

TieredMessageStore 實現了 MessageStore 中的消息讀取相關接口，通過請求中的邏輯位點（queue offset）判斷是否從多級存儲中讀取消息，根據配置（tieredStorageLevel）有四種策略：

DISABLE：禁止從多級存儲中讀取消息； NOT_IN_DISK：不在 DefaultMessageStore 中的消息從多級存儲中讀取； NOT_IN_MEM：不在 page cache 中的消息即冷數據從多級存儲讀取； FORCE：強制所有消息從多級存儲中讀取，目前僅供測試使用。

/**  * Asynchronous get message  * @see #getMessage(String, String, int, long, int, MessageFilter)   getMessage  *  * @param group Consumer group that launches this query.  * @param topic Topic to query.  * @param queueId Queue ID to query.  * @param offset Logical offset to start from.  * @param maxMsgNums Maximum count of messages to query.  * @param messageFilter Message filter used to screen desired   messages.  * @return Matched messages.  */CompletableFuturegetMessageAsync(final String group, final String topic, final int queueId,    final long offset, final int maxMsgNums, final MessageFilter messageFilter);

需要從多級存儲中讀取的消息會交由 TieredMessageFetcher 處理：首先校驗參數是否合法，然后按照邏輯位點（queue offset）發起拉取請求。TieredConsumeQueue/TieredCommitLog 將邏輯位點換算為對應文件的物理位點從 TieredFileSegment 讀取消息。

// TieredMessageFetcher#getMessageAsync similar with TieredMessageStore#getMessageAsyncpublic CompletableFuturegetMessageAsync(String group, String topic, int queueId,        long queueOffset, int maxMsgNums, final MessageFilter messageFilter)

TieredFileSegment 維護每個儲存在文件系統中的物理文件位點，并通過為不同存儲介質實現的接口從中讀取所需的數據。

/**  * Get data from backend file system  *  * @param position the index from where the file will be read  * @param length the data size will be read  * @return data to be read  */CompletableFutureread0(long position, int length);

預讀緩存

TieredMessageFetcher 讀取消息時會預讀一部分消息供下次使用，這些消息暫存在預讀緩存中。

protected final CachereadAheadCache;

預讀緩存的設計參考了 TCP Tahoe 擁塞控制算法，每次預讀的消息量類似擁塞窗口采用加法增、乘法減的機制控制：

加法增：從最小窗口開始，每次增加等同于客戶端 batchSize 的消息量。 乘法減：當緩存的消息超過了緩存過期時間仍未被全部拉取，在清理緩存的同時會將下次預讀消息量減半。

預讀緩存支持在讀取消息量較大時分片并發請求，以取得更大帶寬和更小的延遲。

某個 topic 消息的預讀緩存由消費這個 topic 的所有 group 共享，緩存失效策略為：

所有訂閱這個 topic 的 group 都訪問了緩存 到達緩存過期時間

故障恢復

上文中我們介紹上傳進度由 commit offset 和 dispatch offset 控制。多級存儲會為每個 topic、queue、fileSegment 創建元數據并持久化這兩種位點。當 Broker 重啟后會從元數據中恢復，繼續從 commit offset 開始上傳消息，之前緩存的消息會重新上傳并不會丟失。

開發計劃

面向云原生的存儲系統要最大化利用云上存儲的價值，而對象存儲正是云計算紅利的體現。RocketMQ 多級存儲希望一方面利用對象存儲低成本的優勢延長消息存儲時間、拓展數據的價值；另一方面利用其共享存儲的特性在多副本架構中兼得成本和數據可靠性，以及未來向 Serverless 架構演進。

tag 過濾

多級存儲拉取消息時沒有計算消息的 tag 是否匹配，tag 過濾交給客戶端處理。這樣會帶來額外的網絡開銷，計劃后續在服務端增加 tag 過濾能力。

廣播消費以及多個消費進度不同的消費者

預讀緩存失效需要所有訂閱這個 topic 的 group 都訪問了緩存，這在多個 group 消費進度不一致的情況下很難觸發，導致無用的消息在緩存中堆積。

需要計算出每個 group 的消費 qps 來估算某個 group 能否在緩存失效前用上緩存的消息。如果緩存的消息預期在失效前都不會被再次訪問，那么它應該被立即過期。相應的對于廣播消費，消息的過期策略應被優化為所有 Client 都讀取這條消息后才失效。

和高可用架構的融合

目前主要面臨以下三個問題：

元數據同步：如何可靠的在多個節點間同步元數據，slave 晉升時如何校準和補全缺失的元數據； 禁止上傳超過 confirm offset 的消息：為了避免消息回退，上傳的最大 offset 不能超過 confirm offset； slave 晉升時快速啟動多級存儲：只有 master 節點具有寫權限，在 slave 節點晉升后需要快速拉起多級存儲斷點續傳。

相關鏈接：

[1] README

https://github.com/apache/rocketmq/blob/develop/tieredstore/README.md

版權聲明：本文內容由阿里云實名注冊用戶自發貢獻，版權歸原作者所有，阿里云開發者社區不擁有其著作權，亦不承擔相應法律責任。具體規則請查看《阿里云開發者社區用戶服務協議》和《阿里云開發者社區知識產權保護指引》。如果您發現本社區中有涉嫌抄襲的內容，填寫侵權投訴表單進行舉報，一經查實，本社區將立刻刪除涉嫌侵權內容。

關鍵詞：