輕鬆搞定RocketMQ入門

RocketMQ是一款分佈式、隊列模型的消息中間件，具備如下特色：

可以保證嚴格的消息順序

提供豐富的消息拉取模式

高效的訂閱者水平擴展能力

實時的消息訂閱機制

億級消息堆積能力

RocketMQ網絡部署特

（1）NameServer是一個幾乎無狀態的節點，可集羣部署，節點之間無任何信息同步

（2）Broker部署相對複雜，Broker氛圍Master與Slave，一個Master能夠對應多個Slaver，可是一個Slaver只能對應一個Master，Master與Slaver的對應關係經過指定相同的BrokerName，不一樣的BrokerId來定義，BrokerId爲0表示Master，非0表示Slaver。Master能夠部署多個。每一個Broker與NameServer集羣中的全部節點創建長鏈接，定時註冊Topic信息到全部的NameServer

（3）Producer與NameServer集羣中的其中一個節點（隨機選擇）創建長鏈接，按期從NameServer取Topic路由信息，並向提供Topic服務的Master創建長鏈接，且定時向Master發送心跳。Produce徹底無狀態，可集羣部署

（4）Consumer與NameServer集羣中的其中一個節點（隨機選擇）創建長鏈接，按期從NameServer取Topic路由信息，並向提供Topic服務的Master、Slaver創建長鏈接，且定時向Master、Slaver發送心跳。Consumer便可從Master訂閱消息，也能夠從Slave訂閱消息，訂閱規則由Broker配置決定

RocketMQ儲存特色

（1）零拷貝原理：Consumer消費消息過程，使用了零拷貝，零拷貝包括一下2中方式，RocketMQ使用第一種方式，因小塊數據傳輸的要求效果比sendfile方式好

a )使用mmap+write方式

優勢：即便頻繁調用，使用小文件塊傳輸，效率也很高

缺點：不能很好的利用DMA方式，會比sendfile多消耗CPU資源，內存安全性控制複雜，須要避免JVM Crash問題

b）使用sendfile方式

優勢：能夠利用DMA方式，消耗CPU資源少，大塊文件傳輸效率高，無內存安全新問題

缺點：小塊文件效率低於mmap方式，只能是BIO方式傳輸，不能使用NIO

（2）數據存儲結構

RocketMQ關鍵特性

1.單機支持1W以上的持久化隊

（1）全部數據單獨儲存到commit Log ，徹底順序寫，隨機讀

（2）對最終用戶展示的隊列實際只儲存消息在Commit Log 的位置信息，而且串行方式刷盤

這樣作的好處：

（1）隊列輕量化，單個隊列數據量很是少

（2）對磁盤的訪問串行話，避免磁盤競爭，不會由於隊列增長致使IOWait增高

每一個方案都有優缺點，他的缺點是：

（1）寫雖然是順序寫，可是讀卻變成了隨機讀

（2）讀一條消息，會先讀Consume Queue，再讀Commit Log，增長了開銷

（3）要保證Commit Log 與 Consume Queue徹底的一致，增長了編程的複雜度

以上缺點如何客服：

（1）隨機讀，儘量讓讀命中pagecache，減小IO操做，因此內存越大越好。若是系統中堆積的消息過多，讀數據要訪問硬盤會不會因爲隨機讀致使系統性能急劇降低，答案是否認的。

a）訪問pagecache時，即便只訪問1K的消息，系統也會提早預讀出更多的數據，在下次讀時就可能命中pagecache

b）隨機訪問Commit Log 磁盤數據，系統IO調度算法設置爲NOOP方式，會在必定程度上將徹底的隨機讀變成順序跳躍方式，而順序跳躍方式讀較徹底的隨機讀性能高5倍

（2）因爲Consume Queue存儲數量極少，並且順序讀，在pagecache的與讀取狀況下，Consume Queue的讀性能與內存幾乎一直，即便堆積狀況下。因此能夠認爲Consume Queue徹底不會阻礙讀性能

（3）Commit Log中存儲了全部的元信息，包含消息體，相似於MySQl、Oracle的redolog，因此只要有Commit Log存在， Consume Queue即便丟失數據，仍能夠恢復出來

2.刷盤策略

rocketmq中的全部消息都是持久化的，先寫入系統pagecache，而後刷盤，能夠保證內存與磁盤都有一份數據，訪問時，能夠直接從內存讀取

2.1異步刷盤

在有 RAID 卡， SAS 15000 轉磁盤測試順序寫文件，速度能夠達到 300M 每秒左右，而線上的網卡通常都爲千兆網卡，寫磁盤速度明顯快於數據網絡入口速度，那麼是否能夠作到寫完 內存就向用戶返回，由後臺線程刷盤呢？

