RocketMQ是一款分布式留凭、隊列模型的消息中間件，具有以下特點：
能夠保證嚴(yán)格的消息順序
提供豐富的消息拉取模式
高效的訂閱者水平擴展能力
實時的消息訂閱機制
億級消息堆積能力
RocketMQ網(wǎng)絡(luò)部署特
（1）NameServer是一個幾乎無狀態(tài)的節(jié)點缝驳，可集群部署，節(jié)點之間無任何信息同步
（2）Broker部署相對復(fù)雜，Broker氛圍Master與Slave到逊，一個Master可以對應(yīng)多個Slaver，但是一個Slaver只能對應(yīng)一個Master滤钱，Master與Slaver的對應(yīng)關(guān)系通過指定相同的BrokerName觉壶，不同的BrokerId來定義，BrokerId為0表示Master件缸，非0表示Slaver铜靶。Master可以部署多個。每個Broker與NameServer集群中的所有節(jié)點建立長連接他炊，定時注冊Topic信息到所有的NameServer
（3）Producer與NameServer集群中的其中一個節(jié)點（隨機選擇）建立長連接争剿，定期從NameServer取Topic路由信息，并向提供Topic服務(wù)的Master建立長連接痊末，且定時向Master發(fā)送心跳蚕苇。Produce完全無狀態(tài)，可集群部署
（4）Consumer與NameServer集群中的其中一個節(jié)點（隨機選擇）建立長連接凿叠，定期從NameServer取Topic路由信息捆蜀，并向提供Topic服務(wù)的Master、Slaver建立長連接幔嫂，且定時向Master辆它、Slaver發(fā)送心跳。Consumer即可從Master訂閱消息履恩，也可以從Slave訂閱消息锰茉，訂閱規(guī)則由Broker配置決定
RocketMQ儲存特點
（1）零拷貝原理：Consumer消費消息過程，使用了零拷貝切心，零拷貝包括一下2中方式飒筑，RocketMQ使用第一種方式片吊，因小塊數(shù)據(jù)傳輸?shù)囊笮Ч萻endfile方式好
a )使用mmap+write方式
優(yōu)點：即使頻繁調(diào)用，使用小文件塊傳輸协屡，效率也很高
缺點：不能很好的利用DMA方式俏脊，會比sendfile多消耗CPU資源，內(nèi)存安全性控制復(fù)雜肤晓，需要避免JVM Crash問題
b）使用sendfile方式
優(yōu)點：可以利用DMA方式爷贫，消耗CPU資源少，大塊文件傳輸效率高补憾，無內(nèi)存安全新問題
缺點：小塊文件效率低于mmap方式漫萄，只能是BIO方式傳輸，不能使用NIO
（2）數(shù)據(jù)存儲結(jié)構(gòu)
RocketMQ關(guān)鍵特性
1.單機支持1W以上的持久化隊
（1）所有數(shù)據(jù)單獨儲存到commit Log 盈匾，完全順序?qū)懱谖瘢S機讀
（2）對最終用戶展現(xiàn)的隊列實際只儲存消息在Commit Log 的位置信息，并且串行方式刷盤
這樣做的好處：
（1）隊列輕量化削饵，單個隊列數(shù)據(jù)量非常少
（2）對磁盤的訪問串行話岩瘦，避免磁盤競爭，不會因為隊列增加導(dǎo)致IOWait增高
每個方案都有優(yōu)缺點窿撬，他的缺點是：
（1）寫雖然是順序?qū)懫裘粒亲x卻變成了隨機讀
（2）讀一條消息，會先讀Consume Queue尤仍，再讀Commit Log箫津，增加了開銷
（3）要保證Commit Log 與 Consume Queue完全的一致，增加了編程的復(fù)雜度
以上缺點如何客服：
（1）隨機讀宰啦，盡可能讓讀命中pagecache苏遥，減少IO操作，所以內(nèi)存越大越好赡模。如果系統(tǒng)中堆積的消息過多田炭，讀數(shù)據(jù)要訪問硬盤會不會由于隨機讀導(dǎo)致系統(tǒng)性能急劇下降，答案是否定的漓柑。
a）訪問pagecache時教硫，即使只訪問1K的消息，系統(tǒng)也會提前預(yù)讀出更多的數(shù)據(jù)辆布，在下次讀時就可能命中pagecache
b）隨機訪問Commit Log 磁盤數(shù)據(jù)瞬矩，系統(tǒng)IO調(diào)度算法設(shè)置為NOOP方式，會在一定程度上將完全的隨機讀變成順序跳躍方式锋玲，而順序跳躍方式讀較完全的隨機讀性能高5倍
（2）由于Consume Queue存儲數(shù)量極少景用，而且順序讀，在pagecache的與讀取情況下惭蹂，Consume Queue的讀性能與內(nèi)存幾乎一直伞插，即使堆積情況下割粮。所以可以認(rèn)為Consume Queue完全不會阻礙讀性能
（3）Commit Log中存儲了所有的元信息，包含消息體媚污，類似于MySQl舀瓢、Oracle的redolog，所以只要有Commit Log存在耗美， Consume Queue即使丟失數(shù)據(jù)京髓，仍可以恢復(fù)出來
2.刷盤策略
rocketmq中的所有消息都是持久化的，先寫入系統(tǒng)pagecache幽歼，然后刷盤朵锣，可以保證內(nèi)存與磁盤都有一份數(shù)據(jù)谬盐，訪問時甸私，可以直接從內(nèi)存讀取
2.1異步刷盤
在有 RAID 卡， SAS 15000 轉(zhuǎn)磁盤測試順序?qū)懳募煽俣瓤梢赃_到 300M 每秒左右皇型，而線上的網(wǎng)卡一般都為千兆網(wǎng)卡，寫磁盤速度明顯快于數(shù)據(jù)網(wǎng)絡(luò)入口速度砸烦，那么是否可以做到寫完 內(nèi)存就向用戶返回弃鸦，由后臺線程刷盤呢？
(1). 由于磁盤速度大于網(wǎng)卡速度幢痘，那么刷盤的進度肯定可以跟上消息的寫入速度唬格。
(2). 萬一由于此時系統(tǒng)壓力過大，可能堆積消息颜说，除了寫入 IO购岗，還有讀取 IO，萬一出現(xiàn)磁盤讀取落后情況门粪，會不會導(dǎo)致系統(tǒng)內(nèi)存溢出喊积，答案是否定的，原因如下：
a) 寫入消息到 PAGECACHE 時玄妈，如果內(nèi)存不足乾吻，則嘗試丟棄干凈的 PAGE，騰出內(nèi)存供新消息使用拟蜻，策略是 LRU 方式绎签。
b) 如果干凈頁不足，此時寫入 PAGECACHE 會被阻塞酝锅，系統(tǒng)嘗試刷盤部分?jǐn)?shù)據(jù)诡必，大約每次嘗試 32 個 PAGE，來找出更多干凈 PAGE屈张。
綜上擒权，內(nèi)存溢出的情況不會出現(xiàn)
2.2同步刷盤：
同步刷盤與異步刷盤的唯一區(qū)別是異步刷盤寫完 PAGECACHE 直接返回袱巨，而同步刷盤需要等待刷盤完成才返回，同步刷盤流程如下：
（1）寫入 PAGECACHE 后碳抄，線程等待愉老，通知刷盤線程刷盤。
（2）刷盤線程刷盤后剖效，喚醒前端等待線程嫉入，可能是一批線程。
（3）前端等待線程向用戶返回成功璧尸。
3.消息查詢
3.1按照MessageId查詢消息
MsgId總共16個字節(jié)咒林，包含消息儲存主機地址，消息Commit Log Offset爷光。從MsgId中解析出Broker的地址和Commit Log 偏移地址垫竞，然后按照存儲格式所在位置消息buffer解析成一個完整消息
3.2按照Message Key查詢消息
1.根據(jù)查詢的key的hashcode%slotNum得到具體的槽位置 （slotNum是一個索引文件里面包含的最大槽目數(shù)目，例如圖中所示slotNum=500W）
2.根據(jù)slotValue（slot對應(yīng)位置的值）查找到索引項列表的最后一項（倒序排列蛀序，slotValue總是指向最新的一個索引項）
3.遍歷索引項列表返回查詢時間范圍內(nèi)的結(jié)果集（默認(rèn)一次最大返回的32條記錄）
4.Hash沖突欢瞪，尋找key的slot位置時相當(dāng)于執(zhí)行了兩次散列函數(shù)，一次key的hash徐裸，一次key的hash取值模遣鼓，因此這里存在兩次沖突的情況；第一種重贺，key的hash值不同但模數(shù)相同骑祟，此時查詢的時候會在比較第一次key的hash值（每個索引項保存了key的hash值），過濾掉hash值不想等的情況气笙。第二種次企，hash值相等key不想等，出于性能的考慮沖突的檢測放到客戶端處理（key的原始值是存儲在消息文件中的健民，避免對數(shù)據(jù)文件的解析）抒巢，客戶端比較一次消息體的key是否相同
5.存儲，為了節(jié)省空間索引項中存儲的時間是時間差值（存儲時間——開始時間秉犹，開始時間存儲在索引文件頭中）蛉谜，整個索引文件是定長的，結(jié)構(gòu)也是固定的
4.服務(wù)器消息過濾
RocketMQ的消息過濾方式有別于其他的消息中間件崇堵，是在訂閱時型诚，再做過濾，先來看下Consume Queue存儲結(jié)構(gòu)
1.在Broker端進行Message Tag比較鸳劳，先遍歷Consume Queue狰贯，如果存儲的Message Tag與訂閱的Message Tag不符合，則跳過，繼續(xù)比對下一個涵紊，符合則傳輸給Consumer傍妒。注意Message Tag是字符串形式，Consume Queue中存儲的是其對應(yīng)的hashcode摸柄，比對時也是比對hashcode
2.Consumer收到過濾消息后颤练，同樣也要執(zhí)行在broker端的操作，但是比對的是真實的Message Tag字符串，而不是hashcode
為什么過濾要這么做？
1.Message Tag存儲hashcode合武，是為了在Consume Queue定長方式存儲，節(jié)約空間
2.過濾過程中不會訪問Commit Log 數(shù)據(jù)宇挫，可以保證堆積情況下也能高效過濾
3.即使存在hash沖突，也可以在Consumer端進行修正酪术，保證萬無一失
5.單個JVM進程也能利用機器超大內(nèi)存
1.Producer發(fā)送消息器瘪，消息從socket進入java 堆
2.Producer發(fā)送消息，消息從java堆進入pagecache拼缝，物理內(nèi)存
3.Producer發(fā)送消息娱局，由異步線程刷盤彰亥，消息從pagecache刷入磁盤
4.Consumer拉消息（正常消費）咧七，消息直接從pagecache（數(shù)據(jù)在物理內(nèi)存）轉(zhuǎn)入socket，到達Consumer任斋，不經(jīng)過java堆继阻。這種消費場景最多，線上96G物理內(nèi)存废酷，按照1K消息算瘟檩，可以物理緩存1億條消息
5.Consumer拉消息（異常消費），消息直接從pagecache轉(zhuǎn)入socket
6.Consumer拉消息（異常消費）澈蟆，由于socket訪問了虛擬內(nèi)存墨辛，產(chǎn)生缺頁中斷，此時會產(chǎn)生磁盤IO趴俘，從磁盤Load消息到pagecache睹簇，然后直接從socket發(fā)出去
7.同5
8.同6
6.消息堆積問題解決辦法
1 消息的堆積容量、依賴磁盤大小
2 發(fā)消息的吞吐量大小受影響程度寥闪、無Slave情況太惠，會受一定影響、有Slave情況疲憋，不受影響
3 正常消費的Consumer是否會受影響凿渊、無Slave情況，會受一定影響、有Slave情況埃脏，不受影響
4 訪問堆積在磁盤的消息時搪锣，吞吐量有多大、與訪問的并發(fā)有關(guān)彩掐，最終會降到5000左右
在有Slave情況下淤翔，Master一旦發(fā)現(xiàn)Consumer訪問堆積在磁盤的數(shù)據(jù)時，回想Consumer下達一個重定向指令佩谷，令Consumer從Slave拉取數(shù)據(jù)旁壮，這樣正常的發(fā)消息與正常的消費不會因為堆積受影響，因為系統(tǒng)將堆積場景與非堆積場景分割在了兩個不同的節(jié)點處理谐檀。這里會產(chǎn)生一個問題抡谐，Slave會不會寫性能下降，答案是否定的桐猬。因為Slave的消息寫入只追求吞吐量麦撵，不追求實時性，只要整體的吞吐量高就行了溃肪，而Slave每次都是從Master拉取一批數(shù)據(jù)免胃，如1M，這種批量順序?qū)懭敕绞绞苟逊e情況惫撰，整體吞吐量影響相對較小羔沙，只是寫入RT會變長。
服務(wù)端安裝部署
我是在虛擬機中的CentOS6.5中進行部署厨钻。
1.下載程序
2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解壓到適當(dāng)?shù)哪夸浫?opt/目錄
3.啟動RocketMQ：進入rocketmq/bin 目錄 執(zhí)行
nohup sh mqnamesrv &
4.啟動Broker扼雏，設(shè)置對應(yīng)的NameServer
nohup sh? mqbroker -n "127.0.0.1:9876" &
編寫客戶端
可以查看sameple中的quickstart源碼 1.Consumer 消息消費者
/**
* Consumer，訂閱消息
*/
public class Consumer {
? ? public static void main(String[] args) throws InterruptedException, MQClientException {
? ? ? ? DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("QuickStartConsumer");
? ? ? ? consumer.setNamesrvAddr("127.0.0.1:9876");
? ? ? ? consumer.setInstanceName("QuickStartConsumer");
? ? ? ? consumer.subscribe("QuickStart", "*");
? ? ? ? consumer.registerMessageListener(new MessageListenerConcurrently() {
? ? ? ? ? ? @Override
? ? ? ? ? ? public ConsumeConcurrentlyStatus consumeMessage(List msgs,
? ? ? ? ? ? ? ? ? ? ConsumeConcurrentlyContext context) {
? ? ? ? ? ? ? ? System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
? ? ? ? ? ? ? ? return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
? ? ? ? ? ? }
? ? ? ? });
? ? ? ? consumer.start();
? ? ? ? System.out.println("Consumer Started.");
? ? }
}
2.Producer消息生產(chǎn)者
/**
* Producer夯膀，發(fā)送消息
*
*/
public class Producer {
? ? public static void main(String[] args) throws MQClientException, InterruptedException {
? ? ? ? DefaultMQProducer producer = new DefaultMQProducer("QuickStartProducer");
? ? ? ? producer.setNamesrvAddr("127.0.0.1:9876");
? ? ? ? producer.setInstanceName("QuickStartProducer");
? ? ? ? producer.start();
? ? ? ? for (int i = 0; i < 1000; i++) {
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? Message msg = new Message("QuickStart",// topic
? ? ? ? ? ? ? ? ? ? "TagA",// tag
? ? ? ? ? ? ? ? ? ? ("Hello RocketMQ ,QuickStart" + i).getBytes()// body
? ? ? ? ? ? ? ? ? ? ? ? );
? ? ? ? ? ? ? ? SendResult sendResult = producer.send(msg);
? ? ? ? ? ? ? ? System.out.println(sendResult);
? ? ? ? ? ? }
? ? ? ? ? ? catch (Exception e) {
? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? Thread.sleep(1000);
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? producer.shutdown();
? ? }
}
3.首先運行Consumer程序诗充，一直在運行狀態(tài)接收服務(wù)器端推送過來的消息
23:18:07.587 [main] DEBUG i.n.c.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16
23:18:07.591 [main] DEBUG i.n.util.internal.PlatformDependent - Platform: Windows
23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - Java version: 7
23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noUnsafe: false
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.ByteBuffer.cleaner: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Buffer.address: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: true
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - sun.misc.Unsafe: available
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noJavassist: false
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - Javassist: unavailable
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent -
You don't have Javassist in your class path or you don't have enough permission to load dynamically generated classes.?
Please check the configuration for better performance.
23:18:07.595 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false
23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false
23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512
23:18:08.355 [main] DEBUG i.n.util.internal.ThreadLocalRandom - -Dio.netty.initialSeedUniquifier: 0x8c0d4793e5820c31
23:18:08.446 [NettyClientWorkerThread_1] DEBUG io.netty.util.ResourceLeakDetector
- -Dio.netty.noResourceLeakDetection: false
Consumer Started.
4.再次運行Producer程序，生成消息并發(fā)送到Broker诱建，Producer的日志沖沒了蝴蜓，但是可以看到Broker推送到Consumer的一條消息
ConsumeMessageThread-QuickStartConsumer-3 Receive New Messages: [MessageExt [queueId=0, storeSize=150,
queueOffset=244, sysFlag=0, bornTimestamp=1400772029972, bornHost=/10.162.0.7:54234,
storeTimestamp=1400772016017, storeHost=/127.0.0.1:10911, msgId=0A0A0A5900002A9F0000000000063257,
commitLogOffset=406103, bodyCRC=112549959, reconsumeTimes=0, preparedTransactionOffset=0,
toString()=Message [topic=QuickStart, flag=0,
properties={TAGS=TagA, WAIT=true, MAX_OFFSET=245, MIN_OFFSET=0}, body=29]]]
Consumer最佳實踐
1.消費過程要做到冪等（即消費端去重）
RocketMQ無法做到消息重復(fù)，所以如果業(yè)務(wù)對消息重復(fù)非常敏感俺猿，務(wù)必要在業(yè)務(wù)層面去重茎匠，有以下一些方式：
（1）.將消息的唯一鍵，可以是MsgId辜荠，也可以是消息內(nèi)容中的唯一標(biāo)識字段汽抚，例如訂單ID，消費之前判斷是否在DB或Tair（全局KV存儲）中存在伯病，如果不存在則插入造烁，并消費否过，否則跳過。（實踐過程要考慮原子性問題惭蟋，判斷是否存在可以嘗試插入苗桂，如果報主鍵沖突，則插入失敗告组，直接跳過） msgid一定是全局唯一的標(biāo)識符煤伟，但是可能會存在同樣的消息有兩個不同的msgid的情況（有多種原因），這種情況可能會使業(yè)務(wù)上重復(fù)木缝，建議最好使用消息體中的唯一標(biāo)識字段去重
（2）.使業(yè)務(wù)層面的狀態(tài)機去重
2.批量方式消費
如果業(yè)務(wù)流程支持批量方式消費便锨，則可以很大程度上的提高吞吐量，可以通過設(shè)置Consumer的consumerMessageBatchMaxSize參數(shù)我碟，默認(rèn)是1放案，即一次消費一條參數(shù)
3.跳過非重要的消息
發(fā)生消息堆積時，如果消費速度一直跟不上發(fā)送速度矫俺，可以選擇丟棄不重要的消息
@Override
? ? ? ? ? public ConsumeConcurrentlyStatus consumeMessage(List msgs,
? ? ? ? ? ? ? ? ? ConsumeConcurrentlyContext context) {
? ? ? ? ? ? ? System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
? ? ? ? ? ? ? long offset=msgs.get(0).getQueueOffset();
? ? ? ? ? ? ? String maxOffset=msgs.get(0).getProperty(MessageConst.PROPERTY_MAX_OFFSET);
? ? ? ? ? ? ? long diff=Long.parseLong(maxOffset)-offset;
? ? ? ? ? ? ? if(diff>100000){
? ? ? ? ? ? ? ? ? //處理消息堆積情況
? ? ? ? ? ? ? ? ? return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
? ? ? ? ? ? ? }
? ? ? ? ? ? ? return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
? ? ? ? ? }
如以上代碼所示吱殉，當(dāng)某個隊列的消息數(shù)堆積到 100000 條以上，則嘗試丟棄部分或全部消息厘托，這樣就可以快速追上發(fā)送消息的速度
4.優(yōu)化沒條消息消費過程
舉例如下友雳，某條消息的消費過程如下
1. 根據(jù)消息從 DB 查詢數(shù)據(jù) 1
2. 根據(jù)消息從 DB 查詢數(shù)據(jù)2
3. 復(fù)雜的業(yè)務(wù)計算
4. 向 DB 插入數(shù)據(jù)3
5. 向 DB 插入數(shù)據(jù) 4
這條消息的消費過程與 DB 交互了 4 次，如果按照每次 5ms 計算铅匹，那么總共耗時 20ms押赊，假設(shè)業(yè)務(wù)計算耗時 5ms，那么總過耗時 25ms伊群，如果能把 4 次 DB 交互優(yōu)化為 2 次考杉，那么總耗時就可以優(yōu)化到 15ms舰始，也就是說總體性能提高了 40%。
對于 Mysql 等 DB咽袜，如果部署在磁盤丸卷，那么與 DB 進行交互，如果數(shù)據(jù)沒有命中 cache询刹，每次交互的 RT 會直線上升谜嫉， 如果采用 SSD，則 RT 上升趨勢要明顯好于磁盤凹联。
個別應(yīng)用可能會遇到這種情況：在線下壓測消費過程中沐兰，db 表現(xiàn)非常好，每次 RT 都很短蔽挠，但是上線運行一段時間住闯，RT 就會變長，消費吞吐量直線下降
主要原因是線下壓測時間過短，線上運行一段時間后比原，cache 命中率下降插佛，那么 RT 就會增加。建議在線下壓測時量窘，要測試足夠長時間雇寇，盡可能模擬線上環(huán)境，壓測過程中蚌铜，數(shù)據(jù)的分布也很重要锨侯，數(shù)據(jù)不同，可能 cache 的命中率也會完全不同
Producer最佳實踐
1.發(fā)送消息注意事項
（1） 一個應(yīng)用盡可能用一個 Topic冬殃，消息子類型用 tags 來標(biāo)識识腿，tags 可以由應(yīng)用自由設(shè)置。只有發(fā)送消息設(shè)置了tags造壮，消費方在訂閱消息時渡讼，才可以利用 tags 在 broker 做消息過濾。
（2）每個消息在業(yè)務(wù)層面的唯一標(biāo)識碼耳璧，要設(shè)置到 keys 字段成箫，方便將來定位消息丟失問題。服務(wù)器會為每個消息創(chuàng)建索引（哈希索引）旨枯，應(yīng)用可以通過 topic蹬昌，key 來查詢這條消息內(nèi)容，以及消息被誰消費攀隔。由于是哈希索引皂贩，請務(wù)必保證 key 盡可能唯一，這樣可以避免潛在的哈希沖突昆汹。
（3）消息發(fā)送成功或者失敗明刷，要打印消息日志，務(wù)必要打印 sendresult 和 key 字段
（4）send 消息方法满粗，只要不拋異常辈末，就代表發(fā)送成功。但是發(fā)送成功會有多個狀態(tài)映皆，在 sendResult 里定義
SEND_OK：消息發(fā)送成功
FLUSH_DISK_TIMEOUT：消息發(fā)送成功挤聘，但是服務(wù)器刷盤超時，消息已經(jīng)進入服務(wù)器隊列捅彻，只有此時服務(wù)器宕機组去，消息才會丟失
FLUSH_SLAVE_TIMEOUT：消息發(fā)送成功，但是服務(wù)器同步到 Slave 時超時步淹，消息已經(jīng)進入服務(wù)器隊列从隆，只有此時服務(wù)器宕機诚撵，消息才會丟失
SLAVE_NOT_AVAILABLE：消息發(fā)送成功，但是此時 slave 不可用广料，消息已經(jīng)進入服務(wù)器隊列砾脑，只有此時服務(wù)器宕機，消息才會丟失艾杏。對于精確發(fā)送順序消息的應(yīng)用韧衣，由于順序消息的局限性，可能會涉及到主備自動切換問題购桑，所以如果sendresult 中的 status 字段不等于 SEND_OK畅铭，就應(yīng)該嘗試重試。對于其他應(yīng)用勃蜘，則沒有必要這樣
（5）對于消息不可丟失應(yīng)用硕噩，務(wù)必要有消息重發(fā)機制
2.消息發(fā)送失敗處理
Producer 的 send 方法本身支持內(nèi)部重試，重試邏輯如下：
（1） 至多重試 3 次
（2） 如果發(fā)送失敗缭贡，則輪轉(zhuǎn)到下一個 Broker
（3） 這個方法的總耗時時間不超過 sendMsgTimeout 設(shè)置的值炉擅，默認(rèn) 10s所以，如果本身向 broker 發(fā)送消息產(chǎn)生超時異常阳惹，就不會再做重試
如：
如果調(diào)用 send 同步方法發(fā)送失敗谍失，則嘗試將消息存儲到 db，由后臺線程定時重試莹汤，保證消息一定到達 Broker快鱼。
上述 db 重試方式為什么沒有集成到 MQ 客戶端內(nèi)部做，而是要求應(yīng)用自己去完成纲岭，基于以下幾點考慮：
（1）MQ 的客戶端設(shè)計為無狀態(tài)模式抹竹，方便任意的水平擴展，且對機器資源的消耗僅僅是 cpu止潮、內(nèi)存窃判、網(wǎng)絡(luò)
（2）如果 MQ 客戶端內(nèi)部集成一個 KV 存儲模塊，那么數(shù)據(jù)只有同步落盤才能較可靠沽翔，而同步落盤本身性能開銷較大兢孝，所以通常會采用異步落盤，又由于應(yīng)用關(guān)閉過程不受 MQ 運維人員控制仅偎，可能經(jīng)常會發(fā)生 kill -9 這樣暴力方式關(guān)閉，造成數(shù)據(jù)沒有及時落盤而丟失
（3）Producer 所在機器的可靠性較低雳殊，一般為虛擬機橘沥，不適合存儲重要數(shù)據(jù)。 綜上夯秃，建議重試過程交由應(yīng)用來控制座咆。
3.選擇 oneway 形式發(fā)送
一個 RPC 調(diào)用痢艺，通常是這樣一個過程
（1）客戶端發(fā)送請求到服務(wù)器
（2）服務(wù)器處理該請求
（3）服務(wù)器向客戶端返回應(yīng)答
所以一個 RPC 的耗時時間是上述三個步驟的總和，而某些場景要求耗時非常短介陶，但是對可靠性要求并不高堤舒，例如日志收集類應(yīng)用，此類應(yīng)用可以采用 oneway 形式調(diào)用哺呜，oneway 形式只發(fā)送請求不等待應(yīng)答舌缤，而發(fā)送請求在客戶端實現(xiàn)層面僅僅是一個 os 系統(tǒng)調(diào)用的開銷，即將數(shù)據(jù)寫入客戶端的 socket 緩沖區(qū)某残，此過程耗時通常在微秒級国撵。
RocketMQ不止可以直接推送消息，在消費端注冊監(jiān)聽器進行監(jiān)聽玻墅，還可以由消費端決定自己去拉取數(shù)據(jù)
/**
* PullConsumer介牙，訂閱消息
*/
public class PullConsumer {
? ? //Java緩存
? ? private static final Map offseTable = new HashMap();
? ? public static void main(String[] args) throws MQClientException {
? ? ? ? DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("PullConsumerGroup");
? ? ? ? consumer.setNamesrvAddr("127.0.0.1:9876");
? ? ? ? consumer.start();
? ? ? ? ? ? ? ? //拉取訂閱主題的隊列，默認(rèn)隊列大小是4
? ? ? ? Set mqs = consumer.fetchSubscribeMessageQueues("TopicTestMapBody");
? ? ? ? for (MessageQueue mq : mqs) {
? ? ? ? ? ? System.out.println("Consume from the queue: " + mq);
? ? ? ? ? ? SINGLE_MQ:while(true){
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? PullResult pullResult =
? ? ? ? ? ? ? ? ? ? ? ? ? ? consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
? ? ? ? ? ? ? ? ? ? List list=pullResult.getMsgFoundList();
? ? ? ? ? ? ? ? ? ? if(list!=null&&list.size()<100){
? ? ? ? ? ? ? ? ? ? ? ? for(MessageExt msg:list){
? ? ? ? ? ? ? ? ? ? ? ? ? ? System.out.println(SerializableInterface.deserialize(msg.getBody()));
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? System.out.println(pullResult.getNextBeginOffset());
? ? ? ? ? ? ? ? ? ? putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
? ? ? ? ? ? ? ? ? ? switch (pullResult.getPullStatus()) {
? ? ? ? ? ? ? ? ? ? case FOUND:
? ? ? ? ? ? ? ? ? ? ? ? // TODO
? ? ? ? ? ? ? ? ? ? ? ? break;
? ? ? ? ? ? ? ? ? ? case NO_MATCHED_MSG:
? ? ? ? ? ? ? ? ? ? ? ? break;
? ? ? ? ? ? ? ? ? ? case NO_NEW_MSG:
? ? ? ? ? ? ? ? ? ? ? ? break SINGLE_MQ;
? ? ? ? ? ? ? ? ? ? case OFFSET_ILLEGAL:
? ? ? ? ? ? ? ? ? ? ? ? break;
? ? ? ? ? ? ? ? ? ? default:
? ? ? ? ? ? ? ? ? ? ? ? break;
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? catch (Exception e) {
? ? ? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? consumer.shutdown();
? ? }
? ? private static void putMessageQueueOffset(MessageQueue mq, long offset) {
? ? ? ? offseTable.put(mq, offset);
? ? }
? ? private static long getMessageQueueOffset(MessageQueue mq) {
? ? ? ? Long offset = offseTable.get(mq);
? ? ? ? if (offset != null){
? ? ? ? ? ? System.out.println(offset);
? ? ? ? ? ? return offset;
? ? ? ? }
? ? ? ? return 0;
? ? }
剛開始的沒有細(xì)看PullResult對象澳厢，以為拉取到的結(jié)果沒有MessageExt對象還跑到群里面問別人环础，犯2了
特別要注意 靜態(tài)變量offsetTable的作用，拉取的是按照從offset（理解為下標(biāo)）位置開始拉取剩拢，拉取N條线得，offsetTable記錄下次拉取的offset位置。
在此我向大家推薦一個架構(gòu)學(xué)習(xí)交流群裸扶。交流學(xué)習(xí)群號：478030634 里面會分享一些資深架構(gòu)師錄制的視頻錄像：有Spring框都，MyBatis，Netty源碼分析呵晨，高并發(fā)魏保、高性能、分布式摸屠、微服務(wù)架構(gòu)的原理谓罗，JVM性能優(yōu)化、分布式架構(gòu)等這些成為架構(gòu)師必備的知識體系季二。還能領(lǐng)取免費的學(xué)習(xí)資源檩咱，目前受益良多
文章有點長，大家覺得作者總結(jié)的還可以胯舷，大家可以點擊下方二維碼進行關(guān)注刻蚯。《Java爛豬皮》公眾號聊的不僅僅是Java技術(shù)知識桑嘶，還有面試等干貨炊汹，后期還有大量架構(gòu)干貨。大家一起關(guān)注吧逃顶！關(guān)注爛豬皮讨便，你會了解的更多..............
