從官方這邊獲悉，RocketMQ在4.9.1版本中對消息發送進行了大量的優化，性能提升十分顯著，接下來請跟著我一起來欣賞大神們的杰作。

根據RocketMQ4.9.1的更新日志，我們從中提取到關于消息發送性能優化的【Issues:2883】,詳細鏈接如下：具體優化點如截圖所示：

首先先嘗試對上述優化點做一個簡單的介紹：

• 對WaitNotifyObject的鎖進行優化(item2)

• 移除HAService中的鎖(item3)

• 移除GroupCommitService中的鎖(item4)

• 消除HA中不必要的數組拷貝(item5)

• 調整消息發送幾個參數的默認值(item7)

  • sendMessageThreadPoolNums

  • useReentrantLockWhenPutMessage

  • flushCommitLogTimed

  • endTransactionThreadPoolNums

• 減少瑣的作用范圍(item8-12)

通過閱讀上述的變更，總結出優化手段主要包括如下三點：

• 移除不必要的鎖

• 降低鎖粒度(范圍)

• 修改消息發送相關參數

接下來結合源碼，從中挑選具有代表性功能進行詳細剖析，一起領悟Java高并發編程的魅力。

?

1、移除不必要的鎖

本次性能優化，主要針對的是RocketMQ同步復制場景。

我們首先先來簡單介紹一下RocketMQ主從同步在編程方面的技巧。

RocketMQ主節點將消息寫入內存后， 如果采用同步復制，需要等待從節點成功寫入后才能向消息發送客戶端返回成功，在代碼編寫方面也極具技巧性，時許圖如下圖所示：

溫馨提示：在RocketMQ4.7版本開始對消息發送進行了優化，同步消息發送模型引入了jdk的CompletableFuture實現消息的異步發送。

核心步驟解讀：

消息發送線程調用Commitlog的aysncPutMessage方法寫入消息。

Commitlog調用submitReplicaRequest方法，將任務提交到GroupTransferService中，并獲取一個Future，實現異步編程。值得注意的是這里需要等待，待數據成功寫入從節點（內部基于CompletableFuture機制的內部線程池ForkJoin）。

GroupTransferService中對提交的任務依次進行判斷，判斷對應的請求是否已同步到從節點。

如果已經復制到從節點，則通過Future喚醒，并將結果返回給消息發送端。

GroupTransferService代碼如下圖所示：

為了更加方便大家理解接下來的優化點，首先再總結提煉一下GroupTransferService的設計理念：

• 首先引入兩個List結合，分別命名為讀、寫鏈表。

• 外部調用GroupTransferService的putRequest請求，將存儲在寫鏈表中(requestWrite)。

• GroupTransferService的run方法從requestRead鏈表中獲取任務，判斷這些任務對應的請求的數據是否成功寫入到從節點。

• 每當requestRead中沒有數據可讀時，兩個隊列進行交互，從而實現讀寫分離，降低鎖競爭。

新版本的優化點主要包括：

• 更改putRequest的鎖類型，用自旋鎖替換synchronized

• 去除doWaitTransfer方法中多余的鎖

1.1 使用自旋鎖替換synchronized

正入下圖所示，GroupTransferService向外提供接口putRequest，用來接受外部的同步任務，需要對ArrayList加鎖進行保護，往ArrayList中添加數據屬于一個內存操作，操作耗時小。

故這里沒必要采取synchronized這種synchronized，而是可以自旋鎖，自旋鎖的實現非常輕量級，其實現如下圖所示：

整個鎖的實現就只需引入一個AtomicBoolean，加鎖、釋放鎖都是基于CAS操作，非常的輕量，并且自旋鎖不會發生線程切換。

1.2 去除多余的鎖

