摘抄自：RocketMQ 最佳實踐

最佳實踐

一廓八、生產(chǎn)者

1.1 發(fā)送消息注意事項

Tags的使用

一個應(yīng)用盡可能用一個Topic吱雏，而消息子類型則可以用tags來標(biāo)識（在實際使用中徘公，實際上會一個場景用一個topic）。tags可以由應(yīng)用自由設(shè)置，只有生產(chǎn)者在發(fā)送消息設(shè)置了tags拐叉，消費方在訂閱消息時才可以利用tags通過broker做消息過濾：message.setTags("TagA")。

Keys的使用

每個消息在業(yè)務(wù)層面的唯一標(biāo)識碼要設(shè)置到keys字段，方便將來定位消息丟失問題。服務(wù)器會為每個消息創(chuàng)建索引（哈希索引）着憨，應(yīng)用可以通過topic财饥、key來查詢這條消息內(nèi)容换吧，以及消息被誰消費。由于是哈希索引钥星，請務(wù)必保證key盡可能唯一沾瓦，這樣可以避免潛在的哈希沖突。

   // 訂單Id   
   String orderId = "20034568923546";   
   message.setKeys(orderId);   

日志的打印

消息發(fā)送成功或者失敗要打印消息日志，務(wù)必要打印SendResult和key字段贯莺。send消息方法只要不拋異常风喇，就代表發(fā)送成功。發(fā)送成功會有多個狀態(tài)缕探，在sendResult里定義魂莫。以下對每個狀態(tài)進行說明：

• SEND_OK

消息發(fā)送成功。要注意的是消息發(fā)送成功也不意味著它是可靠的爹耗。要確保不會丟失任何消息耙考，還應(yīng)啟用同步Master服務(wù)器或同步刷盤，即SYNC_MASTER或SYNC_FLUSH潭兽。

• FLUSH_DISK_TIMEOUT

消息發(fā)送成功但是服務(wù)器刷盤超時倦始。此時消息已經(jīng)進入服務(wù)器隊列（內(nèi)存），只有服務(wù)器宕機山卦，消息才會丟失楣号。消息存儲配置參數(shù)中可以設(shè)置刷盤方式和同步刷盤時間長度，如果Broker服務(wù)器設(shè)置了刷盤方式為同步刷盤怒坯，即FlushDiskType=SYNC_FLUSH（默認為異步刷盤方式）炫狱，當(dāng)Broker服務(wù)器未在同步刷盤時間內(nèi)（默認為5s）完成刷盤，則將返回該狀態(tài)——刷盤超時剔猿。

• FLUSH_SLAVE_TIMEOUT

消息發(fā)送成功视译，但是服務(wù)器同步到Slave時超時。此時消息已經(jīng)進入服務(wù)器隊列归敬，只有服務(wù)器宕機酷含，消息才會丟失。如果Broker服務(wù)器的角色是同步Master汪茧，即SYNC_MASTER（默認是異步Master即ASYNC_MASTER）椅亚，并且從Broker服務(wù)器未在同步刷盤時間（默認為5秒）內(nèi)完成與主服務(wù)器的同步，則將返回該狀態(tài)——數(shù)據(jù)同步到Slave服務(wù)器超時舱污。

• SLAVE_NOT_AVAILABLE

消息發(fā)送成功呀舔，但是此時Slave不可用。如果Broker服務(wù)器的角色是同步Master扩灯，即SYNC_MASTER（默認是異步Master服務(wù)器即ASYNC_MASTER）媚赖，但沒有配置slave Broker服務(wù)器，則將返回該狀態(tài)——無Slave服務(wù)器可用珠插。

1.2 消息發(fā)送失敗處理方式

Producer的send方法本身支持內(nèi)部重試惧磺，重試邏輯如下：

• 至多重試2次。
• 如果同步模式發(fā)送失敗捻撑，則輪轉(zhuǎn)到下一個Broker磨隘，如果異步模式發(fā)送失敗缤底，則只會在當(dāng)前Broker進行重試。這個方法的總耗時時間不超過sendMsgTimeout設(shè)置的值番捂，默認10s训堆。
• 如果本身向broker發(fā)送消息產(chǎn)生超時異常，就不會再重試白嘁。

