大家好，我是三友~~

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

而這其中一個(gè)很重要原因就是，它處理消息和拉取消息的速度非?？?/p>

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

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

圖片

如果你對(duì)RocketMQ還不了解，可以從公眾號(hào)后臺(tái)菜單欄中查看我之前寫的關(guān)于RocketMQ的幾篇文章

如果你對(duì)RocketMQ源碼也感興趣，可以從下面這個(gè)倉(cāng)庫fork一下源碼，我在源碼中加了中文注釋，并且后面我還會(huì)持續(xù)更新注釋

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

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

批量發(fā)送消息

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

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

        //用以及集合保存多個(gè)消息
        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()));
        // 發(fā)送消息并得到消息的發(fā)送結(jié)果，然后打印
        SendResult sendResult = producer.send(messages);
        System.out.printf("%s%n", sendResult);

        // 關(guān)閉生產(chǎn)者
        producer.shutdown();
    }

}

通過批量發(fā)送消息，減少了RocketMQ客戶端與服務(wù)端，也就是Broker之間的網(wǎng)絡(luò)通信次數(shù)，提高傳輸效率

不過在使用批量消息的時(shí)候，需要注意以下三點(diǎn)：

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

消息壓縮

RocketMQ在發(fā)送消息的時(shí)候，當(dāng)發(fā)現(xiàn)消息的大小超過4k的時(shí)候，就會(huì)對(duì)消息進(jìn)行壓縮

這是因?yàn)槿绻⑦^大，會(huì)對(duì)網(wǎng)絡(luò)帶寬造成壓力

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

圖片

壓縮消息除了能夠減少網(wǎng)絡(luò)帶寬造成壓力之外，還能夠節(jié)省消息存儲(chǔ)空間

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

消費(fèi)者拉取到的消息其實(shí)也是壓縮后的消息

不過消費(fèi)者在拿到消息之后會(huì)對(duì)消息進(jìn)行解壓縮

當(dāng)我們的業(yè)務(wù)系統(tǒng)拿到消息的時(shí)候，其實(shí)就是解壓縮后的消息

圖片

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

但是由于壓縮和解壓縮的過程都是在客戶端（生產(chǎn)者、消費(fèi)者）完成的

所以就會(huì)導(dǎo)致客戶端消耗更多的CPU資源，對(duì)CPU造成一定的壓力

高性能網(wǎng)絡(luò)通信模型

當(dāng)生產(chǎn)者處理好消息之后，就會(huì)將消息通過網(wǎng)絡(luò)通信發(fā)送給服務(wù)端

而RocketMQ之所以快的一個(gè)非常重要原因就是它擁有高性能網(wǎng)絡(luò)通信模型

RocketMQ網(wǎng)絡(luò)通信這塊底層是基于Netty來實(shí)現(xiàn)的

圖片

Netty是一款非常強(qiáng)大、非常優(yōu)秀的網(wǎng)絡(luò)應(yīng)用程序框架，主要有以下幾個(gè)優(yōu)點(diǎn)：

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

就是因?yàn)镹etty如此的強(qiáng)大，所以不僅僅RocketMQ是基于Netty實(shí)現(xiàn)網(wǎng)絡(luò)通信的

幾乎絕大多數(shù)只要涉及到網(wǎng)絡(luò)通信的Java類框架，底層都離不開Netty的身影

比如知名RPC框架Dubbo、Java gRPC實(shí)現(xiàn)、Redis的親兒子Redisson、分布式任務(wù)調(diào)度平臺(tái)xxl-job等等

它們底層在實(shí)現(xiàn)網(wǎng)絡(luò)通信時(shí)，都是基于Netty框架

零拷貝技術(shù)

當(dāng)消息達(dá)到RocketMQ服務(wù)端之后，為了能夠保證服務(wù)端重啟之后消息也不丟失，此時(shí)就需要將消息持久化到磁盤

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

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

1、傳統(tǒng)IO讀寫方式

說零拷貝之前，先說一下傳統(tǒng)的IO讀寫方式。

比如現(xiàn)在有一個(gè)需求，將磁盤文件通過網(wǎng)絡(luò)傳輸出去

那么整個(gè)傳統(tǒng)的IO讀寫模型如下圖所示

圖片

傳統(tǒng)的IO讀寫其實(shí)就是read + write的操作，整個(gè)過程會(huì)分為如下幾步

