大家好，我是三友~~

RocketMQ作為阿里開源的消息中間件，深受廣大開發者的喜愛

而這其中一個很重要原因就是，它處理消息和拉取消息的速度非常快

那么，問題來了，RocketMQ為什么這么快呢？

接下來，我將從以下10個方面來探討一下RocketMQ這么快的背后原因

圖片

如果你對RocketMQ還不了解，可以從公眾號后臺菜單欄中查看我之前寫的關于RocketMQ的幾篇文章

如果你對RocketMQ源碼也感興趣，可以從下面這個倉庫fork一下源碼，我在源碼中加了中文注釋，并且后面我還會持續更新注釋

https://github.com/sanyou3/rocketmq.git

本文是基于RocketMQ 4.9.x版本講解

批量發送消息

RocketMQ在發送消息的時候支持一次性批量發送多條消息，如下代碼所示：

public class Producer {
    public static void main(String[] args) throws Exception {
        //創建一個生產者，指定生產者組為 sanyouProducer
        DefaultMQProducer producer = new DefaultMQProducer("sanyouProducer");
        // 指定NameServer的地址
        producer.setNamesrvAddr("192.168.200.143:9876");
        // 啟動生產者
        producer.start();

        //用以及集合保存多個消息
        List<Message> messages = new ArrayList<>();
        messages.add(new Message("sanyouTopic", "三友的java日記 0".getBytes()));
        messages.add(new Message("sanyouTopic", "三友的java日記 1".getBytes()));
        messages.add(new Message("sanyouTopic", "三友的java日記 2".getBytes()));
        // 發送消息并得到消息的發送結果，然后打印
        SendResult sendResult = producer.send(messages);
        System.out.printf("%s%n", sendResult);

        // 關閉生產者
        producer.shutdown();
    }

}

通過批量發送消息，減少了RocketMQ客戶端與服務端，也就是Broker之間的網絡通信次數，提高傳輸效率

不過在使用批量消息的時候，需要注意以下三點：

• 每條消息的Topic必須都得是一樣的
• 不支持延遲消息和事務消息
• 不論是普通消息還是批量消息，總大小默認不能超過4m

消息壓縮

RocketMQ在發送消息的時候，當發現消息的大小超過4k的時候，就會對消息進行壓縮

這是因為如果消息過大，會對網絡帶寬造成壓力

不過需要注意的是，如果是批量消息的話，就不會進行壓縮，如下所示：

圖片

壓縮消息除了能夠減少網絡帶寬造成壓力之外，還能夠節省消息存儲空間

RocketMQ在往磁盤存消息的時候，并不會去解壓消息，而是直接將壓縮后的消息存到磁盤

消費者拉取到的消息其實也是壓縮后的消息

不過消費者在拿到消息之后會對消息進行解壓縮

當我們的業務系統拿到消息的時候，其實就是解壓縮后的消息

圖片

雖然壓縮消息能夠減少帶寬壓力和磁盤存儲壓力

但是由于壓縮和解壓縮的過程都是在客戶端（生產者、消費者）完成的

所以就會導致客戶端消耗更多的CPU資源，對CPU造成一定的壓力

高性能網絡通信模型

當生產者處理好消息之后，就會將消息通過網絡通信發送給服務端

而RocketMQ之所以快的一個非常重要原因就是它擁有高性能網絡通信模型

RocketMQ網絡通信這塊底層是基于Netty來實現的

圖片

Netty是一款非常強大、非常優秀的網絡應用程序框架，主要有以下幾個優點：

• 異步和事件驅動：Netty基于事件驅動的架構，使用了異步I/O操作，避免了阻塞式I/O調用的缺陷，能夠更有效地利用系統資源，提高并發處理能力。
• 高性能：Netty針對性能進行了優化，比如使用直接內存進行緩沖，減少垃圾回收的壓力和內存拷貝的開銷，提供了高吞吐量、低延遲的網絡通訊能力。
• 可擴展性：Netty的設計允許用戶自定義各種Handler來處理協議編碼、協議解碼和業務邏輯等。并且，它的模塊可插拔性設計使得用戶可以根據需要輕松地添加或更換組件。
• 簡化API：與Java原生NIO庫相比，Netty提供了更加簡潔易用的API，大大降低了網絡編程的復雜度。
• 安全：Netty內置了對SSL/TLS協議的支持，使得構建安全通信應用變得容易。
• 豐富的協議支持：Netty提供了HTTP、HTTP/2、WebSocket、Google Protocol Buffers等多種協議的編解碼支持，滿足不同網絡應用需求。
• ...

