RocketMQ 的目標(biāo)，是致力于打造一個(gè)消息、事件、流一體的超融合處理平臺(tái)。這意味著它需要滿足各個(gè)場(chǎng)景下各式各樣的要求，而批量處理則是流計(jì)算領(lǐng)域?qū)τ跇O致吞吐量要求的經(jīng)典解法，這當(dāng)然也意味著 RocketMQ 也有一套屬于自己風(fēng)格的批處理模型。

至于什么樣的批量模型才叫“屬于自己風(fēng)格”呢，且聽(tīng)我娓娓道來(lái)。

什么是批處理

批處理的核心思想是將多個(gè)任務(wù)或數(shù)據(jù)集合在一起，進(jìn)行統(tǒng)一處理。這種方法的優(yōu)勢(shì)在于可以充分利用系統(tǒng)資源，減少任務(wù)切換帶來(lái)的開(kāi)銷，從而提高整體效率。比如在工業(yè)制造中，工廠通常會(huì)將相同類型的零部件批量生產(chǎn)，以降低生產(chǎn)成本和提高生產(chǎn)速度。在互聯(lián)網(wǎng)領(lǐng)域，批處理則表現(xiàn)為批量數(shù)據(jù)的存儲(chǔ)、傳輸和處理，以優(yōu)化性能和提升系統(tǒng)吞吐量。

批處理在極致吞吐量需求下的應(yīng)用，更加顯著。例如，在大數(shù)據(jù)分析中，海量的數(shù)據(jù)需要集中處理才能得出有意義的結(jié)果。如果逐條處理數(shù)據(jù)，不僅效率低下，還可能造成系統(tǒng)瓶頸。通過(guò)批處理，可以將數(shù)據(jù)劃分為若干批次，在預(yù)定的時(shí)間窗口內(nèi)統(tǒng)一處理，從而提高系統(tǒng)的并行處理能力，提升整體吞吐量。

此外，批處理其實(shí)并不意味著犧牲延時(shí)，就比如在 CPU Cache 中，對(duì)單個(gè)字節(jié)的操作無(wú)論如何時(shí)間上都是會(huì)優(yōu)于多個(gè)字節(jié)，但是這樣的比較并沒(méi)有意義，因?yàn)檠訒r(shí)的感知并不是無(wú)窮小的，用戶常常并不關(guān)心 CPU 執(zhí)行一條指令需要花多長(zhǎng)時(shí)間，而是執(zhí)行完單個(gè)“任務(wù)/作業(yè)”需要多久，在宏觀的概念上，反而批處理具有更低的延時(shí)。

RocketMQ 批處理模型演進(jìn)

早期批處理模型

下圖，是作為用戶視角上感知比較強(qiáng)的老三樣，分別是 Producer、Consumer、Broker：

而早期批處理模型，實(shí)際上只和 Producer、Broker 有關(guān)，在這條鏈路上會(huì)有批量消息的概念，當(dāng)消息到達(dá) Broker 后這個(gè)概念就會(huì)消失。

基于這點(diǎn)我們來(lái)看具體是怎么回事。首先批量消息的源頭實(shí)際上就是 Producer 端的 Send 接口，在大部分場(chǎng)景下，我們發(fā)送一條消息都會(huì)使用以下的形式去操作：

SendResult send(Message msg);

非常地簡(jiǎn)明扼要，將一條消息發(fā)送到 Broker，如果我們要使用上早期的批處理模型，也只需要稍作修改：

SendResult send(Collection<Message> msgs)

可以看到，將多條消息串成一個(gè)集合，然后依舊是調(diào)用 send 接口，就可以完成早期批處理模型的使用了（從用戶側(cè)視角看就已經(jīng) ok 了），就像下圖一樣，兩軍交戰(zhàn)，誰(shuí)火力更猛高下立判～

那么真就到此為止了嗎？當(dāng)然不是，首先這里的集合是有講究的，并不是隨意將多條消息放在一起，就可以 send 出去的，它需要滿足一些約束條件：

• 相同 Topic。
• 不能是 RetryTopic。
• 不能是定時(shí)消息。
• 相同 isWaitStoreMsgOK 標(biāo)記。