• 用戶調(diào)用read()方法，開始讀取數(shù)據(jù)，此時(shí)發(fā)生一次上下文從用戶態(tài)到內(nèi)核態(tài)的切換，也就是圖示的切換1
• 將磁盤數(shù)據(jù)通過DMA拷貝到內(nèi)核緩存區(qū)
• 將內(nèi)核緩存區(qū)的數(shù)據(jù)拷貝到用戶緩沖區(qū)，這樣用戶，也就是我們寫的代碼就能拿到文件的數(shù)據(jù)
• read()方法返回，此時(shí)就會(huì)從內(nèi)核態(tài)切換到用戶態(tài)，也就是圖示的切換2
• 當(dāng)我們拿到數(shù)據(jù)之后，就可以調(diào)用write()方法，此時(shí)上下文會(huì)從用戶態(tài)切換到內(nèi)核態(tài)，即圖示切換3
• CPU將用戶緩沖區(qū)的數(shù)據(jù)拷貝到Socket緩沖區(qū)
• 將Socket緩沖區(qū)數(shù)據(jù)拷貝至網(wǎng)卡
• write()方法返回，上下文重新從內(nèi)核態(tài)切換到用戶態(tài)，即圖示切換4

整個(gè)過程發(fā)生了4次上下文切換和4次數(shù)據(jù)的拷貝，這在高并發(fā)場(chǎng)景下肯定會(huì)嚴(yán)重影響讀寫性能。

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

2、零拷貝

零拷貝技術(shù)是一個(gè)思想，指的是指計(jì)算機(jī)執(zhí)行操作時(shí)，CPU不需要先將數(shù)據(jù)從某處內(nèi)存復(fù)制到另一個(gè)特定區(qū)域。

實(shí)現(xiàn)零拷貝的有以下兩種方式：

• mmap()
• sendfile()

mmap()

mmap（memory map）是一種內(nèi)存映射文件的方法，即將一個(gè)文件或者其它對(duì)象映射到進(jìn)程的地址空間，實(shí)現(xiàn)文件磁盤地址和進(jìn)程虛擬地址空間中一段虛擬地址的一一對(duì)映關(guān)系。

簡(jiǎn)單地說就是內(nèi)核緩沖區(qū)和應(yīng)用緩沖區(qū)進(jìn)行映射

用戶在操作應(yīng)用緩沖區(qū)時(shí)就好像在操作內(nèi)核緩沖區(qū)

比如你往應(yīng)用緩沖區(qū)寫數(shù)據(jù)，就好像直接往內(nèi)核緩沖區(qū)寫數(shù)據(jù)，這個(gè)過程不涉及到CPU拷貝

而傳統(tǒng)IO就需要將在寫完應(yīng)用緩沖區(qū)之后需要將數(shù)據(jù)通過CPU拷貝到內(nèi)核緩沖區(qū)

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

此時(shí)我們就可以將內(nèi)核緩沖區(qū)的數(shù)據(jù)直接CPU拷貝到Socket緩沖區(qū)

整個(gè)IO模型就會(huì)如下圖所示：

圖片

基于mmap IO讀寫其實(shí)就變成mmap + write的操作，也就是用mmap替代傳統(tǒng)IO中的read操作

• 當(dāng)用戶發(fā)起mmap調(diào)用的時(shí)候會(huì)發(fā)生上下文切換1，進(jìn)行內(nèi)存映射，然后數(shù)據(jù)被拷貝到內(nèi)核緩沖區(qū)，mmap返回，發(fā)生上下文切換2
• 隨后用戶調(diào)用write，發(fā)生上下文切換3，將內(nèi)核緩沖區(qū)的數(shù)據(jù)拷貝到Socket緩沖區(qū)，write返回，發(fā)生上下文切換4。

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

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

說了這么多，那么在Java中，如何去實(shí)現(xiàn)mmap，也就是內(nèi)核緩沖區(qū)和應(yīng)用緩沖區(qū)映射呢？

其實(shí)在Java NIO類庫中就提供了相應(yīng)的API，當(dāng)然底層也還是調(diào)用Linux系統(tǒng)的mmap()實(shí)現(xiàn)的，代碼如下所示

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

MappedByteBuffer，你可以認(rèn)為操作這個(gè)對(duì)象就好像直接操作內(nèi)核緩沖區(qū)

比如可以通過MappedByteBuffer讀寫磁盤文件，此時(shí)就好像直接從內(nèi)核緩沖區(qū)讀寫數(shù)據(jù)