“鎖”的濫用是一個非常普遍的現象，多線程環境編程是一個非常復雜的交互過程，在編寫代碼過程中我們可能覺得自己無法預知這段代碼是否會被多個線程并發執行，為了謹慎起見，就直接簡單粗暴的對其進行加鎖，帶來的自然是性能的損耗，這里將該鎖去除，我們就要結合該類的調用鏈條，判斷是否需要加鎖。

整個GroupTransferService中在多線程環境中運行需要被保護的主要是requestRead與requestWrite集合，引入的鎖的目的也是確保這兩個集合在多線程環境下安全訪問，故我們首先應該梳理一下GroupTransferService的核心方法的運作流程：

doWaitTransfer方法操作的主要對象是requestRead鏈表，而且該方法只會被GroupTransferService線程調用，并且requestRead中方法會在swapRequest中被修改，但這兩個方法是串行執行，而且在同一個線程中，故無需引入鎖，該鎖可以移除。

但由于該鎖被移除，在swapRequests中進行加鎖，因為requestWrite這個隊列會被多個線程訪問，優化后的代碼如下：

從這個角度來看，其實主要是將鎖的類型由synchronized替換為更加輕量的自旋鎖。

?

2、降低鎖的范圍

被鎖包裹的代碼塊是串行執行，即無法并發，在無法避免鎖的情況下，降低鎖的代碼塊，能有效提高并發度，圖解如下：

如果多個線程區訪問lock1,lock2，在lock1中domSomeThing1、domSomeThing2這兩個方法都必須串行執行，而多個線程同時訪問lock2方法，doSomeThing1能被多個線程同時執行，只有doSomething2時才需要串行執行，其整體并發效果肯定是lock2，基于這樣理論：得出一個鎖使用的最佳實踐：被鎖包裹的代碼塊越少越好。

在老版本中，消息寫入加鎖的代碼塊比較大，一些可以并發執行的動作也被鎖包裹，例如生成offsetMsgId。

新版本采用函數式編程的思路，只是定義來獲取msgId的方法，在進行消息寫入時并不會執行，降低鎖的粒度，使得offsetMsgId的生成并行化，其編程手段之巧妙，值得我們學習。

?

3、調整消息發送相關的參數

sendMessageThreadPoolNums

Broker端消息發送端線程池數量，該值在4.9.0版本之前默認為1，新版本調整為操作系統的CPU核數，并且不小于4。該參數的調整有利有弊。提高了消息發送的并發度，但同時會導致消息順序的亂序,其示例圖如下同步發送下不會有順序問題，可放心修改

在順序消費場景，該參數不建議修改。在實際過程中應該對RocketMQ集群進行治理，順序消費的場景使用專門集群。

useReentrantLockWhenPutMessage MQ消息寫入時對內存加鎖使用的鎖類型，低版本之前默認為false,表示默認使用自旋鎖；新版本使用ReentrantLock。自旋主要的優勢是沒有線程切換成本，但自旋容易造成CPU的浪費，內存寫入大部分情況下是很快，但RocketMQ比較依賴頁緩存，如果出現也緩存抖動，帶來的CPU浪費是非常不值得，在sendMessageThreadPoolNums設置超過1之后，鎖的類型使用ReentrantLock更加穩定。

flushCommitLogTimed 首先我們通過觀察源碼了解一下該參數的含義：

其主要作用是控制刷盤線程阻塞等待的方式，低版本flushCommitLogTimed為false，默認使用CountDownLatch，而高版本則直接使用Thread.sleep。猜想的原因是刷盤線程比較獨立，無需與其他線程進行直接的交互協作，故無需使用CountDownLatch這種專門用來線程協作的“外來和尚”。

endTransactionThreadPoolNums

主要用于設置事務消息線程池的大小。

新版本主要是可通過調整發送線程池來動態調節事務消息的值，這個大家可以根據壓測結果動態調整。

有道無術，術可成；有術無道，止于術

歡迎大家關注Java之道公眾號

好文章，我在看??

總結

以上是生活随笔為你收集整理的RocketMQ这样做，压测后性能提高30%的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。