以上策略也是在一定程度上保證了消息可以發(fā)送成功。如果業(yè)務(wù)對消息可靠性要求比較高膘流，建議應(yīng)用增加相應(yīng)的重試邏輯：比如調(diào)用send同步方法發(fā)送失敗時絮缅，則嘗試將消息存儲到db，然后由后臺線程定時重試呼股，確保消息一定到達Broker耕魄。

上述db重試方式為什么沒有集成到MQ客戶端內(nèi)部做，而是要求應(yīng)用自己去完成彭谁，主要基于以下幾點考慮：首先吸奴，MQ的客戶端設(shè)計為無狀態(tài)模式，方便任意的水平擴展缠局，且對機器資源的消耗僅僅是cpu则奥、內(nèi)存、網(wǎng)絡(luò)狭园。其次读处，如果MQ客戶端內(nèi)部集成一個KV存儲模塊，那么數(shù)據(jù)只有同步落盤才能較可靠唱矛，而同步落盤本身性能開銷較大罚舱，所以通常會采用異步落盤，又由于應(yīng)用關(guān)閉過程不受MQ運維人員控制绎谦，可能經(jīng)常會發(fā)生 kill -9 這樣暴力方式關(guān)閉管闷，造成數(shù)據(jù)沒有及時落盤而丟失。第三窃肠，Producer所在機器的可靠性較低包个，一般為虛擬機，不適合存儲重要數(shù)據(jù)冤留。綜上赃蛛，建議重試過程交由應(yīng)用來控制。

1.3 選擇oneway形式發(fā)送

通常消息的發(fā)送是這樣一個過程：

• 客戶端發(fā)送請求到服務(wù)器
• 服務(wù)器處理請求
• 服務(wù)器向客戶端返回應(yīng)答

所以搀菩，一次消息發(fā)送的耗時時間是上述三個步驟的總和呕臂，而某些場景要求耗時非常短，但是對可靠性要求并不高肪跋，例如日志收集類應(yīng)用歧蒋，此類應(yīng)用可以采用oneway形式調(diào)用，oneway形式只發(fā)送請求不等待應(yīng)答，而發(fā)送請求在客戶端實現(xiàn)層面僅僅是一個操作系統(tǒng)系統(tǒng)調(diào)用的開銷谜洽，即將數(shù)據(jù)寫入客戶端的socket緩沖區(qū)萝映，此過程耗時通常在微秒級。

二阐虚、消費者

2.1 消費過程冪等

RocketMQ 無法避免消息重復(fù)（Exactly-Once）序臂，所以如果業(yè)務(wù)對消費重復(fù)非常敏感，務(wù)必要在業(yè)務(wù)層面進行去重處理实束“赂眩可以借助關(guān)系數(shù)據(jù)庫進行去重。首先需要確定消息的唯一鍵咸灿，可以是 msgId构订，也可以是消息內(nèi)容中的唯一標(biāo)識字段，例如訂單 Id 等避矢。在消費之前判斷唯一鍵是否在關(guān)系數(shù)據(jù)庫中存在悼瘾。如果不存在則插入，并消費审胸，否則跳過亥宿。（實際過程要考慮原子性問題，判斷是否存在可以嘗試插入砂沛，如果報主鍵沖突箩绍，則插入失敗，直接跳過）

msgId 一定是全局唯一標(biāo)識符尺上，但是實際使用中材蛛，可能會存在相同的消息有兩個不同 msgId 的情況（消費者主動重發(fā)、因客戶端重投機制導(dǎo)致的重復(fù)等）怎抛，這種情況就需要使業(yè)務(wù)字段進行重復(fù)消費卑吭。

冪等的設(shè)計
新增、查詢马绝、更新豆赏、刪除四種操作，查詢和刪除天然是冪等的富稻。冪等實際上就是針對新增和更新來講的掷邦，當(dāng)然在實際業(yè)務(wù)中新增和更新根據(jù)業(yè)務(wù)要求也不一定需要冪等。
冪等的設(shè)計實際上可以看做是鎖的設(shè)計椭赋。冪等設(shè)計的關(guān)鍵是鎖的key抚岗。
新增：幾種方案，

