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

至于什么樣的批量模型才叫“屬于自己風格”呢，且聽我娓娓道來。

什么是批處理

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

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

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

RocketMQ 批處理模型演進

早期批處理模型

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

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

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

SendResult send(Message msg);

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

SendResult send(Collection<Message> msgs)

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

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

• 相同 Topic。
• 不能是 RetryTopic。
• 不能是定時消息。
• 相同 isWaitStoreMsgOK 標記。

這些約束條件暫時先不展開，因為就如同它字面意思一樣淺顯易懂，但是這也意味著它的使用并不是隨心所欲的，有一定的學習成本，也有一定的開發(fā)要求，使用前需要根據(jù)這些約束條件自行分類，然后再裝進“大炮”中點火發(fā)射。

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

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

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

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

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

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

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

首先我們要有一個共識，那就是對于消息隊列這種系統(tǒng)，整體性能上限比值“消費/生產(chǎn)”應該要滿足至少大于等于一，因為大部分情況下，我們的生產(chǎn)出來的消息至少應該被消費一次（否則直接都不用 Send 了豈不美哉）。

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

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

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

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

• CommitLog 的掃描過程并行難度高，因為每條消息的長度是不一致的，無法簡單地分割出消息邊界來分配任務。
• 單條消息的索引構(gòu)建任務并不重，因此不能簡單忽略掉任務流轉(zhuǎn)過程中的開銷（隊列入隊出隊）。
• 寫入 ConsumeQueue 文件的時候要求寫入時機隊列維度有序，否則會帶來額外的檢查開銷等。

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

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

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

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

BatchCQ 模型

但是這并不夠！因為早期的模型出于兼容性等考慮，所以依舊束手束腳的，于是 BatchCQ 模型誕生了，主要原因分為兩個維度：

• 性能上：

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

• 能力上：

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

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

圖片

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

圖片

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

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

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

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

AutoBatch 模型

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

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

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

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

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

首先到底有多簡單呢？讓我們來看一下：

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

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

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


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


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

圖片

那么它具體輕量在哪？又高效在哪？下面這個簡易的流程圖應該能給大家一個答案：

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

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

才藝展示

準備工作

圖片

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

早期批處理模型

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

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

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

流水線優(yōu)化是需要在服務端開啟的，下面是一個簡單的配置例子：

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


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


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

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

BatchCQ 模型

// Topic 的各種屬性在 TopicAttributes 中設置


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");

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

RocketMQ

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

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

Kafka

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

壓測結(jié)果

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

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

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

相關(guān)鏈接：

[1] ecs.c7.8xlarge

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