當(dāng)然也可以直接通過MappedByteBuffer將文件的數(shù)據(jù)拷貝到Socket緩沖區(qū)，實(shí)現(xiàn)上述文件傳輸?shù)哪Ｐ?/p>

這里我就不貼相應(yīng)的代碼了

RocketMQ在存儲(chǔ)文件時(shí)，就是通過mmap技術(shù)來實(shí)現(xiàn)高效的文件讀寫

圖片

RocketMQ中使用mmap代碼

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

sendfile()

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

sendfile()主要是用于文件傳輸，比如將文件傳輸?shù)搅硪粋€(gè)文件，又或者是網(wǎng)絡(luò)

當(dāng)基于sendfile()時(shí)，一次文件傳輸?shù)倪^程就如下圖所示：

圖片

用戶發(fā)起sendfile()調(diào)用時(shí)會(huì)發(fā)生切換1，之后數(shù)據(jù)通過DMA拷貝到內(nèi)核緩沖區(qū)，之后再將內(nèi)核緩沖區(qū)的數(shù)據(jù)CPU拷貝到Socket緩沖區(qū)，最后拷貝到網(wǎng)卡，sendfile()返回，發(fā)生切換2。

同樣地，Java NIO類庫中也提供了相應(yīng)的API實(shí)現(xiàn)sendfile

當(dāng)然底層還是操作系統(tǒng)的sendfile()

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

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

在如上代碼中，并沒有文件的讀寫操作，而是直接將文件的數(shù)據(jù)傳輸?shù)絫arget目標(biāo)緩沖區(qū)

也就是說，sendfile傳輸文件時(shí)是無法知道文件的具體的數(shù)據(jù)的

但是mmap不一樣，mmap可以來直接修改內(nèi)核緩沖區(qū)的數(shù)據(jù)

假設(shè)如果需要對(duì)文件的內(nèi)容進(jìn)行修改之后再傳輸，mmap可以滿足

小總結(jié)

在傳統(tǒng)IO中，如果想將用戶緩存區(qū)的數(shù)據(jù)放到內(nèi)核緩沖區(qū)，需要經(jīng)過CPU拷貝

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

• mmap()
• sendfile()

mmap()是將用戶緩沖區(qū)和內(nèi)核緩沖區(qū)共享，操作用戶緩沖區(qū)就好像直接操作內(nèi)核緩沖區(qū)，讀寫數(shù)據(jù)時(shí)不需要CPU拷貝

Java中可以使用MappedByteBuffer這個(gè)API來達(dá)到操作內(nèi)核緩沖區(qū)的效果

sendfile()主要是用于文件傳輸，可以通過sendfile()將一個(gè)文件內(nèi)容傳輸?shù)搅硪粋€(gè)文件中或者是網(wǎng)絡(luò)中

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

上面出現(xiàn)的API都是Java NIO標(biāo)準(zhǔn)類庫中的

如果你看的還是很迷糊，那直接記住一個(gè)結(jié)論

之所以基于零拷貝技術(shù)能夠高效的實(shí)現(xiàn)文件的讀寫操作，主要因?yàn)槭菧p少了CPU拷貝次數(shù)和上下文切換次數(shù)

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

順序?qū)?/h2>

RocketMQ在存儲(chǔ)消息時(shí)，除了使用零拷貝技術(shù)來實(shí)現(xiàn)文件的高效讀寫之外

還使用順序?qū)懙姆绞教岣邤?shù)據(jù)寫入的速度

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

這種順序?qū)懙姆绞接捎跍p少了磁頭的移動(dòng)和尋道時(shí)間，在大規(guī)模數(shù)據(jù)寫入的場(chǎng)景下，使得數(shù)據(jù)寫入的速度更快

高效的數(shù)據(jù)存儲(chǔ)結(jié)構(gòu)

Topic和隊(duì)列的關(guān)系

在RocketMQ中，默認(rèn)會(huì)為每個(gè)Topic在每個(gè)服務(wù)端Broker實(shí)例上創(chuàng)建4個(gè)隊(duì)列

圖片

如果有兩個(gè)Broker，那么默認(rèn)就會(huì)有8個(gè)隊(duì)列

每個(gè)Broker上的隊(duì)列上的編號(hào)（queueId）都是從0開始

CommitLog

前面一直說，當(dāng)消息到達(dá)RocektMQ服務(wù)端時(shí)，需要將消息存到磁盤文件

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

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

圖片

消息在寫入到文件時(shí)，除了包含消息本身的內(nèi)容數(shù)據(jù)，也還會(huì)包含其它信息，比如