1. 如上所述哪怔，DB方案：msgId持久化到DB宣蔚，DB先查詢msgId看是否存在向抢，不存在則繼續(xù)消費，適用于并發(fā)度不高的情況胚委。并發(fā)度高的情況下挟鸠，考慮原子性，可以以msgId做為唯一主鍵亩冬，艘希，msgId持久化時成功，則消費消息硅急。實際上這也是一種鎖覆享。
2. 對于對響應(yīng)要求更高的系統(tǒng)，可以使用 redis 分布式鎖設(shè)計铜秆，設(shè)置msgId成功，則消費消息

更新：對于消息來講讶迁，沒有更新连茧。下訂單后扣庫存，對于庫存的更新這種是需要做冪等的巍糯。更新的冪等性可以使用樂觀鎖（版本號啸驯，防止ABA）、redis分布式鎖祟峦、有限狀態(tài)機等罚斗。

2.2 消費速度慢的處理方式

提高消費并行度

絕大部分消息消費行為都屬于 IO 密集型，即可能是操作數(shù)據(jù)庫宅楞，或者調(diào)用 RPC针姿，這類消費行為的消費速度在于后端數(shù)據(jù)庫或者外系統(tǒng)的吞吐量，通過增加消費并行度厌衙，可以提高總的消費吞吐量距淫，但是并行度增加到一定程度，反而會下降婶希。所以榕暇，應(yīng)用必須要設(shè)置合理的并行度。 如下有幾種修改消費并行度的方法：

• 同一個 ConsumerGroup 下喻杈，通過增加 Consumer 實例數(shù)量來提高并行度（需要注意的是超過訂閱隊列數(shù)的 Consumer 實例無效）彤枢。可以通過加機器筒饰，或者在已有機器啟動多個進程的方式缴啡。
• 提高單個 Consumer 的消費并行線程，通過修改參數(shù) consumeThreadMin瓷们、consumeThreadMax實現(xiàn)盟猖。

批量方式消費

某些業(yè)務(wù)流程如果支持批量方式消費讨衣，則可以很大程度上提高消費吞吐量，例如訂單扣款類應(yīng)用式镐，一次處理一個訂單耗時 1 s反镇，一次處理 10 個訂單可能也只耗時 2 s，這樣即可大幅度提高消費的吞吐量娘汞，通過設(shè)置 consumer 的 consumeMessageBatchMaxSize 返個參數(shù)歹茶，默認是 1，即一次只消費一條消息你弦，例如設(shè)置為 N惊豺，那么每次消費的消息數(shù)小于等于 N。

有一種場景：假設(shè)是順序消息禽作，批量消費時尸昧，還可以對消息進行去重處理，減少消息消費消耗旷偿。

跳過非重要消息

發(fā)生消息堆積時烹俗，如果消費速度一直追不上發(fā)送速度，如果業(yè)務(wù)對數(shù)據(jù)要求不高的話萍程，可以選擇丟棄不重要的消息幢妄。例如，當(dāng)某個隊列的消息數(shù)堆積到100000條以上茫负，則嘗試丟棄部分或全部消息蕉鸳，這樣就可以快速追上發(fā)送消息的速度。示例代碼如下：

    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        long offset = msgs.get(0).getQueueOffset();
        String maxOffset = msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET);
        long diff = Long.parseLong(maxOffset) - offset;
        if (diff > 100000) {
            // TODO 消息堆積情況的特殊處理
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        // TODO 正常消費過程
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }    

在實際使用中忍法，有一個經(jīng)典場景：數(shù)據(jù)中心或者指標(biāo)中心的構(gòu)造潮尝，通常會有一個全量的離線任務(wù)在跑（T+1或者T+1H），另外為了保證準(zhǔn)實時饿序，各個指標(biāo)源還會發(fā)送一些實時變更消息衍锚，指標(biāo)中心接收到這些消息后，進行指標(biāo)的實時更新嗤堰。
通常如果這些實時變更消息發(fā)生了嚴(yán)重的積壓時戴质，會考慮直接跑一次全量（或者各種小全量），而積壓的消息直接丟棄踢匣。