(1). 因爲磁盤速度大於網卡速度，那麼刷盤的進度確定能夠跟上消息的寫入速度。

(2). 萬一因爲此時系統壓力過大，可能堆積消息，除了寫入 IO，還有讀取 IO，萬一出現磁盤讀取落後狀況，會不會致使系統內存溢出，答案是否認的，緣由以下：

a) 寫入消息到 PAGECACHE 時，若是內存不足，則嘗試丟棄乾淨的 PAGE，騰出內存供新消息使用，策略是 LRU 方式。

b) 若是乾淨頁不足，此時寫入 PAGECACHE 會被阻塞，系統嘗試刷盤部分數據，大約每次嘗試 32 個 PAGE，來找出更多幹淨 PAGE。
綜上，內存溢出的狀況不會出現

2.2同步刷盤：

同步刷盤與異步刷盤的惟一區別是異步刷盤寫完 PAGECACHE 直接返回，而同步刷盤須要等待刷盤完成才返回，同步刷盤流程以下：

（1）寫入 PAGECACHE 後，線程等待，通知刷盤線程刷盤。

（2）刷盤線程刷盤後，喚醒前端等待線程，多是一批線程。

（3）前端等待線程向用戶返回成功。

3.消息查詢

3.1按照MessageId查詢消息

MsgId總共16個字節，包含消息儲存主機地址，消息Commit Log Offset。從MsgId中解析出Broker的地址和Commit Log 偏移地址，而後按照存儲格式所在位置消息buffer解析成一個完整消息

3.2按照Message Key查詢消息

1.根據查詢的key的hashcode%slotNum獲得具體的槽位置 （slotNum是一個索引文件裏面包含的最大槽目數目，例如圖中所示slotNum=500W）

2.根據slotValue（slot對應位置的值）查找到索引項列表的最後一項（倒序排列，slotValue老是指向最新的一個索引項）

3.遍歷索引項列表返回查詢時間範圍內的結果集（默認一次最大返回的32條記錄）

4.Hash衝突，尋找key的slot位置時至關於執行了兩次散列函數，一次key的hash，一次key的hash取值模，所以這裏存在兩次衝突的狀況；第一種，key的hash值不一樣但模數相同，此時查詢的時候會在比較第一次key的hash值（每一個索引項保存了key的hash值），過濾掉hash值不想等的狀況。第二種，hash值相等key不想等，出於性能的考慮衝突的檢測放到客戶端處理（key的原始值是存儲在消息文件中的，避免對數據文件的解析），客戶端比較一次消息體的key是否相同

5.存儲，爲了節省空間索引項中存儲的時間是時間差值（存儲時間——開始時間，開始時間存儲在索引文件頭中），整個索引文件是定長的，結構也是固定的

4.服務器消息過濾

RocketMQ的消息過濾方式有別於其餘的消息中間件，是在訂閱時，再作過濾，先來看下Consume Queue存儲結構

1.在Broker端進行Message Tag比較，先遍歷Consume Queue，若是存儲的Message Tag與訂閱的Message Tag不符合，則跳過，繼續比對下一個，符合則傳輸給Consumer。注意Message Tag是字符串形式，Consume Queue中存儲的是其對應的hashcode，比對時也是比對hashcode

2.Consumer收到過濾消息後，一樣也要執行在broker端的操做，可是比對的是真實的Message Tag字符串，而不是hashcode

爲何過濾要這麼作？

1.Message Tag存儲hashcode，是爲了在Consume Queue定長方式存儲，節約空間

2.過濾過程當中不會訪問Commit Log 數據，能夠保證堆積狀況下也能高效過濾

3.即便存在hash衝突，也能夠在Consumer端進行修正，保證萬無一失

5.單個JVM進程也能利用機器超大內存

1.Producer發送消息，消息從socket進入java 堆

2.Producer發送消息，消息從java堆進入pagecache，物理內存

3.Producer發送消息，由異步線程刷盤，消息從pagecache刷入磁盤

4.Consumer拉消息（正常消費），消息直接從pagecache（數據在物理內存）轉入socket，到達Consumer，不通過java堆。這種消費場景最多，線上96G物理內存，按照1K消息算，能夠物理緩存1億條消息

5.Consumer拉消息（異常消費），消息直接從pagecache轉入socket

6.Consumer拉消息（異常消費），因爲socket訪問了虛擬內存，產生缺頁中斷，此時會產生磁盤IO，從磁盤Load消息到pagecache，而後直接從socket發出去

7.同5

8.同6

6.消息堆積問題解決辦法

1 消息的堆積容量、依賴磁盤大小