這些約束條件暫時(shí)先不展開(kāi)，因?yàn)榫腿缤置嬉馑家粯訙\顯易懂，但是這也意味著它的使用并不是隨心所欲的，有一定的學(xué)習(xí)成本，也有一定的開(kāi)發(fā)要求，使用前需要根據(jù)這些約束條件自行分類，然后再裝進(jìn)“大炮”中點(diǎn)火發(fā)射。

這里可能有人會(huì)問(wèn)，這不是為難我胖虎嗎？為什么要加這么多約束？是不是故意的？實(shí)際上并非如此，我們可以想象一下，假如我們是商家：

• 客戶 A 買了兩件物品，在發(fā)貨階段我們很自然的就可以將其打包在一起（將多個(gè) Message 串成一個(gè) ArrayList），然后一次性交給快遞小哥給它 Send 出去，甚至還能省一筆郵費(fèi)呢～
• 客戶 B 和客戶 C 各買了一件物品，此時(shí)我效仿之前的行為打包到一起，然后告訴快遞小哥這里面一個(gè)發(fā)到黑龍江，一個(gè)發(fā)到海南，然后掏出一筆郵費(fèi)，然后。。。就沒(méi)有然后了。

很顯然，第二個(gè)場(chǎng)景很可能會(huì)收到快遞小哥一個(gè)大大的白眼，這種事情理所應(yīng)當(dāng)?shù)淖霾涣?，這也是為什么屬于同一個(gè) Collection<Message> 的消息必須要滿足各種各樣的約束條件了，在 Broker 實(shí)際收到一個(gè)“批量消息”時(shí)，會(huì)做以下處理：

首先它會(huì)根據(jù)這一批消息的某些屬性，挑選出對(duì)應(yīng)的隊(duì)列，也就是上圖中最底下的「p1、p2......」，在選定好隊(duì)列之后，就可以進(jìn)行后續(xù)的寫入等操作了，這也是為什么必須要求相同 Topic，因?yàn)椴煌?Topic 是沒(méi)法選定同一個(gè)隊(duì)列的。

接下來(lái)就到了上圖所示流程，可以看到這里分別來(lái)了三個(gè)消息，分別是 《四條消息》《一條消息》《三條消息》，接下來(lái)他們會(huì)依次進(jìn)入 unPack 流程，這個(gè)流程有點(diǎn)像序列化過(guò)程，因?yàn)閺目蛻舳税l(fā)送上來(lái)的消息都是內(nèi)存結(jié)構(gòu)的，距離實(shí)際存儲(chǔ)在文件系統(tǒng)中的結(jié)構(gòu)還有一些不同。在 unPack 過(guò)程中，會(huì)分別解包成：四條消息、一條消息、三條消息；此時(shí)和連續(xù) Send 八條消息是沒(méi)有任何區(qū)別的，也就是在這一刻，批量消息的生命周期就走到了盡頭，此刻往后，“眾生平等、不分你我”。

也正是這個(gè)機(jī)制，Consumer 其實(shí)并不知道 Producer 發(fā)送的時(shí)候“到底是發(fā)射弓箭，還是點(diǎn)燃大炮”。這么做有個(gè)非常好的優(yōu)點(diǎn)，那就是有著最高的兼容性，一切的一切好像和單條消息 Send 的經(jīng)典用法沒(méi)有任何區(qū)別，在這種情況下，每條消息都有最高的自由度，例如各自獨(dú)立的 tag、獨(dú)立的 keys、唯一的 msgId 等等，而基于這些所衍生出來(lái)的生態(tài)（例如消息軌跡）都是無(wú)縫銜接的。也就是說(shuō)：只需要更換發(fā)送者使用的 Send 接口，就可以獲得極大的發(fā)送性能提升，而消費(fèi)者端無(wú)需任何改動(dòng)。

索引構(gòu)建流水線改造

首先我們要有一個(gè)共識(shí)，那就是對(duì)于消息隊(duì)列這種系統(tǒng)，整體性能上限比值“消費(fèi)/生產(chǎn)”應(yīng)該要滿足至少大于等于一，因?yàn)榇蟛糠智闆r下，我們的生產(chǎn)出來(lái)的消息至少應(yīng)該被消費(fèi)一次（否則直接都不用 Send 了豈不美哉）。