就是因為Netty如此的強大，所以不僅僅RocketMQ是基于Netty實現網絡通信的

幾乎絕大多數只要涉及到網絡通信的Java類框架，底層都離不開Netty的身影

比如知名RPC框架Dubbo、Java gRPC實現、Redis的親兒子Redisson、分布式任務調度平臺xxl-job等等

它們底層在實現網絡通信時，都是基于Netty框架

零拷貝技術

當消息達到RocketMQ服務端之后，為了能夠保證服務端重啟之后消息也不丟失，此時就需要將消息持久化到磁盤

由于涉及到消息持久化操作，就涉及到磁盤文件的讀寫操作

RocketMQ為了保證磁盤文件的高性能讀寫，使用到了一個叫零拷貝的技術

1、傳統IO讀寫方式

說零拷貝之前，先說一下傳統的IO讀寫方式。

比如現在有一個需求，將磁盤文件通過網絡傳輸出去

那么整個傳統的IO讀寫模型如下圖所示

圖片

傳統的IO讀寫其實就是read + write的操作，整個過程會分為如下幾步

• 用戶調用read()方法，開始讀取數據，此時發生一次上下文從用戶態到內核態的切換，也就是圖示的切換1
• 將磁盤數據通過DMA拷貝到內核緩存區
• 將內核緩存區的數據拷貝到用戶緩沖區，這樣用戶，也就是我們寫的代碼就能拿到文件的數據
• read()方法返回，此時就會從內核態切換到用戶態，也就是圖示的切換2
• 當我們拿到數據之后，就可以調用write()方法，此時上下文會從用戶態切換到內核態，即圖示切換3
• CPU將用戶緩沖區的數據拷貝到Socket緩沖區
• 將Socket緩沖區數據拷貝至網卡
• write()方法返回，上下文重新從內核態切換到用戶態，即圖示切換4

整個過程發生了4次上下文切換和4次數據的拷貝，這在高并發場景下肯定會嚴重影響讀寫性能。

所以為了減少上下文切換次數和數據拷貝次數，就引入了零拷貝技術。

2、零拷貝

零拷貝技術是一個思想，指的是指計算機執行操作時，CPU不需要先將數據從某處內存復制到另一個特定區域。

實現零拷貝的有以下兩種方式：

• mmap()
• sendfile()

mmap()

mmap（memory map）是一種內存映射文件的方法，即將一個文件或者其它對象映射到進程的地址空間，實現文件磁盤地址和進程虛擬地址空間中一段虛擬地址的一一對映關系。

簡單地說就是內核緩沖區和應用緩沖區進行映射

用戶在操作應用緩沖區時就好像在操作內核緩沖區

比如你往應用緩沖區寫數據，就好像直接往內核緩沖區寫數據，這個過程不涉及到CPU拷貝

而傳統IO就需要將在寫完應用緩沖區之后需要將數據通過CPU拷貝到內核緩沖區

同樣地上述文件傳輸功能，如果使用mmap的話，由于我們可以直接操作內核緩沖區

此時我們就可以將內核緩沖區的數據直接CPU拷貝到Socket緩沖區

整個IO模型就會如下圖所示：

圖片

基于mmap IO讀寫其實就變成mmap + write的操作，也就是用mmap替代傳統IO中的read操作

• 當用戶發起mmap調用的時候會發生上下文切換1，進行內存映射，然后數據被拷貝到內核緩沖區，mmap返回，發生上下文切換2
• 隨后用戶調用write，發生上下文切換3，將內核緩沖區的數據拷貝到Socket緩沖區，write返回，發生上下文切換4。