優(yōu)化每條消息消費過程

舉例如下告匠，某條消息的消費過程如下：

• 根據(jù)消息從 DB 查詢【數(shù)據(jù) 1】
• 根據(jù)消息從 DB 查詢【數(shù)據(jù) 2】
• 復(fù)雜的業(yè)務(wù)計算
• 向 DB 插入【數(shù)據(jù) 3】
• 向 DB 插入【數(shù)據(jù) 4】

這條消息的消費過程中有4次與 DB 的 交互，如果按照每次 5ms 計算离唬，那么總共耗時 20ms后专，假設(shè)業(yè)務(wù)計算耗時 5ms，那么總過耗時 25ms输莺，所以如果能把 4 次 DB 交互優(yōu)化為 2 次戚哎，那么總耗時就可以優(yōu)化到 15ms裸诽，即總體性能提高了 40%。所以應(yīng)用如果對時延敏感的話型凳，可以把DB部署在SSD硬盤丈冬，相比于SCSI磁盤，前者的RT會小很多甘畅。

2.3 消費打印日志

如果消息量較少埂蕊，建議在消費入口方法打印消息，消費耗時等疏唾，方便后續(xù)排查問題蓄氧。

新版的 RocketMQ 提供了消息軌跡功能，開啟后也能進行消息的追蹤

   public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
        log.info("RECEIVE_MSG_BEGIN: " + msgs.toString());
        // TODO 正常消費過程
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }   

如果能打印每條消息消費耗時槐脏，那么在排查消費慢等線上問題時喉童，會更方便。

2.4 其他消費建議

關(guān)于消費者和訂閱

第一件需要注意的事情是顿天，不同的消費者組可以獨立的消費一些 topic堂氯，并且每個消費者組都有自己的消費偏移量，請確保 同一組內(nèi)的每個消費者訂閱信息保持一致露氮。

關(guān)于有序消息

消費者將鎖定每個消息隊列祖灰，以確保他們被逐個消費钟沛，雖然這將會導(dǎo)致性能下降畔规，但是當(dāng)你關(guān)心消息順序的時候會很有用。我們不建議拋出異常恨统，你可以返回 ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT 作為替代叁扫。

關(guān)于并發(fā)消費

顧名思義，消費者將并發(fā)消費這些消息畜埋，建議你使用它來獲得良好性能莫绣，我們不建議拋出異常，你可以返回 ConsumeConcurrentlyStatus.RECONSUME_LATER 作為替代悠鞍。

關(guān)于消費狀態(tài)Consume Status

對于并發(fā)的消費監(jiān)聽器对室，你可以返回 RECONSUME_LATER 來通知消費者現(xiàn)在不能消費這條消息，并且希望可以稍后重新消費它咖祭。然后掩宜，你可以繼續(xù)消費其他消息。對于有序的消息監(jiān)聽器么翰，因為你關(guān)心它的順序牺汤，所以不能跳過消息，但是你可以返回SUSPEND_CURRENT_QUEUE_A_MOMENT 告訴消費者等待片刻浩嫌。

關(guān)于Blocking

不建議阻塞監(jiān)聽器檐迟，因為它會阻塞線程池补胚，并最終可能會終止消費進程

關(guān)于線程數(shù)設(shè)置

消費者使用 ThreadPoolExecutor 在內(nèi)部對消息進行消費，所以你可以通過設(shè)置 setConsumeThreadMin 或 setConsumeThreadMax 來改變它追迟。

關(guān)于消費位點

當(dāng)建立一個新的消費者組時溶其，需要決定是否需要消費已經(jīng)存在于 Broker 中的歷史消息CONSUME_FROM_LAST_OFFSET 將會忽略歷史消息，并消費之后生成的任何消息怔匣。CONSUME_FROM_FIRST_OFFSET 將會消費每個存在于 Broker 中的信息握联。你也可以使用 CONSUME_FROM_TIMESTAMP 來消費在指定時間戳后產(chǎn)生的消息。