其實(shí)在以往，發(fā)送性能沒(méi)有被拔高之前，它就是整個(gè)生產(chǎn)到消費(fèi)鏈路上的短板，也就是說(shuō)消費(fèi)速率可以輕松超過(guò)生產(chǎn)速率，整個(gè)過(guò)程也就非常協(xié)調(diào)。but！在使用早期批處理模型后，生產(chǎn)速率的大幅度提升就暴露了另外一個(gè)問(wèn)題，也就是會(huì)出現(xiàn)消費(fèi)速率跟不上生產(chǎn)的情況，這種情況下，去談?wù)麄€(gè)系統(tǒng)的性能都是“無(wú)稽之談”。

而出現(xiàn)消費(fèi)速率短板的原因，還要從索引構(gòu)建講起。由于消費(fèi)是要找到具體的消息位置，那就必須依賴于索引，也就是說(shuō)，一條消息的索引構(gòu)建完成之前，是無(wú)法被消費(fèi)到的。下圖就是索引構(gòu)建流程的簡(jiǎn)易圖：

這是整個(gè)直接決定消費(fèi)速率上限的流程。通過(guò)一個(gè)叫 ReputMessageService 的線程，順序掃描 CommitLog 文件，將其分割為一條一條的消息，再對(duì)這些消息進(jìn)行校驗(yàn)等行為，將其轉(zhuǎn)換成一條條的索引信息，并寫入對(duì)應(yīng)分區(qū)的 ConsumeQueue 文件。

整個(gè)過(guò)程是完全串行的，從分割消息，到轉(zhuǎn)換索引，到寫入文件，每一條消息都要經(jīng)過(guò)這么一次流轉(zhuǎn)。因?yàn)橐婚_(kāi)始是串行實(shí)現(xiàn)，所以改造起來(lái)也非常的自然，那就是通過(guò)流水線改造，提高它的并發(fā)度，這里面有幾個(gè)需要解決的問(wèn)題：

• CommitLog 的掃描過(guò)程并行難度高，因?yàn)槊織l消息的長(zhǎng)度是不一致的，無(wú)法簡(jiǎn)單地分割出消息邊界來(lái)分配任務(wù)。
• 單條消息的索引構(gòu)建任務(wù)并不重，因此不能簡(jiǎn)單忽略掉任務(wù)流轉(zhuǎn)過(guò)程中的開(kāi)銷（隊(duì)列入隊(duì)出隊(duì)）。
• 寫入 ConsumeQueue 文件的時(shí)候要求寫入時(shí)機(jī)隊(duì)列維度有序，否則會(huì)帶來(lái)額外的檢查開(kāi)銷等。

針對(duì)這幾個(gè)難點(diǎn)，在設(shè)計(jì)中也引入了“批量處理”的思路，其實(shí)大到架構(gòu)設(shè)計(jì)、小到實(shí)現(xiàn)細(xì)節(jié)，處處都體現(xiàn)了這一理念，下圖就是改造后的流程：

由于 CommitLog 掃描過(guò)程很難并行化處理，那就干脆不做并行化改造了，就使用單線程去順序掃描，但是掃描的時(shí)候會(huì)進(jìn)行一個(gè)簡(jiǎn)單的批處理，掃描出來(lái)的消息并不是單條的，而是盡可能湊齊一個(gè)較大的 buffer 塊，默認(rèn)是 4MB，這個(gè)由多條消息構(gòu)成的 buffer 塊我們不妨將其稱為一個(gè) batch msg。