上下文切換的次數仍然是4次，但是拷貝次數只有3次，少了一次CPU拷貝。

所以總的來說，使用mmap就可以直接少一次CPU拷貝。

說了這么多，那么在Java中，如何去實現mmap，也就是內核緩沖區和應用緩沖區映射呢？

其實在Java NIO類庫中就提供了相應的API，當然底層也還是調用Linux系統的mmap()實現的，代碼如下所示

FileChannel fileChannel = new RandomAccessFile("test.txt", "rw").getChannel();
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, fileChannel.size());

MappedByteBuffer，你可以認為操作這個對象就好像直接操作內核緩沖區

比如可以通過MappedByteBuffer讀寫磁盤文件，此時就好像直接從內核緩沖區讀寫數據

當然也可以直接通過MappedByteBuffer將文件的數據拷貝到Socket緩沖區，實現上述文件傳輸的模型

這里我就不貼相應的代碼了

RocketMQ在存儲文件時，就是通過mmap技術來實現高效的文件讀寫

圖片

RocketMQ中使用mmap代碼

雖然前面一直說mmap不涉及CPU拷貝，但在某些特定場景下，尤其是在寫操作或特定的系統優化策略下，還是可能涉及CPU拷貝。

sendfile()

sendfile()跟mmap()一樣，也會減少一次CPU拷貝，但是它同時也會減少兩次上下文切換。

sendfile()主要是用于文件傳輸，比如將文件傳輸到另一個文件，又或者是網絡

當基于sendfile()時，一次文件傳輸的過程就如下圖所示：

圖片

用戶發起sendfile()調用時會發生切換1，之后數據通過DMA拷貝到內核緩沖區，之后再將內核緩沖區的數據CPU拷貝到Socket緩沖區，最后拷貝到網卡，sendfile()返回，發生切換2。

同樣地，Java NIO類庫中也提供了相應的API實現sendfile

當然底層還是操作系統的sendfile()