三每瞒、Broker

3.1 Broker 角色

Broker 角色分為 ASYNC_MASTER（異步主機）金闽、SYNC_MASTER（同步主機）以及SLAVE（從機）。如果對消息的可靠性要求比較嚴(yán)格剿骨，可以采用 SYNC_MASTER加SLAVE的部署方式代芜。如果對消息可靠性要求不高，可以采用ASYNC_MASTER加SLAVE的部署方式浓利。如果只是測試方便挤庇，則可以選擇僅ASYNC_MASTER或僅SYNC_MASTER的部署方式。

3.2 FlushDiskType

SYNC_FLUSH（同步刷新）相比于ASYNC_FLUSH（異步處理）會損失很多性能贷掖，但是也更可靠嫡秕，所以需要根據(jù)實際的業(yè)務(wù)場景做好權(quán)衡。

3.3 Broker 配置

四、NameServer

RocketMQ 中窟哺，Name Servers 被設(shè)計用來做簡單的路由管理泻轰。其職責(zé)包括：

• Brokers 定期向每個 namesrv 注冊路由數(shù)據(jù)。
• namesrv 為客戶端且轨，包括生產(chǎn)者浮声，消費者和命令行客戶端提供最新的路由信息。

五旋奢、客戶端配置

相對于 RocketMQ 的 Broker 集群泳挥，生產(chǎn)者和消費者都是客戶端。本小節(jié)主要描述生產(chǎn)者和消費者公共的行為配置黄绩。

5.1 客戶端尋址方式

RocketMQ 可以令客戶端找到 namesrv, 然后通過 namesrv 再找到Broker羡洁。如下所示有多種配置方式，優(yōu)先級由高到低爽丹，高優(yōu)先級會覆蓋低優(yōu)先級筑煮。

• 代碼中指定 namesrv 地址辛蚊，多個 namesrv 地址之間用分號分割

producer.setNamesrvAddr("192.168.0.1:9876;192.168.0.2:9876");  

consumer.setNamesrvAddr("192.168.0.1:9876;192.168.0.2:9876");

• Java啟動參數(shù)中指定 namesrv 地址

-Drocketmq.namesrv.addr=192.168.0.1:9876;192.168.0.2:9876  

• 環(huán)境變量指定 namesrv 地址

export NAMESRV_ADDR=192.168.0.1:9876;192.168.0.2:9876   

• HTTP靜態(tài)服務(wù)器尋址（默認）

即一個簡易版的配置中心，通過這樣的方式真仲，不僅可以實現(xiàn)分鐘級的熱更新（定時 pull 模式）袋马，還可以將 namesrv 地址做一定程度的隱藏。

客戶端啟動后秸应，會定時訪問一個靜態(tài) HTTP 服務(wù)器虑凛，地址如下：http://jmenv.tbsite.net:8080/rocketmq/nsaddr，這個URL的返回內(nèi)容如下：

192.168.0.1:9876;192.168.0.2:9876   

客戶端默認每隔2分鐘訪問一次這個 HTTP 服務(wù)器软啼，并更新本地的 namesrv 地址桑谍。URL 已經(jīng)在代碼中硬編碼，可通過修改 /etc/hosts 文件來改變要訪問的服務(wù)器祸挪，例如在 /etc/hosts 增加如下配置：

10.232.22.67    jmenv.tbsite.net   

推薦使用 HTTP 靜態(tài)服務(wù)器尋址方式锣披，好處是客戶端部署簡單，且 namesrv 集群可以熱升級贿条。

5.2 客戶端配置

DefaultMQProducer雹仿、TransactionMQProducer、DefaultMQPushConsumer整以、DefaultMQPullConsumer 都繼承于 ClientConfig 類胧辽，ClientConfig 為客戶端的公共配置類」冢客戶端的配置都是get邑商、set形式，每個參數(shù)都可以用spring來配置帆调，也可以在代碼中配置奠骄，例如 namesrvAddr 這個參數(shù)可以這樣配置豆同，producer.setNamesrvAddr("192.168.0.1:9876")番刊，其他參數(shù)同理。