然后就是對(duì)這些 batch msg 進(jìn)行并行解析，將 batch msg 以單條消息的粒度掃描出來(lái)，并構(gòu)建對(duì)應(yīng)的 DispatchRequest 結(jié)構(gòu)，最終依次落盤到 ConsumeQueue 文件中。其中的關(guān)鍵點(diǎn)在于 batch msg 的順序如何保證，以及 DispatchRequest 在流轉(zhuǎn)時(shí)怎么保證順序和效率。為此我專門實(shí)現(xiàn)了一個(gè)輕量級(jí)的隊(duì)列 DispatchRequestOrderlyQueue，這個(gè) Queue 采用環(huán)狀結(jié)構(gòu)，可以隨著順序標(biāo)號(hào)不斷遞進(jìn)，并且能做到“無(wú)序入隊(duì)，有序出隊(duì)”，詳細(xì)設(shè)計(jì)和實(shí)現(xiàn)均在開(kāi)源 RocketMQ 倉(cāng)庫(kù)中，這里就不多贅述。

在經(jīng)過(guò)改造后，索引構(gòu)建流程不再成為扯后腿的一員，從原本眼中釘?shù)慕巧烂离[身了～

BatchCQ 模型

但是這并不夠！因?yàn)樵缙诘哪Ｐ统鲇诩嫒菪缘瓤紤]，所以依舊束手束腳的，于是 BatchCQ 模型誕生了，主要原因分為兩個(gè)維度：

• 性能上：

• 早期模型中，Broker 端在準(zhǔn)備寫入階段需要進(jìn)行解包，會(huì)有一定的額外開(kāi)銷。
• CommitLog 文件中不具備批量信息，索引需要分多次構(gòu)建。

• 能力上：

• 無(wú)法實(shí)現(xiàn)端到端的批量行為，如加密、壓縮。

那 BatchCQ 又是如何改進(jìn)上述的問(wèn)題的呢？其實(shí)也非常地直觀，那就是“見(jiàn)字如面”，將 ConsumeQueue 也批量化。這個(gè)模型去掉 Broker 端寫入前的解包行為，索引也只進(jìn)行一次構(gòu)建：

圖片

就像上圖所示，如果把索引比做信封，原先每個(gè)信封只能包含一份索引信息，在批量化后則可以塞下任意數(shù)量的索引信息，具體的存儲(chǔ)結(jié)構(gòu)也發(fā)生了較大變化：

圖片

比如說(shuō)如果來(lái)了兩批消息，分別是（3+2）條，在普通的 CQ 模型里會(huì)分別插入 5 個(gè) slot，分別索引到 5 條消息。但是在 BatchCQ 模型中，（3+2）條消息會(huì)只插入 2 個(gè) slot，分別索引到 3 條以及 2 條。

也是因?yàn)檫@個(gè)特點(diǎn)，所以 CQ 原有的格式也發(fā)生了變化，為了記錄更多信息不得不加入 Base Offset、Batch Num 等元素，而這些更改也讓原來(lái)定位索引位置的邏輯發(fā)生了變化。

• 普通 CQ：每個(gè) Slot 定長(zhǎng)，【Slot 長(zhǎng)度 * QueueOffset】位點(diǎn)可以直接找到索引，復(fù)雜度 O(1)。
• BatchCQ：通過(guò)二分法查找，復(fù)雜度 O(log n)。

雖然這部分只涉及到了 ConsumeQueue 的修改，但是它作為核心鏈路的一環(huán)，影響是很大的，首先一批的消息會(huì)被當(dāng)作同一條消息來(lái)處理，不需要重新 unPack ，而且這些消息都會(huì)具有相同的 TAG、Keys 甚至 MessageId，想唯一區(qū)分同一批的消息，只能根據(jù)它們的 QueueOffset 了，這一點(diǎn)會(huì)讓消息軌跡等依靠 MessageId 的能力無(wú)法直接兼容使用，但是消息的處理粒度依然可以保持不變（依賴的是 QueueOffset）。

AutoBatch 模型

這里又要從頭說(shuō)起，在早期批處理模型的總結(jié)里，提到了一個(gè)比較大的缺陷，那就是“使用起來(lái)不夠順手”，用戶是需要關(guān)心各種約束條件的，就像前面提到的 Topic、消息類型、特殊 Flag 等，在 BatchCQ 里面其實(shí)是新增了 Keys、Tag 等維度的限制，錯(cuò)誤使用會(huì)出現(xiàn)一些非預(yù)期的情況。