FileChannel channel = FileChannel.open(Paths.get("./test.txt"), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
//調用transferTo方法向目標數據傳輸
channel.transferTo(position, len, target);

FileChannel的transferTo方法底層就是基于sendfile來的

在如上代碼中，并沒有文件的讀寫操作，而是直接將文件的數據傳輸到target目標緩沖區

也就是說，sendfile傳輸文件時是無法知道文件的具體的數據的

但是mmap不一樣，mmap可以來直接修改內核緩沖區的數據

假設如果需要對文件的內容進行修改之后再傳輸，mmap可以滿足

小總結

在傳統IO中，如果想將用戶緩存區的數據放到內核緩沖區，需要經過CPU拷貝

而基于零拷貝技術可以減少CPU拷貝次數，常見的有兩種：

• mmap()
• sendfile()

mmap()是將用戶緩沖區和內核緩沖區共享，操作用戶緩沖區就好像直接操作內核緩沖區，讀寫數據時不需要CPU拷貝

Java中可以使用MappedByteBuffer這個API來達到操作內核緩沖區的效果

sendfile()主要是用于文件傳輸，可以通過sendfile()將一個文件內容傳輸到另一個文件中或者是網絡中

sendfile()在整個過程中是無法對文件內容進行修改的，如果想修改之后再傳輸，可以通過mmap來修改內容之后再傳輸

上面出現的API都是Java NIO標準類庫中的

如果你看的還是很迷糊，那直接記住一個結論

之所以基于零拷貝技術能夠高效的實現文件的讀寫操作，主要因為是減少了CPU拷貝次數和上下文切換次數

在RocketMQ中，底層是基于mmap()來實現文件的高效讀寫的

順序寫

RocketMQ在存儲消息時，除了使用零拷貝技術來實現文件的高效讀寫之外

還使用順序寫的方式提高數據寫入的速度

RocketMQ會將消息按照順序一條一條地寫入文件中

這種順序寫的方式由于減少了磁頭的移動和尋道時間，在大規模數據寫入的場景下，使得數據寫入的速度更快

高效的數據存儲結構

Topic和隊列的關系

在RocketMQ中，默認會為每個Topic在每個服務端Broker實例上創建4個隊列

圖片

如果有兩個Broker，那么默認就會有8個隊列

每個Broker上的隊列上的編號（queueId）都是從0開始

CommitLog

前面一直說，當消息到達RocektMQ服務端時，需要將消息存到磁盤文件

RocketMQ給這個存消息的文件起了一個高大上的名字：CommitLog

由于消息會很多，所以為了防止文件過大，CommitLog在物理磁盤文件上被分為多個磁盤文件，每個文件默認的固定大小是1G

圖片

消息在寫入到文件時，除了包含消息本身的內容數據，也還會包含其它信息，比如

• 消息的Topic
• 消息所在隊列的id，生產者發送消息時會攜帶這個隊列id
• 消息生產者的ip和端口
• ...

這些數據會和消息本身按照一定的順序同時寫到CommitLog文件中

圖片

上圖中黃色排列順序和實際的存的內容并非實際情況，我只是舉個例子

ConsumeQueue

除了CommitLog文件之外，RocketMQ還會為每個隊列創建一個磁盤文件

RocketMQ給這個文件也起了一個高大上的名字：ConsumeQueue

圖片

當消息被存到CommitLog之后，其實還會往這條消息所在隊列的ConsumeQueue文件中插一條數據

每個隊列的ConsumeQueue也是由多個文件組成，每個文件默認是存30萬條數據

插入ConsumeQueue中的每條數據由20個字節組成，包含3部分信息

• 消息在CommitLog的起始位置（8個字節），也被稱為偏移量
• 消息在CommitLog存儲的長度（8個字節）
• 消息tag的hashCode（4個字節）

圖片

每條數據也有自己的編號（offset），默認從0開始，依次遞增

所以，通過ConsumeQueue中存的數據可以從CommitLog中找到對應的消息

那么這個ConsumeQueue有什么作用呢？

其實通過名字也能猜到，這其實跟消息消費有關

當消費者拉取消息的時候，會告訴服務端四個比較重要的信息

• 自己需要拉取哪個Topic的消息
• 從Topic中的哪個隊列（queueId）拉取
• 從隊列的哪個位置（offset）拉取消息
• 拉取多少條消息(默認32條)

圖片

服務端接收到消息之后，總共分為四步處理：

• 首先會找到對應的Topic
• 之后根據queueId找到對應的ConsumeQueue文件
• 然后根據offset位置，從ConsumeQueue中讀取跟拉取消息條數一樣條數的數據

由于ConsumeQueue每條數據都是20個字節，所以根據offset的位置可以很快定位到應該從文件的哪個位置開始讀取數據

• 最后解析每條數據，根據偏移量和消息的長度到CommitLog文件查找真正的消息內容

整個過程如下圖所示：

圖片

所以，從這可以看出，當消費者在拉取消息時，ConsumeQueue其實就相當于是一個索引文件，方便快速查找在CommitLog中的消息

并且無論CommitLog存多少消息，整個查找消息的時間復雜度都是O(1)

由于ConsumeQueue每條數據都是20個字節，所以如果需要找第n條數據，只需要從第n * 20個字節的位置開始讀20個字節的數據即可，這個過程是O(1)的

當從ConsumeQueue找到數據之后，解析出消息在CommitLog存儲的起始位置和大小，之后就直接根據這兩個信息就可以從CommitLog中找到這條消息了，這個過程也是O(1)的

所以整個查找消息的過程就是O(1)的

所以從這就可以看出，ConsumeQueue和CommitLog相互配合，就能保證快速查找到消息，消費者從而就可以快速拉取消息

異步處理

RocketMQ在處理消息時，有很多異步操作，這里我舉兩個例子：

• 異步刷盤
• 異步主從復制

異步刷盤

前面說到，文件的內容都是先寫到內核緩沖區，也可以說是PageCache

而寫到PageCache并不能保證消息一定不丟失

因為如果服務器掛了，這部分數據還是可能會丟失的

所以為了解決這個問題，RocketMQ會開啟一個后臺線程

這個后臺線程默認每隔0.5s會將消息從PageCache刷到磁盤中

這樣就能保證消息真正的持久化到磁盤中

圖片

異步主從復制

在RocketMQ中，支持主從復制的集群模式

圖片

這種模式下，寫消息都是寫入到主節點，讀消息一般也是從主節點讀，但是有些情況下可能會從從節點讀

從節點在啟動的時候會跟主節點建立網絡連接

當主節點將消息存儲的CommitLog文件之后，會通過后臺一個異步線程，不停地將消息發送給從節點

從節點接收到消息之后，就直接將消息存到CommitLog文件

圖片

小總結

就是因為有這些異步操作，大大提高了消息存儲的效率

不過值得注意的，盡管異步可以提高效率，但是也增加了不確定性，比如丟消息等等

當然RocketMQ也支持同步等待消息刷盤和主從復制成功，但這肯定會導致性能降低

所以在項目中可以根據自己的業務需要選擇對應的刷盤和主從復制的策略

批量處理

除了異步之外，RocketMQ還大量使用了批量處理機制

比如前面說過，消費者拉取消息的時候，可以指定拉取拉取消息的條數，批量拉取消息

這種批量拉取機制可以減少消費者跟RocketMQ服務端的網絡通信次數，提高效率

除了批量拉取消息之外，RocketMQ在提交消費進度的時候也使用了批量處理機制

所謂的提交消費進度就是指

當消費者在成功消費消息之后，需要將所消費消息的offset（ConsumeQueue中的offset）提交給RocketMQ服務端

告訴RocketMQ，這個Queue的消息我已經消費到了這個位置了

這樣一旦消費者重啟了或者其它啥的要從這個Queue重新開始拉取消息的時候

此時他只需要問問RocketMQ服務端上次這個Queue消息消費到哪個位置了

圖片

之后消費者只需要從這個位置開始消費消息就行了，這樣就解決了接著消費的問題

RocketMQ在提交消費進度的時候并不是說每消費一條消息就提交一下這條消息對應的offset

而是默認每隔5s定時去批量提交一次這5s鐘消費消息的offset

鎖優化

由于RocketMQ內部采用了很多線程異步處理機制

這就一定會產生并發情況下的線程安全問題

在這種情況下，RocketMQ進行了多方面的鎖優化以提高性能和并發能力

就比如拿消息存儲來說

為了保證消息是按照順序一條一條地寫入到CommitLog文件中，就需要對這個寫消息的操作進行加鎖

而RocketMQ默認使用ReentrantLock來加鎖，并不是synchronized

圖片

當然除了默認情況外，RocketMQ還提供了一種基于CAS加鎖的實現

圖片

這種實現可以在寫消息壓力較低的情況下使用

當然除了寫消息之外，在一些其它的地方，RocketMQ也使用了基于CAS的原子操作來代替傳統的鎖機制

例如使用大量使用了AtomicInteger、AtomicLong等原子類來實現并發控制，避免了顯式的鎖競爭，提高了性能

線程池隔離

RocketMQ在處理請求的時候，會為不同的請求分配不同的線程池進行處理

比如對于消息存儲請求和拉取消息請求來說

Broker會有專門為它們分配兩個不同的線程池去分別處理這些請求

圖片

這種讓不同的業務由不同的線程池去處理的方式，能夠有效地隔離不同業務邏輯之間的線程資源的影響

比如消息存儲請求處理過慢并不會影響處理拉取消息請求

所以RocketMQ通過線程隔離及時可以有效地提高系統的并發性能和穩定性

總結

到這我就從10個方面講完了RocketMQ為什么這么快背后的原因

不知道你讀完文章之后有什么感受

其實實際上RocketMQ快的原因遠遠不止我上面說的這幾點

RocketMQ本身還做了很多其它的優化，比如拉取消息的長輪詢機制、文件預熱機制等等

正是因為有各種各樣設計細節上的優化，才最終決定了RocketMQ出色的性能表現

好了，本文就講到這里，如果覺得本文對你有點幫助，歡迎點贊、在看、收藏、轉發分享給其他需要的人

你的支持就是我更新的最大動力，感謝感謝！