2 發消息的吞吐量大小受影響程度、無Slave狀況，會受必定影響、有Slave狀況，不受影響

3 正常消費的Consumer是否會受影響、無Slave狀況，會受必定影響、有Slave狀況，不受影響

4 訪問堆積在磁盤的消息時，吞吐量有多大、與訪問的併發有關，最終會降到5000左右

在有Slave狀況下，Master一旦發現Consumer訪問堆積在磁盤的數據時，回想Consumer下達一個重定向指令，令Consumer從Slave拉取數據，這樣正常的發消息與正常的消費不會由於堆積受影響，由於系統將堆積場景與非堆積場景分割在了兩個不一樣的節點處理。這裏會產生一個問題，Slave會不會寫性能降低，答案是否認的。由於Slave的消息寫入只追求吞吐量，不追求實時性，只要總體的吞吐量高就好了，而Slave每次都是從Master拉取一批數據，如1M，這種批量順序寫入方式使堆積狀況，總體吞吐量影響相對較小，只是寫入RT會變長。

服務端安裝部署

我是在虛擬機中的CentOS6.5中進行部署。

1.下載程序

2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解壓到適當的目錄如/opt/目錄

3.啓動RocketMQ：進入rocketmq/bin 目錄 執行

4.啓動Broker，設置對應的NameServer

編寫客戶端

能夠查看sameple中的quickstart源碼 1.Consumer 消息消費者

2.Producer消息生產者

3.首先運行Consumer程序，一直在運行狀態接收服務器端推送過來的消息

4.再次運行Producer程序，生成消息併發送到Broker，Producer的日誌衝沒了，可是能夠看到Broker推送到Consumer的一條消息

Consumer最佳實踐

1.消費過程要作到冪等（即消費端去重）

RocketMQ沒法作到消息重複，因此若是業務對消息重複很是敏感，務必要在業務層面去重，有如下一些方式：

（1）.將消息的惟一鍵，能夠是MsgId，也能夠是消息內容中的惟一標識字段，例如訂單ID，消費以前判斷是否在DB或Tair（全局KV存儲）中存在，若是不存在則插入，並消費，不然跳過。（實踐過程要考慮原子性問題，判斷是否存在能夠嘗試插入，若是報主鍵衝突，則插入失敗，直接跳過） msgid必定是全局惟一的標識符，可是可能會存在一樣的消息有兩個不一樣的msgid的狀況（有多種緣由），這種狀況可能會使業務上重複，建議最好使用消息體中的惟一標識字段去重

（2）.使業務層面的狀態機去重

2.批量方式消費

若是業務流程支持批量方式消費，則能夠很大程度上的提升吞吐量，能夠經過設置Consumer的consumerMessageBatchMaxSize參數，默認是1，即一次消費一條參數

3.跳過非重要的消息

發生消息堆積時，若是消費速度一直跟不上發送速度，能夠選擇丟棄不重要的消息

如以上代碼所示，當某個隊列的消息數堆積到 100000 條以上，則嘗試丟棄部分或所有消息，這樣就能夠快速追上發送消息的速度

4.優化沒條消息消費過程

舉例以下，某條消息的消費過程以下

1. 根據消息從 DB 查詢數據 1
2. 根據消息從 DB 查詢數據2
3. 複雜的業務計算
4. 向 DB 插入數據3
5. 向 DB 插入數據 4

這條消息的消費過程與 DB 交互了 4 次，若是按照每次 5ms 計算，那麼總共耗時 20ms，假設業務計算耗時 5ms，那麼總過耗時 25ms，若是能把 4 次 DB 交互優化爲 2 次，那麼總耗時就能夠優化到 15ms，也就是說整體性能提升了 40%。

對於 Mysql 等 DB，若是部署在磁盤，那麼與 DB 進行交互，若是數據沒有命中 cache，每次交互的 RT 會直線上升， 若是採用 SSD，則 RT 上升趨勢要明顯好於磁盤。

個別應用可能會遇到這種狀況：在線下壓測消費過程當中，db 表現很是好，每次 RT 都很短，可是上線運行一段時間，RT 就會變長，消費吞吐量直線降低

主要緣由是線下壓測時間太短，線上運行一段時間後，cache 命中率降低，那麼 RT 就會增長。建議在線下壓測時，要測試足夠長時間，儘量模擬線上環境，壓測過程當中，數據的分佈也很重要，數據不一樣，可能 cache 的命中率也會徹底不一樣

Producer最佳實踐

1.發送消息注意事項

（1） 一個應用盡量用一個 Topic，消息子類型用 tags 來標識，tags 能夠由應用自由設置。只有發送消息設置了tags，消費方在訂閱消息時，才能夠利用 tags 在 broker 作消息過濾。