不難看出，無(wú)論是早期批處理模型、還是 BatchCQ 模型，使用起來(lái)都有一定的學(xué)習(xí)成本，除了需要關(guān)注各種使用方式外，想要用好，還有一些隱藏在暗處的問(wèn)題需要主動(dòng)去解決：

• 無(wú)論是早期的批處理模型，還是 batchCQ 模型，都需要發(fā)送端自行將消息分類打包。
• 消息分類和打包成本高，分類需要關(guān)心分類依據(jù)，打包需要關(guān)心觸發(fā)時(shí)機(jī)。
• 分類依據(jù)復(fù)雜，早期批處理模型需要關(guān)注多個(gè)屬性，batchCQ 在這基礎(chǔ)上新增了多個(gè)限制。
• 打包時(shí)機(jī)不易掌握，使用不當(dāng)容易出現(xiàn)性能下降、時(shí)延不穩(wěn)定、分區(qū)不均衡等問(wèn)題。

為了解決以上問(wèn)題，AutoBatch 應(yīng)運(yùn)而生，它就是一臺(tái)能自動(dòng)分揀的無(wú)情打包機(jī)器，全天候運(yùn)轉(zhuǎn)，精密又高效，將以往需要用戶關(guān)注的細(xì)節(jié)統(tǒng)統(tǒng)屏蔽，它具有以下幾個(gè)優(yōu)點(diǎn)：

• AutoBatch 托管分類和打包能力，只需要簡(jiǎn)單配置即可使用。
• 用戶側(cè)不感知托管的過(guò)程，使用原有發(fā)送接口即可享受批處理帶來(lái)的性能提升，同時(shí)兼容同步發(fā)送和異步發(fā)送。
• AutoBatch 同時(shí)兼容早期的批處理模型和 batchCQ 模型。
• 實(shí)現(xiàn)輕量，性能優(yōu)秀，設(shè)計(jì)上優(yōu)化延時(shí)抖動(dòng)、小分區(qū)等問(wèn)題。

首先到底有多簡(jiǎn)單呢？讓我們來(lái)看一下：

// 發(fā)送端開(kāi)啟 AutoBatch 能力
rmqProducer.setAutoBatch(true);

也就是說(shuō)，只需要加入這么一行，就可以開(kāi)啟 RocketMQ 的性能模式，獲得早期的批處理模型或者 BatchCQ 模型帶來(lái)的極致吞吐量提升。在開(kāi)啟 AutoBatch 的開(kāi)關(guān)后，用戶所有已有的行為都不需要作出改變，使用原來(lái)經(jīng)典的 Send（Message msg）即可；當(dāng)然也可以進(jìn)行更精細(xì)的內(nèi)存控制和延時(shí)控制：

// 設(shè)置單個(gè) MessageBatch 大小（kb）
rmqProducer.batchMaxBytes(32 * 1024);


// 設(shè)置最大聚合等待時(shí)間（ms）
rmqProducer.batchMaxDelayMs(10);


// 設(shè)置所有聚合器最大內(nèi)存使用（kb）
rmqProducer.totalBatchMaxBytes(32 * 1024 * 1024);

圖片

那么它具體輕量在哪？又高效在哪？下面這個(gè)簡(jiǎn)易的流程圖應(yīng)該能給大家一個(gè)答案：

首先它只引入了一個(gè)單線程的背景線程——background thread，這個(gè)背景線程以 1/2 的 maxDelayMs 周期運(yùn)行，將掃描到超過(guò)等待時(shí)機(jī)緩沖區(qū)的消息提交到異步發(fā)送的線程池中，此時(shí)就完成了時(shí)間維度的聚合?？臻g維度的聚合則是由發(fā)送線程在傳遞時(shí)進(jìn)行檢查，如果滿足 maxBytes，則原地發(fā)送。

整個(gè)設(shè)計(jì)非常地精簡(jiǎn)，只額外引入了一個(gè)周期運(yùn)行的線程，這樣做可以避免因?yàn)?AutoBatch 模型本身出現(xiàn)性能短板，而且 batchMessage 的序列化過(guò)程也做了精簡(jiǎn)，去掉了發(fā)送時(shí)候所有的檢測(cè)（在聚合過(guò)程中已提前分類）。