• 消息的Topic
• 消息所在隊(duì)列的id，生產(chǎn)者發(fā)送消息時(shí)會(huì)攜帶這個(gè)隊(duì)列id
• 消息生產(chǎn)者的ip和端口
• ...

這些數(shù)據(jù)會(huì)和消息本身按照一定的順序同時(shí)寫到CommitLog文件中

圖片

上圖中黃色排列順序和實(shí)際的存的內(nèi)容并非實(shí)際情況，我只是舉個(gè)例子

ConsumeQueue

除了CommitLog文件之外，RocketMQ還會(huì)為每個(gè)隊(duì)列創(chuàng)建一個(gè)磁盤文件

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

圖片

當(dāng)消息被存到CommitLog之后，其實(shí)還會(huì)往這條消息所在隊(duì)列的ConsumeQueue文件中插一條數(shù)據(jù)

每個(gè)隊(duì)列的ConsumeQueue也是由多個(gè)文件組成，每個(gè)文件默認(rèn)是存30萬條數(shù)據(jù)

插入ConsumeQueue中的每條數(shù)據(jù)由20個(gè)字節(jié)組成，包含3部分信息

• 消息在CommitLog的起始位置（8個(gè)字節(jié)），也被稱為偏移量
• 消息在CommitLog存儲(chǔ)的長(zhǎng)度（8個(gè)字節(jié)）
• 消息tag的hashCode（4個(gè)字節(jié)）

圖片

每條數(shù)據(jù)也有自己的編號(hào)（offset），默認(rèn)從0開始，依次遞增

所以，通過ConsumeQueue中存的數(shù)據(jù)可以從CommitLog中找到對(duì)應(yīng)的消息

那么這個(gè)ConsumeQueue有什么作用呢？

其實(shí)通過名字也能猜到，這其實(shí)跟消息消費(fèi)有關(guān)

當(dāng)消費(fèi)者拉取消息的時(shí)候，會(huì)告訴服務(wù)端四個(gè)比較重要的信息

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

圖片

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

• 首先會(huì)找到對(duì)應(yīng)的Topic
• 之后根據(jù)queueId找到對(duì)應(yīng)的ConsumeQueue文件
• 然后根據(jù)offset位置，從ConsumeQueue中讀取跟拉取消息條數(shù)一樣條數(shù)的數(shù)據(jù)

由于ConsumeQueue每條數(shù)據(jù)都是20個(gè)字節(jié)，所以根據(jù)offset的位置可以很快定位到應(yīng)該從文件的哪個(gè)位置開始讀取數(shù)據(jù)

• 最后解析每條數(shù)據(jù)，根據(jù)偏移量和消息的長(zhǎng)度到CommitLog文件查找真正的消息內(nèi)容

整個(gè)過程如下圖所示：

圖片

所以，從這可以看出，當(dāng)消費(fèi)者在拉取消息時(shí)，ConsumeQueue其實(shí)就相當(dāng)于是一個(gè)索引文件，方便快速查找在CommitLog中的消息

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

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

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

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

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

異步處理

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

• 異步刷盤
• 異步主從復(fù)制

異步刷盤

前面說到，文件的內(nèi)容都是先寫到內(nèi)核緩沖區(qū)，也可以說是PageCache

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

因?yàn)槿绻?wù)器掛了，這部分?jǐn)?shù)據(jù)還是可能會(huì)丟失的

所以為了解決這個(gè)問題，RocketMQ會(huì)開啟一個(gè)后臺(tái)線程

這個(gè)后臺(tái)線程默認(rèn)每隔0.5s會(huì)將消息從PageCache刷到磁盤中

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

圖片

異步主從復(fù)制

在RocketMQ中，支持主從復(fù)制的集群模式

圖片

這種模式下，寫消息都是寫入到主節(jié)點(diǎn)，讀消息一般也是從主節(jié)點(diǎn)讀，但是有些情況下可能會(huì)從從節(jié)點(diǎn)讀

從節(jié)點(diǎn)在啟動(dòng)的時(shí)候會(huì)跟主節(jié)點(diǎn)建立網(wǎng)絡(luò)連接

當(dāng)主節(jié)點(diǎn)將消息存儲(chǔ)的CommitLog文件之后，會(huì)通過后臺(tái)一個(gè)異步線程，不停地將消息發(fā)送給從節(jié)點(diǎn)

從節(jié)點(diǎn)接收到消息之后，就直接將消息存到CommitLog文件

圖片

