1000w云上開發(fā)者 全棧云產(chǎn)品0元試用：點擊

作者：張森澤

(相關(guān)資料圖)

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

設(shè)計總覽

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

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

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

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

Quick Start

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

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

可選項：支持修改 tieredMetadataServiceProvider 切換元數(shù)據(jù)存儲的實現(xiàn)，默認(rèn)是基于 json 的文件存儲

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

技術(shù)架構(gòu)

architecture

接入層 ：TieredMessageStore/TieredDispatcher/TieredMessageFetcher

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

容器層 ：TieredCommitLog/TieredConsumeQueue/TieredIndexFile/TieredFileQueue

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

驅(qū)動層 ：TieredFileSegment

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

消息上傳

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

TieredDispatcher

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

upload buffer

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

上傳進度控制

每個隊列都會有兩個關(guān)鍵位點控制上傳進度：

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

upload progress

類比消費者，dispatch offset 相當(dāng)于拉取消息的位點，commit offset 相當(dāng)于確認(rèn)消費的位點。commit offset 到 dispatch offset 之間的部分相當(dāng)于已拉取未消費的消息。

消息讀取

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

DISABLE：禁止從多級存儲中讀取消息； NOT_IN_DISK：不在 DefaultMessageStore 中的消息從多級存儲中讀??； NOT_IN_MEM：不在 page cache 中的消息即冷數(shù)據(jù)從多級存儲讀??； 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 處理：首先校驗參數(shù)是否合法，然后按照邏輯位點（queue offset）發(fā)起拉取請求。TieredConsumeQueue/TieredCommitLog 將邏輯位點換算為對應(yīng)文件的物理位點從 TieredFileSegment 讀取消息。

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

TieredFileSegment 維護每個儲存在文件系統(tǒng)中的物理文件位點，并通過為不同存儲介質(zhì)實現(xiàn)的接口從中讀取所需的數(shù)據(jù)。

/**  * 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);

預(yù)讀緩存

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

protected final CachereadAheadCache;

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

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

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

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

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

故障恢復(fù)

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

開發(fā)計劃

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

tag 過濾

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

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

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

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

和高可用架構(gòu)的融合

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

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

相關(guān)鏈接：

[1] README

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

版權(quán)聲明：本文內(nèi)容由阿里云實名注冊用戶自發(fā)貢獻(xiàn)，版權(quán)歸原作者所有，阿里云開發(fā)者社區(qū)不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。具體規(guī)則請查看《阿里云開發(fā)者社區(qū)用戶服務(wù)協(xié)議》和《阿里云開發(fā)者社區(qū)知識產(chǎn)權(quán)保護指引》。如果您發(fā)現(xiàn)本社區(qū)中有涉嫌抄襲的內(nèi)容，填寫侵權(quán)投訴表單進行舉報，一經(jīng)查實，本社區(qū)將立刻刪除涉嫌侵權(quán)內(nèi)容。

關(guān)鍵詞：