（2）每一個消息在業務層面的惟一標識碼，要設置到 keys 字段，方便未來定位消息丟失問題。服務器會爲每一個消息建立索引（哈希索引），應用能夠經過 topic，key 來查詢這條消息內容，以及消息被誰消費。因爲是哈希索引，請務必保證 key 儘量惟一，這樣能夠避免潛在的哈希衝突。

（3）消息發送成功或者失敗，要打印消息日誌，務必要打印 sendresult 和 key 字段

（4）send 消息方法，只要不拋異常，就表明發送成功。可是發送成功會有多個狀態，在 sendResult 裏定義

SEND_OK：消息發送成功

FLUSH_DISK_TIMEOUT：消息發送成功，可是服務器刷盤超時，消息已經進入服務器隊列，只有此時服務器宕機，消息纔會丟失

FLUSH_SLAVE_TIMEOUT：消息發送成功，可是服務器同步到 Slave 時超時，消息已經進入服務器隊列，只有此時服務器宕機，消息纔會丟失

SLAVE_NOT_AVAILABLE：消息發送成功，可是此時 slave 不可用，消息已經進入服務器隊列，只有此時服務器宕機，消息纔會丟失。對於精確發送順序消息的應用，因爲順序消息的侷限性，可能會涉及到主備自動切換問題，因此若是sendresult 中的 status 字段不等於 SEND_OK，就應該嘗試重試。對於其餘應用，則沒有必要這樣

（5）對於消息不可丟失應用，務必要有消息重發機制

2.消息發送失敗處理

Producer 的 send 方法自己支持內部重試，重試邏輯以下：

（1） 至多重試 3 次

（2） 若是發送失敗，則輪轉到下一個 Broker

（3） 這個方法的總耗時時間不超過 sendMsgTimeout 設置的值，默認 10s因此，若是自己向 broker 發送消息產生超時異常，就不會再作重試

如：

若是調用 send 同步方法發送失敗，則嘗試將消息存儲到 db，由後臺線程定時重試，保證消息必定到達 Broker。

上述 db 重試方式爲何沒有集成到 MQ 客戶端內部作，而是要求應用本身去完成，基於如下幾點考慮：

（1）MQ 的客戶端設計爲無狀態模式，方便任意的水平擴展，且對機器資源的消耗僅僅是 cpu、內存、網絡

（2）若是 MQ 客戶端內部集成一個 KV 存儲模塊，那麼數據只有同步落盤才能較可靠，而同步落盤自己性能開銷較大，因此一般會採用異步落盤，又因爲應用關閉過程不受 MQ 運維人員控制，可能常常會發生 kill -9 這樣暴力方式關閉，形成數據沒有及時落盤而丟失

（3）Producer 所在機器的可靠性較低，通常爲虛擬機，不適合存儲重要數據。 綜上，建議重試過程交由應用來控制。

3.選擇 oneway 形式發送

一個 RPC 調用，一般是這樣一個過程

（1）客戶端發送請求到服務器

（2）服務器處理該請求

（3）服務器向客戶端返回應答

因此一個 RPC 的耗時時間是上述三個步驟的總和，而某些場景要求耗時很是短，可是對可靠性要求並不高，例如日誌收集類應用，此類應用能夠採用 oneway 形式調用，oneway 形式只發送請求不等待應答，而發送請求在客戶端實現層面僅僅是一個 os 系統調用的開銷，即將數據寫入客戶端的 socket 緩衝區，此過程耗時一般在微秒級。

RocketMQ不止能夠直接推送消息，在消費端註冊監聽器進行監聽，還能夠由消費端決定本身去拉取數據

剛開始的沒有細看PullResult對象，覺得拉取到的結果沒有MessageExt對象還跑到羣裏面問別人，犯2了

特別要注意 靜態變量offsetTable的做用，拉取的是按照從offset（理解爲下標）位置開始拉取，拉取N條，offsetTable記錄下次拉取的offset位置。

在此我向你們推薦一個架構學習交流羣。交流學習羣號：478030634 裏面會分享一些資深架構師錄製的視頻錄像：有Spring，MyBatis，Netty源碼分析，高併發、高性能、分佈式、微服務架構的原理，JVM性能優化、分佈式架構等這些成爲架構師必備的知識體系。還能領取免費的學習資源，目前受益良多

文章有點長，你們以爲做者總結的還能夠，你們能夠點擊下方二維碼進行關注。《Java爛豬皮》公衆號聊的不只僅是Java技術知識，還有面試等乾貨，後期還有大量架構乾貨。你們一塊兒關注吧！關注爛豬皮，你會了解的更多..............