小總結(jié)

就是因?yàn)橛羞@些異步操作，大大提高了消息存儲(chǔ)的效率

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

當(dāng)然RocketMQ也支持同步等待消息刷盤和主從復(fù)制成功，但這肯定會(huì)導(dǎo)致性能降低

所以在項(xiàng)目中可以根據(jù)自己的業(yè)務(wù)需要選擇對(duì)應(yīng)的刷盤和主從復(fù)制的策略

批量處理

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

比如前面說過，消費(fèi)者拉取消息的時(shí)候，可以指定拉取拉取消息的條數(shù)，批量拉取消息

這種批量拉取機(jī)制可以減少消費(fèi)者跟RocketMQ服務(wù)端的網(wǎng)絡(luò)通信次數(shù)，提高效率

除了批量拉取消息之外，RocketMQ在提交消費(fèi)進(jìn)度的時(shí)候也使用了批量處理機(jī)制

所謂的提交消費(fèi)進(jìn)度就是指

當(dāng)消費(fèi)者在成功消費(fèi)消息之后，需要將所消費(fèi)消息的offset（ConsumeQueue中的offset）提交給RocketMQ服務(wù)端

告訴RocketMQ，這個(gè)Queue的消息我已經(jīng)消費(fèi)到了這個(gè)位置了

這樣一旦消費(fèi)者重啟了或者其它啥的要從這個(gè)Queue重新開始拉取消息的時(shí)候

此時(shí)他只需要問問RocketMQ服務(wù)端上次這個(gè)Queue消息消費(fèi)到哪個(gè)位置了

圖片

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

RocketMQ在提交消費(fèi)進(jìn)度的時(shí)候并不是說每消費(fèi)一條消息就提交一下這條消息對(duì)應(yīng)的offset

而是默認(rèn)每隔5s定時(shí)去批量提交一次這5s鐘消費(fèi)消息的offset

鎖優(yōu)化

由于RocketMQ內(nèi)部采用了很多線程異步處理機(jī)制

這就一定會(huì)產(chǎn)生并發(fā)情況下的線程安全問題

在這種情況下，RocketMQ進(jìn)行了多方面的鎖優(yōu)化以提高性能和并發(fā)能力

就比如拿消息存儲(chǔ)來說

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

而RocketMQ默認(rèn)使用ReentrantLock來加鎖，并不是synchronized

圖片

當(dāng)然除了默認(rèn)情況外，RocketMQ還提供了一種基于CAS加鎖的實(shí)現(xiàn)

圖片

這種實(shí)現(xiàn)可以在寫消息壓力較低的情況下使用

當(dāng)然除了寫消息之外，在一些其它的地方，RocketMQ也使用了基于CAS的原子操作來代替?zhèn)鹘y(tǒng)的鎖機(jī)制

例如使用大量使用了AtomicInteger、AtomicLong等原子類來實(shí)現(xiàn)并發(fā)控制，避免了顯式的鎖競(jìng)爭(zhēng)，提高了性能

線程池隔離

RocketMQ在處理請(qǐng)求的時(shí)候，會(huì)為不同的請(qǐng)求分配不同的線程池進(jìn)行處理

比如對(duì)于消息存儲(chǔ)請(qǐng)求和拉取消息請(qǐng)求來說

Broker會(huì)有專門為它們分配兩個(gè)不同的線程池去分別處理這些請(qǐng)求

圖片

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

比如消息存儲(chǔ)請(qǐng)求處理過慢并不會(huì)影響處理拉取消息請(qǐng)求

所以RocketMQ通過線程隔離及時(shí)可以有效地提高系統(tǒng)的并發(fā)性能和穩(wěn)定性

總結(jié)

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

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

其實(shí)實(shí)際上RocketMQ快的原因遠(yuǎn)遠(yuǎn)不止我上面說的這幾點(diǎn)

RocketMQ本身還做了很多其它的優(yōu)化，比如拉取消息的長(zhǎng)輪詢機(jī)制、文件預(yù)熱機(jī)制等等

正是因?yàn)橛懈鞣N各樣設(shè)計(jì)細(xì)節(jié)上的優(yōu)化，才最終決定了RocketMQ出色的性能表現(xiàn)

好了，本文就講到這里，如果覺得本文對(duì)你有點(diǎn)幫助，歡迎點(diǎn)贊、在看、收藏、轉(zhuǎn)發(fā)分享給其他需要的人

你的支持就是我更新的最大動(dòng)力，感謝感謝！