才藝展示

準(zhǔn)備工作

圖片

為 Openmessaging-Benchmark 進(jìn)行壓測(cè)環(huán)境，首先部署一套開(kāi)源社區(qū)上最新的 RocketMQ，然后配置好 Namesrv 接入點(diǎn)等信息，然后打開(kāi) RocketMQ 的性能模式——AutoBatch，將 autoBatch 字段設(shè)置為 true：

早期批處理模型

bin/benchmark --drivers driver-rocketmq/rocketmq.yaml workloads/1-topic-100-partitions-1kb-4p-4c-1000k.yaml

開(kāi)啟 autobatch 能力后，就會(huì)使用早期批處理模型進(jìn)行性能提升，可以看到提升幅度非常大，由原來(lái)的 8w 提升至 27w 附近，為原來(lái)的 300%。

索引構(gòu)建流水線優(yōu)化

流水線優(yōu)化是需要在服務(wù)端開(kāi)啟的，下面是一個(gè)簡(jiǎn)單的配置例子：

// 開(kāi)啟索引構(gòu)建流水線優(yōu)化
enableBuildConsumeQueueCnotallow=true


// 調(diào)整內(nèi)存中消息最大消費(fèi)閾值
maxTransferBytesOnMessageInMemory=256M
maxTransferCountOnMessageInMemory=32K


// 調(diào)整磁盤中消息最大消費(fèi)閾值
maxTransferBytesOnMessageInDisk=64M
maxTransferCountOnMessageInDisk=32K

可以看到，只有開(kāi)啟索引構(gòu)建優(yōu)化，才能做到穩(wěn)穩(wěn)地達(dá)到 27w 的吞吐，在沒(méi)有開(kāi)啟的時(shí)候，消費(fèi)速率不足會(huì)觸發(fā)冷讀直至影響到整個(gè)系統(tǒng)的穩(wěn)定性，同時(shí)也不具備生產(chǎn)意義，所以在使用批量模型的時(shí)候也務(wù)必需要開(kāi)啟索引構(gòu)建優(yōu)化。

BatchCQ 模型

// Topic 的各種屬性在 TopicAttributes 中設(shè)置


public static final EnumAttribute QUEUE_TYPE_ATTRIBUTE = new EnumAttribute("queue.type", false, newHashSet("BatchCQ", "SimpleCQ"), "SimpleCQ");
topicConfig.getAttributes().put("+" + TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName(), "BatchCQ");

當(dāng)使用 BatchCQ 模型的時(shí)候，與早期批處理模型已經(jīng)有了天壤之別，因此我們尋求了和開(kāi)源 Kafka 的對(duì)比，部署架構(gòu)如下：

RocketMQ

3 主 3 備架構(gòu)，使用輕量級(jí) Container 部署。

• 節(jié)點(diǎn) 1: Master-A，Slave-C
• 節(jié)點(diǎn) 2: Master-C，Slave-B
• 節(jié)點(diǎn) 3: Master-B，Slave-A

Kafka

3 個(gè)節(jié)點(diǎn)，設(shè)置分區(qū)副本數(shù)為 2。

壓測(cè)結(jié)果

可以看到，在使用 BatchCQ 類型的 Topic 時(shí)，RocketMQ 與 Kafka 的性能基本持平：

• 16-partitions，二者吞吐量相差 5% 以內(nèi)，且 RocketMQ 則具有明顯更低的延時(shí)表現(xiàn)。
• 10000-partitions，得益于 RocketMQ 的存儲(chǔ)結(jié)構(gòu)更為集中，在大量分區(qū)場(chǎng)景下吞吐量幾乎保持不變。而Kafka在默認(rèn)配置的情況下出現(xiàn)報(bào)錯(cuò)無(wú)法使用。

因此在極致吞吐量的需求下，BatchCQ 模型能夠很好地承接極致需求的流量，而且如果更換性能更好的本地磁盤，同樣的機(jī)器配置能達(dá)到更高的上限。

相關(guān)鏈接：

[1] ecs.c7.8xlarge

https://help.aliyun.com/zh/ecs/user-guide/overview-of-instance-families