客戶端的公共配置

Producer配置

PushConsumer 配置

PullConsumer配置

Message數(shù)據(jù)結(jié)構(gòu)

六、系統(tǒng)配置

本小節(jié)主要介紹系統(tǒng)（JVM/OS）相關(guān)的配置沃缘。

6.1 JVM選項

推薦使用最新發(fā)布的JDK 1.8版本躯枢。通過設(shè)置相同的 Xms 和 Xmx 值來防止 JVM 調(diào)整堆大小以獲得更好的性能。簡單的 JVM 配置如下所示：

-server -Xms8g -Xmx8g -Xmn4g   

如果您不關(guān)心 Broker 的啟動時間槐臀，還有一種更好的選擇锄蹂，就是通過“預(yù)觸摸” Java 堆以確保在 JVM 初始化期間每個頁面都將被分配。那些不關(guān)心啟動時間的人可以啟用它：
-XX:+AlwaysPreTouch
禁用偏置鎖定可能會減少JVM暫停水慨，
-XX:-UseBiasedLocking
至于垃圾回收得糜，建議使用帶JDK 1.8的G1收集器。

-XX:+UseG1GC 
-XX:G1HeapRegionSize=16m   
-XX:G1ReservePercent=25 
-XX:InitiatingHeapOccupancyPercent=30

這些GC選項看起來有點激進晰洒，但事實證明它在我們的生產(chǎn)環(huán)境中具有良好的性能朝抖。另外不要把 -XX:MaxGCPauseMillis 的值設(shè)置太小，否則 JVM 將使用一個小的年輕代來實現(xiàn)這個目標(biāo)谍珊，這將導(dǎo)致非常頻繁的minor GC治宣，所以 建議使用 rolling GC 日志文件：

-XX:+UseGCLogFileRotation   
-XX:NumberOfGCLogFiles=5 
-XX:GCLogFileSize=30m

如果寫入GC文件會增加代理的延遲，可以考慮將GC日志文件重定向到內(nèi)存文件系統(tǒng)：

-Xloggc:/dev/shm/mq_gc_%p.log123   

6.2 Linux內(nèi)核參數(shù)

os.sh 腳本在 bin 文件夾中列出了許多內(nèi)核參數(shù)砌滞，可以進行微小的更改然后用于生產(chǎn)用途侮邀。下面的參數(shù)需要注意，更多細節(jié)請參考/proc/sys/vm/*的 文檔

• vm.extra_free_kbytes贝润，告訴VM在后臺回收（kswapd）啟動的閾值與直接回收（通過分配進程）的閾值之間保留額外的可用內(nèi)存豌拙。RocketMQ 使用此參數(shù)來避免內(nèi)存分配中的長延遲。（與具體內(nèi)核版本相關(guān)）
• vm.min_free_kbytes题暖，如果將其設(shè)置為低于 1024KB按傅，將會巧妙的將系統(tǒng)破壞，并且系統(tǒng)在高負載下容易出現(xiàn)死鎖胧卤。
• vm.max_map_count唯绍，限制一個進程可能具有的最大內(nèi)存映射區(qū)域數(shù)。RocketMQ 將使用 mmap 加載 CommitLog 和 ConsumeQueue枝誊，因此建議將為此參數(shù)設(shè)置較大的值况芒。（agressiveness --> aggressiveness）
• vm.swappiness，定義內(nèi)核交換內(nèi)存頁面的積極程度叶撒。較高的值會增加攻擊性绝骚，較低的值會減少交換量。建議將值設(shè)置為10來避免交換延遲祠够。
• File descriptor limits压汪，RocketMQ需要為文件（CommitLog和ConsumeQueue）和網(wǎng)絡(luò)連接打開文件描述符。我們建議設(shè)置文件描述符的值為655350古瓤。
• Disk scheduler止剖，RocketMQ 建議使用 I/O 截止時間調(diào)度器腺阳，它試圖為請求提供有保證的延遲。