分散式開放訊息系統(RocketMQ)的原理與實踐
阿新 • • 發佈：2018-12-30
備註：
1.如果您此前未接觸過RocketMQ，請先閱讀附錄部分，以便了解RocketMQ的整體架構和相關術語
2.文中的MQServer與Broker表示同一概念
分散式訊息系統作為實現分散式系統可擴充套件、可伸縮性的關鍵元件，需要具有高吞吐量、高可用等特點。而談到訊息系統的設計，就回避不了兩個問題：
訊息的順序問題
訊息的重複問題
RocketMQ作為阿里開源的一款高效能、高吞吐量的訊息中介軟體，它是怎樣來解決這兩個問題的？RocketMQ 有哪些關鍵特性？其實現原理是怎樣的？
關鍵特性以及其實現原理
一、順序訊息
訊息有序指的是一類訊息消費時，能按照發送的順序來消費。例如：一個訂單產生了 3 條訊息，分別是訂單建立、訂單付款、訂單完成。消費時，要按照這個順序消費才有意義。但同時訂單之間又是可以並行消費的。
假如生產者產生了2條訊息：M1、M2，要保證這兩條訊息的順序，應該怎樣做？你腦中想到的可能是這樣：
你可能會採用這種方式保證訊息順序
M1傳送到S1後，M2傳送到S2，如果要保證M1先於M2被消費，那麼需要M1到達消費端後，通知S2，然後S2再將M2傳送到消費端。
這個模型存在的問題是，如果M1和M2分別傳送到兩臺Server上，就不能保證M1先達到，也就不能保證M1被先消費，那麼就需要在MQ Server叢集維護訊息的順序。那麼如何解決？一種簡單的方式就是將M1、M2傳送到同一個Server上：
保證訊息順序，你改進後的方法
這樣可以保證M1先於M2到達MQServer（客戶端等待M1成功後再發送M2），根據先達到先被消費的原則，M1會先於M2被消費，這樣就保證了訊息的順序。
這個模型，理論上可以保證訊息的順序，但在實際運用中你應該會遇到下面的問題：
網路延遲問題
只要將訊息從一臺伺服器發往另一臺伺服器，就會存在網路延遲問題。如上圖所示，如果傳送M1耗時大於傳送M2的耗時，那麼M2就先被消費，仍然不能保證訊息的順序。即使M1和M2同時到達消費端，由於不清楚消費端1和消費端2的負載情況，仍然有可能出現M2先於M1被消費。如何解決這個問題？將M1和M2發往同一個消費者即可，且傳送M1後，需要消費端響應成功後才能傳送M2。
但又會引入另外一個問題，如果傳送M1後，消費端1沒有響應，那是繼續傳送M2呢，還是重新發送M1？一般為了保證訊息一定被消費，肯定會選擇重發M1到另外一個消費端2，就如下圖所示。
保證訊息順序的正確姿勢
這樣的模型就嚴格保證訊息的順序，細心的你仍然會發現問題，消費端1沒有響應Server時有兩種情況，一種是M1確實沒有到達，另外一種情況是消費端1已經響應，但是Server端沒有收到。如果是第二種情況，重發M1，就會造成M1被重複消費。也就是我們後面要說的第二個問題，訊息重複問題。
回過頭來看訊息順序問題，嚴格的順序訊息非常容易理解，而且處理問題也比較容易，要實現嚴格的順序訊息，簡單且可行的辦法就是：
保證生產者 - MQServer - 消費者是一對一對一的關係
但是這樣設計，並行度就成為了訊息系統的瓶頸（吞吐量不夠），也會導致更多的異常處理，比如：只要消費端出現問題，就會導致整個處理流程阻塞，我們不得不花費更多的精力來解決阻塞的問題。
但我們的最終目標是要叢集的高容錯性和高吞吐量。這似乎是一對不可調和的矛盾，那麼阿里是如何解決的？
世界上解決一個計算機問題最簡單的方法：“恰好”不需要解決它！—— 沈詢
有些問題，看起來很重要，但實際上我們可以通過合理的設計或者將問題分解來規避。如果硬要把時間花在解決它們身上，實際上是浪費的，效率低下的。從這個角度來看訊息的順序問題，我們可以得出兩個結論：
1、不關注亂序的應用實際大量存在
2、佇列無序並不意味著訊息無序
最後我們從原始碼角度分析RocketMQ怎麼實現傳送順序訊息。
一般訊息是通過輪詢所有佇列來發送的（負載均衡策略），順序訊息可以根據業務，比如說訂單號相同的訊息傳送到同一個佇列。下面的示例中，OrderId相同的訊息，會發送到同一個佇列：
// RocketMQ預設提供了兩種MessageQueueSelector實現：隨機/Hash
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
在獲取到路由資訊以後，會根據MessageQueueSelector實現的演算法來選擇一個佇列，同一個OrderId獲取到的佇列是同一個佇列。
private SendResult send() {
// 獲取topic路由資訊
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
MessageQueue mq = null;
// 根據我們的演算法，選擇一個傳送佇列
// 這裡的arg = orderId
mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
if (mq != null) {
return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}
}
}
二、訊息重複
上面在解決訊息順序問題時，引入了一個新的問題，就是訊息重複。那麼RocketMQ是怎樣解決訊息重複的問題呢？還是“恰好”不解決。
造成訊息的重複的根本原因是：網路不可達。只要通過網路交換資料，就無法避免這個問題。所以解決這個問題的辦法就是不解決，轉而繞過這個問題。那麼問題就變成了：如果消費端收到兩條一樣的訊息，應該怎樣處理？
1、消費端處理訊息的業務邏輯保持冪等性
2、保證每條訊息都有唯一編號且保證訊息處理成功與去重表的日誌同時出現
第1條很好理解，只要保持冪等性，不管來多少條重複訊息，最後處理的結果都一樣。第2條原理就是利用一張日誌表來記錄已經處理成功的訊息的ID，如果新到的訊息ID已經在日誌表中，那麼就不再處理這條訊息。
我們可以看到第1條的解決方式，很明顯應該在消費端實現，不屬於訊息系統要實現的功能。第2條可以訊息系統實現，也可以業務端實現。正常情況下出現重複訊息的概率不一定大，且由訊息系統實現的話，肯定會對訊息系統的吞吐量和高可用有影響，所以最好還是由業務端自己處理訊息重複的問題，這也是RocketMQ不解決訊息重複的問題的原因。
RocketMQ不保證訊息不重複，如果你的業務需要保證嚴格的不重複訊息，需要你自己在業務端去重。
三、事務訊息
RocketMQ除了支援普通訊息，順序訊息，另外還支援事務訊息。首先討論一下什麼是事務訊息以及支援事務訊息的必要性。我們以一個轉帳的場景為例來說明這個問題：Bob向Smith轉賬100塊。
在單機環境下，執行事務的情況，大概是下面這個樣子：
單機環境下轉賬事務示意圖
當用戶增長到一定程度，Bob和Smith的賬戶及餘額資訊已經不在同一臺伺服器上了，那麼上面的流程就變成了這樣：
叢集環境下轉賬事務示意圖
這時候你會發現，同樣是一個轉賬的業務，在叢集環境下，耗時居然成倍的增長，這顯然是不能夠接受的。那我們如何來規避這個問題？
大事務 = 小事務 + 非同步
將大事務拆分成多個小事務非同步執行。這樣基本上能夠將跨機事務的執行效率優化到與單機一致。轉賬的事務就可以分解成如下兩個小事務：
小事務+非同步訊息
圖中執行本地事務（Bob賬戶扣款）和傳送非同步訊息應該保持同時成功或者失敗中，也就是扣款成功了，傳送訊息一定要成功，如果扣款失敗了，就不能再發送訊息。那問題是：我們是先扣款還是先發送訊息呢？
首先我們看下，先發送訊息，大致的示意圖如下：
事務訊息：先發送訊息
存在的問題是：如果訊息傳送成功，但是扣款失敗，消費端就會消費此訊息，進而向Smith賬戶加錢。
先發訊息不行，那我們就先扣款唄，大致的示意圖如下：
事務訊息-先扣款
存在的問題跟上面類似：如果扣款成功，傳送訊息失敗，就會出現Bob扣錢了，但是Smith賬戶未加錢。
可能大家會有很多的方法來解決這個問題，比如：直接將發訊息放到Bob扣款的事務中去，如果傳送失敗，丟擲異常，事務回滾。這樣的處理方式也符合“恰好”不需要解決的原則。RocketMQ支援事務訊息，下面我們來看看RocketMQ是怎樣來實現的。
RocketMQ實現傳送事務訊息
RocketMQ第一階段傳送Prepared訊息時，會拿到訊息的地址，第二階段執行本地事物，第三階段通過第一階段拿到的地址去訪問訊息，並修改狀態。細心的你可能又發現問題了，如果確認訊息傳送失敗了怎麼辦？RocketMQ會定期掃描訊息叢集中的事物訊息，這時候發現了Prepared訊息，它會向訊息傳送者確認，Bob的錢到底是減了還是沒減呢？如果減了是回滾還是繼續傳送確認訊息呢？RocketMQ會根據傳送端設定的策略來決定是回滾還是繼續傳送確認訊息。這樣就保證了訊息傳送與本地事務同時成功或同時失敗。
那我們來看下RocketMQ原始碼，是不是這樣來處理事務訊息的。客戶端傳送事務訊息的部分（完整程式碼請檢視：rocketmq-example工程下的com.alibaba.rocketmq.example.transaction.TransactionProducer）：
// 未決事務，MQ伺服器回查客戶端
// 也就是上文所說的，當RocketMQ發現`Prepared訊息`時，會根據這個Listener實現的策略來決斷事務
TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
// 構造事務訊息的生產者
TransactionMQProducer producer = new TransactionMQProducer("groupName");
// 設定事務決斷處理類
producer.setTransactionCheckListener(transactionCheckListener);
// 本地事務的處理邏輯，相當於示例中檢查Bob賬戶並扣錢的邏輯
TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
producer.start()
// 構造MSG，省略構造引數
Message msg = new Message(......);
// 傳送訊息
SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
producer.shutdown();
接著檢視sendMessageInTransaction方法的原始碼，總共分為3個階段：傳送Prepared訊息、執行本地事務、傳送確認訊息。
public TransactionSendResult sendMessageInTransaction(.....) {
// 邏輯程式碼，非實際程式碼
// 1.傳送訊息
sendResult = this.send(msg);
// sendResult.getSendStatus() == SEND_OK
// 2.如果訊息傳送成功，處理與訊息關聯的本地事務單元
LocalTransactionState localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
// 3.結束事務
this.endTransaction(sendResult, localTransactionState, localException);
}
endTransaction方法會將請求發往broker(mq server)去更新事物訊息的最終狀態：
根據sendResult找到Prepared訊息
根據localTransaction更新訊息的最終狀態
如果endTransaction方法執行失敗，導致資料沒有傳送到broker，broker會有回查執行緒定時（預設1分鐘）掃描每個儲存事務狀態的表格檔案，如果是已經提交或者回滾的訊息直接跳過，如果是prepared狀態則會向Producer發起CheckTransaction請求，Producer會呼叫DefaultMQProducerImpl.checkTransactionState()方法來處理broker的定時回撥請求，而checkTransactionState會呼叫我們的事務設定的決斷方法，最後呼叫endTransactionOneway讓broker來更新訊息的最終狀態。
再回到轉賬的例子，如果Bob的賬戶的餘額已經減少，且訊息已經發送成功，Smith端開始消費這條訊息，這個時候就會出現消費失敗和消費超時兩個問題？解決超時問題的思路就是一直重試，直到消費端消費訊息成功，整個過程中有可能會出現訊息重複的問題，按照前面的思路解決即可。
消費事務訊息
這樣基本上可以解決超時問題，但是如果消費失敗怎麼辦？阿里提供給我們的解決方法是：人工解決。大家可以考慮一下，按照事務的流程，因為某種原因Smith加款失敗，需要回滾整個流程。如果訊息系統要實現這個回滾流程的話，系統複雜度將大大提升，且很容易出現Bug，估計出現Bug的概率會比消費失敗的概率大很多。我們需要衡量是否值得花這麼大的代價來解決這樣一個出現概率非常小的問題，這也是大家在解決疑難問題時需要多多思考的地方。
20160321補充：在3.2.6版本中移除了事務訊息的實現，所以此版本不支援事務訊息，具體情況請參考rocketmq的issues：
https://github.com/alibaba/RocketMQ/issues/65
https://github.com/alibaba/RocketMQ/issues/138
https://github.com/alibaba/RocketMQ/issues/156
四、Producer如何傳送訊息
Producer輪詢某topic下的所有佇列的方式來實現傳送方的負載均衡，如下圖所示：
producer傳送訊息負載均衡
首先分析一下RocketMQ的客戶端傳送訊息的原始碼：
// 構造Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 初始化Producer，整個應用生命週期內，只需要初始化1次
producer.start();
// 構造Message
Message msg = new Message("TopicTest1",// topic
"TagA",// tag：給訊息打標籤,用於區分一類訊息，可為null
"OrderID188",// key：自定義Key，可以用於去重，可為null
("Hello MetaQ").getBytes());// body：訊息內容
// 傳送訊息並返回結果
SendResult sendResult = producer.send(msg);
// 清理資源，關閉網路連線，登出自己
producer.shutdown();
在整個應用生命週期內，生產者需要呼叫一次start方法來初始化，初始化主要完成的任務有：
如果沒有指定namesrv地址，將會自動定址
啟動定時任務：更新namesrv地址、從namsrv更新topic路由資訊、清理已經掛掉的broker、向所有broker傳送心跳...
啟動負載均衡的服務
初始化完成後，開始傳送訊息，傳送訊息的主要程式碼如下：
private SendResult sendDefaultImpl(Message msg,......) {
// 檢查Producer的狀態是否是RUNNING
this.makeSureStateOK();
// 檢查msg是否合法：是否為null、topic,body是否為空、body是否超長
Validators.checkMessage(msg, this.defaultMQProducer);
// 獲取topic路由資訊
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
// 從路由資訊中選擇一個訊息佇列
MessageQueue mq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
// 將訊息傳送到該佇列上去
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}
程式碼中需要關注的兩個方法tryToFindTopicPublishInfo和selectOneMessageQueue。前面說過在producer初始化時，會啟動定時任務獲取路由資訊並更新到本地快取，所以tryToFindTopicPublishInfo會首先從快取中獲取topic路由資訊，如果沒有獲取到，則會自己去namesrv獲取路由資訊。selectOneMessageQueue方法通過輪詢的方式，返回一個佇列，以達到負載均衡的目的。
如果Producer傳送訊息失敗，會自動重試，重試的策略：
重試次數 < retryTimesWhenSendFailed（可配置）
總的耗時（包含重試n次的耗時） < sendMsgTimeout（傳送訊息時傳入的引數）
同時滿足上面兩個條件後，Producer會選擇另外一個佇列傳送訊息
五、訊息儲存
RocketMQ的訊息儲存是由consume queue和commit log配合完成的。
1、Consume Queue
consume queue是訊息的邏輯佇列，相當於字典的目錄，用來指定訊息在物理檔案commit log上的位置。
我們可以在配置中指定consumequeue與commitlog儲存的目錄
每個topic下的每個queue都有一個對應的consumequeue檔案，比如：
${rocketmq.home}/store/consumequeue/${topicName}/${queueId}/${fileName}
Consume Queue檔案組織，如圖所示：
Consume Queue檔案組織示意圖
根據topic和queueId來組織檔案，圖中TopicA有兩個佇列0,1，那麼TopicA和QueueId=0組成一個ConsumeQueue，TopicA和QueueId=1組成另一個ConsumeQueue。
按照消費端的GroupName來分組重試佇列，如果消費端消費失敗，訊息將被髮往重試佇列中，比如圖中的%RETRY%ConsumerGroupA。
按照消費端的GroupName來分組死信佇列，如果消費端消費失敗，並重試指定次數後，仍然失敗，則發往死信佇列，比如圖中的%DLQ%ConsumerGroupA。
死信佇列（Dead Letter Queue）一般用於存放由於某種原因無法傳遞的訊息，比如處理失敗或者已經過期的訊息。
Consume Queue中儲存單元是一個20位元組定長的二進位制資料，順序寫順序讀，如下圖所示：
consumequeue檔案儲存單元格式
CommitLog Offset是指這條訊息在Commit Log檔案中的實際偏移量
Size儲存中訊息的大小
Message Tag HashCode儲存訊息的Tag的雜湊值：主要用於訂閱時訊息過濾（訂閱時如果指定了Tag，會根據HashCode來快速查詢到訂閱的訊息）
2、Commit Log
CommitLog：訊息存放的物理檔案，每臺broker上的commitlog被本機所有的queue共享，不做任何區分。
檔案的預設位置如下，仍然可通過配置檔案修改：
${user.home} \store\${commitlog}\${fileName}
CommitLog的訊息儲存單元長度不固定，檔案順序寫，隨機讀。訊息的儲存結構如下表所示，按照編號順序以及編號對應的內容依次儲存。
Commit Log儲存單元結構圖
3、訊息儲存實現
訊息儲存實現，比較複雜，也值得大家深入瞭解，後面會單獨成文來分析，這小節只以程式碼說明一下具體的流程。
// Set the storage time
msg.setStoreTimestamp(System.currentTimeMillis());
// Set the message body BODY CRC (consider the most appropriate setting
msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
synchronized (this) {
long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
// Here settings are stored timestamp, in order to ensure an orderly global
msg.setStoreTimestamp(beginLockTimestamp);
// MapedFile：操作物理檔案在記憶體中的對映以及將記憶體資料持久化到物理檔案中
MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile();
// 將Message追加到檔案commitlog
result = mapedFile.appendMessage(msg, this.appendMessageCallback);
switch (result.getStatus()) {
case PUT_OK:break;
case END_OF_FILE:
// Create a new file, re-write the message
mapedFile = this.mapedFileQueue.getLastMapedFile();
result = mapedFile.appendMessage(msg, this.appendMessageCallback);
break;
DispatchRequest dispatchRequest = new DispatchRequest(
topic,// 1
queueId,// 2
result.getWroteOffset(),// 3
result.getWroteBytes(),// 4
tagsCode,// 5
msg.getStoreTimestamp(),// 6
result.getLogicsOffset(),// 7
msg.getKeys(),// 8
/**
* Transaction
*/
msg.getSysFlag(),// 9
msg.getPreparedTransactionOffset());// 10
// 1.分發訊息位置到ConsumeQueue
// 2.分發到IndexService建立索引
this.defaultMessageStore.putDispatchRequest(dispatchRequest);
}
4、訊息的索引檔案
如果一個訊息包含key值的話，會使用IndexFile儲存訊息索引，檔案的內容結構如圖：
訊息索引
索引檔案主要用於根據key來查詢訊息的，流程主要是：
根據查詢的 key 的 hashcode%slotNum 得到具體的槽的位置(slotNum 是一個索引檔案裡面包含的最大槽的數目，例如圖中所示 slotNum=5000000)
根據 slotValue(slot 位置對應的值)查詢到索引項列表的最後一項(倒序排列,slotValue 總是指向最新的一個索引項)
遍歷索引項列表返回查詢時間範圍內的結果集(預設一次最大返回的 32 條記錄)
六、訊息訂閱
RocketMQ訊息訂閱有兩種模式，一種是Push模式，即MQServer主動向消費端推送；另外一種是Pull模式，即消費端在需要時，主動到MQServer拉取。但在具體實現時，Push和Pull模式都是採用消費端主動拉取的方式。
首先看下消費端的負載均衡：
消費端負載均衡
消費端會通過RebalanceService執行緒，10秒鐘做一次基於topic下的所有佇列負載：
遍歷Consumer下的所有topic，然後根據topic訂閱所有的訊息
獲取同一topic和Consumer Group下的所有Consumer
然後根據具體的分配策略來分配消費佇列，分配的策略包含：平均分配、消費端配置等
如同上圖所示：如果有 5 個佇列，2 個 consumer，那麼第一個 Consumer 消費 3 個佇列，第二 consumer 消費 2 個佇列。這裡採用的就是平均分配策略，它類似於我們的分頁，TOPIC下面的所有queue就是記錄，Consumer的個數就相當於總的頁數，那麼每頁有多少條記錄，就類似於某個Consumer會消費哪些佇列。
通過這樣的策略來達到大體上的平均消費，這樣的設計也可以很方面的水平擴充套件Consumer來提高消費能力。
消費端的Push模式是通過長輪詢的模式來實現的，就如同下圖：
Push模式示意圖
Consumer端每隔一段時間主動向broker傳送拉訊息請求，broker在收到Pull請求後，如果有訊息就立即返回資料，Consumer端收到返回的訊息後，再回調消費者設定的Listener方法。如果broker在收到Pull請求時，訊息佇列裡沒有資料，broker端會阻塞請求直到有資料傳遞或超時才返回。
當然，Consumer端是通過一個執行緒將阻塞佇列LinkedBlockingQueue<PullRequest>中的PullRequest傳送到broker拉取訊息，以防止Consumer一致被阻塞。而Broker端，在接收到Consumer的PullRequest時，如果發現沒有訊息，就會把PullRequest扔到ConcurrentHashMap中快取起來。broker在啟動時，會啟動一個執行緒不停的從ConcurrentHashMap取出PullRequest檢查，直到有資料返回。
七、RocketMQ的其他特性
前面的6個特性都是基本上都是點到為止，想要深入瞭解，還需要大家多多檢視原始碼，多多在實際中運用。當然除了已經提到的特性外，RocketMQ還支援：
定時訊息
訊息的刷盤策略
主動同步策略：同步雙寫、非同步複製
海量訊息堆積能力
高效通訊
.......
其中涉及到的很多設計思路和解決方法都值得我們深入研究：
訊息的儲存設計：既要滿足海量訊息的堆積能力，又要滿足極快的查詢效率，還要保證寫入的效率。
高效的通訊元件設計：高吞吐量，毫秒級的訊息投遞能力都離不開高效的通訊。
.......
RocketMQ最佳實踐
一、Producer最佳實踐
1、一個應用盡可能用一個 Topic，訊息子型別用 tags 來標識，tags 可以由應用自由設定。只有傳送訊息設定了tags，消費方在訂閱訊息時，才可以利用 tags 在 broker 做訊息過濾。
2、每個訊息在業務層面的唯一標識碼，要設定到 keys 欄位，方便將來定位訊息丟失問題。由於是雜湊索引，請務必保證 key 儘可能唯一，這樣可以避免潛在的雜湊衝突。
3、訊息傳送成功或者失敗，要列印訊息日誌，務必要列印 sendresult 和 key 欄位。
4、對於訊息不可丟失應用，務必要有訊息重發機制。例如：訊息傳送失敗，儲存到資料庫，能有定時程式嘗試重發或者人工觸發重發。
5、某些應用如果不關注訊息是否傳送成功，請直接使用sendOneWay方法傳送訊息。
二、Consumer最佳實踐
1、消費過程要做到冪等（即消費端去重）
2、儘量使用批量方式消費方式，可以很大程度上提高消費吞吐量。
3、優化每條訊息消費過程
三、其他配置
線上應該關閉autoCreateTopicEnable，即在配置檔案中將其設定為false。
RocketMQ在傳送訊息時，會首先獲取路由資訊。如果是新的訊息，由於MQServer上面還沒有建立對應的Topic，這個時候，如果上面的配置開啟的話，會返回預設TOPIC的（RocketMQ會在每臺broker上面建立名為TBW102的TOPIC）路由資訊，然後Producer會選擇一臺Broker傳送訊息，選中的broker在儲存訊息時，發現訊息的topic還沒有建立，就會自動建立topic。後果就是：以後所有該TOPIC的訊息，都將傳送到這臺broker上，達不到負載均衡的目的。
所以基於目前RocketMQ的設計，建議關閉自動建立TOPIC的功能，然後根據訊息量的大小，手動建立TOPIC。
RocketMQ設計相關
RocketMQ的設計假定：
每臺PC機器都可能宕機不可服務
任意叢集都有可能處理能力不足
最壞的情況一定會發生
內網環境需要低延遲來提供最佳使用者體驗
RocketMQ的關鍵設計：
分散式叢集化
強資料安全
海量資料堆積
毫秒級投遞延遲（推拉模式）
這是RocketMQ在設計時的假定前提以及需要到達的效果。我想這些假定適用於所有的系統設計。隨著我們系統的服務的增多，每位開發者都要注意自己的程式是否存在單點故障，如果掛了應該怎麼恢復、能不能很好的水平擴充套件、對外的介面是否足夠高效、自己管理的資料是否足夠安全...... 多多規範自己的設計，才能開發出高效健壯的程式。
附錄：RocketMQ涉及到的幾個專業術語和整體架構介紹
一、RocketMQ中的專業術語
Topic
topic表示訊息的第一級型別，比如一個電商系統的訊息可以分為：交易訊息、物流訊息...... 一條訊息必須有一個Topic。
Tag
Tag表示訊息的第二級型別，比如交易訊息又可以分為：交易建立訊息，交易完成訊息..... 一條訊息可以沒有Tag。RocketMQ提供2級訊息分類，方便大家靈活控制。
Queue
一個topic下，我們可以設定多個queue(訊息佇列)。當我們傳送訊息時，需要要指定該訊息的topic。RocketMQ會輪詢該topic下的所有佇列，將訊息傳送出去。
Producer 與 Producer Group
Producer表示訊息佇列的生產者。訊息佇列的本質就是實現了publish-subscribe模式，生產者生產訊息，消費者消費訊息。所以這裡的Producer就是用來生產和傳送訊息的，一般指業務系統。
Producer Group是一類Producer的集合名稱，這類Producer通常傳送一類訊息，且傳送邏輯一致。
Consumer 與 Consumer Group
訊息消費者，一般由後臺系統非同步消費訊息。
Push Consumer
Consumer 的一種，應用通常向 Consumer 物件註冊一個 Listener 介面，一旦收到訊息，Consumer 物件立刻回撥 Listener 介面方法。
Pull Consumer
Consumer 的一種，應用通常主動呼叫 Consumer 的拉訊息方法從 Broker 拉訊息，主動權由應用控制。
Consumer Group是一類Consumer的集合名稱，這類Consumer通常消費一類訊息，且消費邏輯一致。
Broker
訊息的中轉者，負責儲存和轉發訊息。可以理解為訊息佇列伺服器，提供了訊息的接收、儲存、拉取和轉發服務。broker是RocketMQ的核心，它不不能掛的，所以需要保證broker的高可用。
廣播消費
一條訊息被多個Consumer消費，即使這些Consumer屬於同一個Consumer Group，訊息也會被Consumer Group中的每個Consumer都消費一次。在廣播消費中的Consumer Group概念可以認為在訊息劃分方面無意義。
叢集消費
一個Consumer Group中的Consumer例項平均分攤消費訊息。例如某個Topic有 9 條訊息，其中一個Consumer Group有 3 個例項(可能是 3 個程序,或者 3 臺機器)，那麼每個例項只消費其中的 3 條訊息。
NameServer
NameServer即名稱服務，兩個功能：
接收broker的請求，註冊broker的路由資訊
介面client的請求，根據某個topic獲取其到broker的路由資訊
NameServer沒有狀態，可以橫向擴充套件。每個broker在啟動的時候會到NameServer註冊；Producer在傳送訊息前會根據topic到NameServer獲取路由(到broker)資訊；Consumer也會定時獲取topic路由資訊。
二、RocketMQ Overview
rocketmq overview
Producer向一些佇列輪流傳送訊息，佇列集合稱為Topic，Consumer如果做廣播消費，則一個consumer例項消費這個Topic對應的所有佇列；如果做叢集消費，則多個Consumer例項平均消費這個Topic對應的佇列集合。
再看下RocketMQ物理部署結構圖：
RocketMQ網路部署圖
RocketMQ網路部署特點：
Name Server 是一個幾乎無狀態節點，可叢集部署，節點之間無任何資訊同步。
Broker部署相對複雜，Broker分為Master與Slave，一個Master可以對應多個Slave，但是一個Slave只能對應一個Master，Master與Slave的對應關係通過指定相同的BrokerName，不同的BrokerId來定義，BrokerId=0表示Master，非0表示Slave。Master也可以部署多個。每個Broker與Name Server叢集中的所有節點建立長連線，定時註冊Topic資訊到所有Name Server。
Producer與Name Server叢集中的其中一個節點（隨機選擇）建立長連線，定期從Name Server取Topic路由資訊，並向提供Topic 服務的Master建立長連線，且定時向Master傳送心跳。Producer 完全無狀態，可叢集部署。
Consumer與Name Server叢集中的其中一個節點（隨機選擇）建立長連線，定期從Name Server取Topic 路由資訊，並向提供Topic服務的Master、Slave建立長連線，且定時向Master、Slave傳送心跳。Consumer既可以從Master訂閱訊息，也可以從Slave訂閱訊息，訂閱規則由Broker配置決定。
三、其他參考資料
RocketMQ使用者指南
RocketMQ原理簡介
RocketMQ最佳實踐
阿里分散式開放訊息服務(ONS)原理與實踐2
阿里分散式開放訊息服務(ONS)原理與實踐3
RocketMQ原理解析
備註：水平有限，難免疏漏，如果問題請留言
1.如果您此前未接觸過RocketMQ，請先閱讀附錄部分，以便了解RocketMQ的整體架構和相關術語
2.文中的MQServer與Broker表示同一概念
分散式訊息系統作為實現分散式系統可擴充套件、可伸縮性的關鍵元件，需要具有高吞吐量、高可用等特點。而談到訊息系統的設計，就回避不了兩個問題：
訊息的順序問題
訊息的重複問題
RocketMQ作為阿里開源的一款高效能、高吞吐量的訊息中介軟體，它是怎樣來解決這兩個問題的？RocketMQ 有哪些關鍵特性？其實現原理是怎樣的？
關鍵特性以及其實現原理
一、順序訊息
訊息有序指的是一類訊息消費時，能按照發送的順序來消費。例如：一個訂單產生了 3 條訊息，分別是訂單建立、訂單付款、訂單完成。消費時，要按照這個順序消費才有意義。但同時訂單之間又是可以並行消費的。
假如生產者產生了2條訊息：M1、M2，要保證這兩條訊息的順序，應該怎樣做？你腦中想到的可能是這樣：
你可能會採用這種方式保證訊息順序
M1傳送到S1後，M2傳送到S2，如果要保證M1先於M2被消費，那麼需要M1到達消費端後，通知S2，然後S2再將M2傳送到消費端。
這個模型存在的問題是，如果M1和M2分別傳送到兩臺Server上，就不能保證M1先達到，也就不能保證M1被先消費，那麼就需要在MQ Server叢集維護訊息的順序。那麼如何解決？一種簡單的方式就是將M1、M2傳送到同一個Server上：
保證訊息順序，你改進後的方法
這樣可以保證M1先於M2到達MQServer（客戶端等待M1成功後再發送M2），根據先達到先被消費的原則，M1會先於M2被消費，這樣就保證了訊息的順序。
這個模型，理論上可以保證訊息的順序，但在實際運用中你應該會遇到下面的問題：
網路延遲問題
只要將訊息從一臺伺服器發往另一臺伺服器，就會存在網路延遲問題。如上圖所示，如果傳送M1耗時大於傳送M2的耗時，那麼M2就先被消費，仍然不能保證訊息的順序。即使M1和M2同時到達消費端，由於不清楚消費端1和消費端2的負載情況，仍然有可能出現M2先於M1被消費。如何解決這個問題？將M1和M2發往同一個消費者即可，且傳送M1後，需要消費端響應成功後才能傳送M2。
但又會引入另外一個問題，如果傳送M1後，消費端1沒有響應，那是繼續傳送M2呢，還是重新發送M1？一般為了保證訊息一定被消費，肯定會選擇重發M1到另外一個消費端2，就如下圖所示。
保證訊息順序的正確姿勢
這樣的模型就嚴格保證訊息的順序，細心的你仍然會發現問題，消費端1沒有響應Server時有兩種情況，一種是M1確實沒有到達，另外一種情況是消費端1已經響應，但是Server端沒有收到。如果是第二種情況，重發M1，就會造成M1被重複消費。也就是我們後面要說的第二個問題，訊息重複問題。
回過頭來看訊息順序問題，嚴格的順序訊息非常容易理解，而且處理問題也比較容易，要實現嚴格的順序訊息，簡單且可行的辦法就是：
保證生產者 - MQServer - 消費者是一對一對一的關係
但是這樣設計，並行度就成為了訊息系統的瓶頸（吞吐量不夠），也會導致更多的異常處理，比如：只要消費端出現問題，就會導致整個處理流程阻塞，我們不得不花費更多的精力來解決阻塞的問題。
但我們的最終目標是要叢集的高容錯性和高吞吐量。這似乎是一對不可調和的矛盾，那麼阿里是如何解決的？
世界上解決一個計算機問題最簡單的方法：“恰好”不需要解決它！—— 沈詢
有些問題，看起來很重要，但實際上我們可以通過合理的設計或者將問題分解來規避。如果硬要把時間花在解決它們身上，實際上是浪費的，效率低下的。從這個角度來看訊息的順序問題，我們可以得出兩個結論：
1、不關注亂序的應用實際大量存在
2、佇列無序並不意味著訊息無序
最後我們從原始碼角度分析RocketMQ怎麼實現傳送順序訊息。
一般訊息是通過輪詢所有佇列來發送的（負載均衡策略），順序訊息可以根據業務，比如說訂單號相同的訊息傳送到同一個佇列。下面的示例中，OrderId相同的訊息，會發送到同一個佇列：
// RocketMQ預設提供了兩種MessageQueueSelector實現：隨機/Hash
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
在獲取到路由資訊以後，會根據MessageQueueSelector實現的演算法來選擇一個佇列，同一個OrderId獲取到的佇列是同一個佇列。
private SendResult send() {
// 獲取topic路由資訊
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
MessageQueue mq = null;
// 根據我們的演算法，選擇一個傳送佇列
// 這裡的arg = orderId
mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
if (mq != null) {
return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}
}
}
二、訊息重複
上面在解決訊息順序問題時，引入了一個新的問題，就是訊息重複。那麼RocketMQ是怎樣解決訊息重複的問題呢？還是“恰好”不解決。
造成訊息的重複的根本原因是：網路不可達。只要通過網路交換資料，就無法避免這個問題。所以解決這個問題的辦法就是不解決，轉而繞過這個問題。那麼問題就變成了：如果消費端收到兩條一樣的訊息，應該怎樣處理？
1、消費端處理訊息的業務邏輯保持冪等性
2、保證每條訊息都有唯一編號且保證訊息處理成功與去重表的日誌同時出現
第1條很好理解，只要保持冪等性，不管來多少條重複訊息，最後處理的結果都一樣。第2條原理就是利用一張日誌表來記錄已經處理成功的訊息的ID，如果新到的訊息ID已經在日誌表中，那麼就不再處理這條訊息。
我們可以看到第1條的解決方式，很明顯應該在消費端實現，不屬於訊息系統要實現的功能。第2條可以訊息系統實現，也可以業務端實現。正常情況下出現重複訊息的概率不一定大，且由訊息系統實現的話，肯定會對訊息系統的吞吐量和高可用有影響，所以最好還是由業務端自己處理訊息重複的問題，這也是RocketMQ不解決訊息重複的問題的原因。
RocketMQ不保證訊息不重複，如果你的業務需要保證嚴格的不重複訊息，需要你自己在業務端去重。
三、事務訊息
RocketMQ除了支援普通訊息，順序訊息，另外還支援事務訊息。首先討論一下什麼是事務訊息以及支援事務訊息的必要性。我們以一個轉帳的場景為例來說明這個問題：Bob向Smith轉賬100塊。
在單機環境下，執行事務的情況，大概是下面這個樣子：
單機環境下轉賬事務示意圖
當用戶增長到一定程度，Bob和Smith的賬戶及餘額資訊已經不在同一臺伺服器上了，那麼上面的流程就變成了這樣：
叢集環境下轉賬事務示意圖
這時候你會發現，同樣是一個轉賬的業務，在叢集環境下，耗時居然成倍的增長，這顯然是不能夠接受的。那我們如何來規避這個問題？
大事務 = 小事務 + 非同步
將大事務拆分成多個小事務非同步執行。這樣基本上能夠將跨機事務的執行效率優化到與單機一致。轉賬的事務就可以分解成如下兩個小事務：
小事務+非同步訊息
圖中執行本地事務（Bob賬戶扣款）和傳送非同步訊息應該保持同時成功或者失敗中，也就是扣款成功了，傳送訊息一定要成功，如果扣款失敗了，就不能再發送訊息。那問題是：我們是先扣款還是先發送訊息呢？
首先我們看下，先發送訊息，大致的示意圖如下：
事務訊息：先發送訊息
存在的問題是：如果訊息傳送成功，但是扣款失敗，消費端就會消費此訊息，進而向Smith賬戶加錢。
先發訊息不行，那我們就先扣款唄，大致的示意圖如下：
事務訊息-先扣款
存在的問題跟上面類似：如果扣款成功，傳送訊息失敗，就會出現Bob扣錢了，但是Smith賬戶未加錢。
可能大家會有很多的方法來解決這個問題，比如：直接將發訊息放到Bob扣款的事務中去，如果傳送失敗，丟擲異常，事務回滾。這樣的處理方式也符合“恰好”不需要解決的原則。RocketMQ支援事務訊息，下面我們來看看RocketMQ是怎樣來實現的。
RocketMQ實現傳送事務訊息
RocketMQ第一階段傳送Prepared訊息時，會拿到訊息的地址，第二階段執行本地事物，第三階段通過第一階段拿到的地址去訪問訊息，並修改狀態。細心的你可能又發現問題了，如果確認訊息傳送失敗了怎麼辦？RocketMQ會定期掃描訊息叢集中的事物訊息，這時候發現了Prepared訊息，它會向訊息傳送者確認，Bob的錢到底是減了還是沒減呢？如果減了是回滾還是繼續傳送確認訊息呢？RocketMQ會根據傳送端設定的策略來決定是回滾還是繼續傳送確認訊息。這樣就保證了訊息傳送與本地事務同時成功或同時失敗。
那我們來看下RocketMQ原始碼，是不是這樣來處理事務訊息的。客戶端傳送事務訊息的部分（完整程式碼請檢視：rocketmq-example工程下的com.alibaba.rocketmq.example.transaction.TransactionProducer）：
// 未決事務，MQ伺服器回查客戶端
// 也就是上文所說的，當RocketMQ發現`Prepared訊息`時，會根據這個Listener實現的策略來決斷事務
TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
// 構造事務訊息的生產者
TransactionMQProducer producer = new TransactionMQProducer("groupName");
// 設定事務決斷處理類
producer.setTransactionCheckListener(transactionCheckListener);
// 本地事務的處理邏輯，相當於示例中檢查Bob賬戶並扣錢的邏輯
TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
producer.start()
// 構造MSG，省略構造引數
Message msg = new Message(......);
// 傳送訊息
SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
producer.shutdown();
接著檢視sendMessageInTransaction方法的原始碼，總共分為3個階段：傳送Prepared訊息、執行本地事務、傳送確認訊息。
public TransactionSendResult sendMessageInTransaction(.....) {
// 邏輯程式碼，非實際程式碼
// 1.傳送訊息
sendResult = this.send(msg);
// sendResult.getSendStatus() == SEND_OK
// 2.如果訊息傳送成功，處理與訊息關聯的本地事務單元
LocalTransactionState localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
// 3.結束事務
this.endTransaction(sendResult, localTransactionState, localException);
}
endTransaction方法會將請求發往broker(mq server)去更新事物訊息的最終狀態：
根據sendResult找到Prepared訊息
根據localTransaction更新訊息的最終狀態
如果endTransaction方法執行失敗，導致資料沒有傳送到broker，broker會有回查執行緒定時（預設1分鐘）掃描每個儲存事務狀態的表格檔案，如果是已經提交或者回滾的訊息直接跳過，如果是prepared狀態則會向Producer發起CheckTransaction請求，Producer會呼叫DefaultMQProducerImpl.checkTransactionState()方法來處理broker的定時回撥請求，而checkTransactionState會呼叫我們的事務設定的決斷方法，最後呼叫endTransactionOneway讓broker來更新訊息的最終狀態。
再回到轉賬的例子，如果Bob的賬戶的餘額已經減少，且訊息已經發送成功，Smith端開始消費這條訊息，這個時候就會出現消費失敗和消費超時兩個問題？解決超時問題的思路就是一直重試，直到消費端消費訊息成功，整個過程中有可能會出現訊息重複的問題，按照前面的思路解決即可。
消費事務訊息
這樣基本上可以解決超時問題，但是如果消費失敗怎麼辦？阿里提供給我們的解決方法是：人工解決。大家可以考慮一下，按照事務的流程，因為某種原因Smith加款失敗，需要回滾整個流程。如果訊息系統要實現這個回滾流程的話，系統複雜度將大大提升，且很容易出現Bug，估計出現Bug的概率會比消費失敗的概率大很多。我們需要衡量是否值得花這麼大的代價來解決這樣一個出現概率非常小的問題，這也是大家在解決疑難問題時需要多多思考的地方。
20160321補充：在3.2.6版本中移除了事務訊息的實現，所以此版本不支援事務訊息，具體情況請參考rocketmq的issues：
https://github.com/alibaba/RocketMQ/issues/65
https://github.com/alibaba/RocketMQ/issues/138
https://github.com/alibaba/RocketMQ/issues/156
四、Producer如何傳送訊息
Producer輪詢某topic下的所有佇列的方式來實現傳送方的負載均衡，如下圖所示：
producer傳送訊息負載均衡
首先分析一下RocketMQ的客戶端傳送訊息的原始碼：
// 構造Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 初始化Producer，整個應用生命週期內，只需要初始化1次
producer.start();
// 構造Message
Message msg = new Message("TopicTest1",// topic
"TagA",// tag：給訊息打標籤,用於區分一類訊息，可為null
"OrderID188",// key：自定義Key，可以用於去重，可為null
("Hello MetaQ").getBytes());// body：訊息內容
// 傳送訊息並返回結果
SendResult sendResult = producer.send(msg);
// 清理資源，關閉網路連線，登出自己
producer.shutdown();
在整個應用生命週期內，生產者需要呼叫一次start方法來初始化，初始化主要完成的任務有：
如果沒有指定namesrv地址，將會自動定址
啟動定時任務：更新namesrv地址、從namsrv更新topic路由資訊、清理已經掛掉的broker、向所有broker傳送心跳...
啟動負載均衡的服務
初始化完成後，開始傳送訊息，傳送訊息的主要程式碼如下：
private SendResult sendDefaultImpl(Message msg,......) {
// 檢查Producer的狀態是否是RUNNING
this.makeSureStateOK();
// 檢查msg是否合法：是否為null、topic,body是否為空、body是否超長
Validators.checkMessage(msg, this.defaultMQProducer);
// 獲取topic路由資訊
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
// 從路由資訊中選擇一個訊息佇列
MessageQueue mq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
// 將訊息傳送到該佇列上去
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}
程式碼中需要關注的兩個方法tryToFindTopicPublishInfo和selectOneMessageQueue。前面說過在producer初始化時，會啟動定時任務獲取路由資訊並更新到本地快取，所以tryToFindTopicPublishInfo會首先從快取中獲取topic路由資訊，如果沒有獲取到，則會自己去namesrv獲取路由資訊。selectOneMessageQueue方法通過輪詢的方式，返回一個佇列，以達到負載均衡的目的。
如果Producer傳送訊息失敗，會自動重試，重試的策略：
重試次數 < retryTimesWhenSendFailed（可配置）
總的耗時（包含重試n次的耗時） < sendMsgTimeout（傳送訊息時傳入的引數）
同時滿足上面兩個條件後，Producer會選擇另外一個佇列傳送訊息
五、訊息儲存
RocketMQ的訊息儲存是由consume queue和commit log配合完成的。
1、Consume Queue
consume queue是訊息的邏輯佇列，相當於字典的目錄，用來指定訊息在物理檔案commit log上的位置。
我們可以在配置中指定consumequeue與commitlog儲存的目錄
每個topic下的每個queue都有一個對應的consumequeue檔案，比如：
${rocketmq.home}/store/consumequeue/${topicName}/${queueId}/${fileName}
Consume Queue檔案組織，如圖所示：
Consume Queue檔案組織示意圖
根據topic和queueId來組織檔案，圖中TopicA有兩個佇列0,1，那麼TopicA和QueueId=0組成一個ConsumeQueue，TopicA和QueueId=1組成另一個ConsumeQueue。
按照消費端的GroupName來分組重試佇列，如果消費端消費失敗，訊息將被髮往重試佇列中，比如圖中的%RETRY%ConsumerGroupA。
按照消費端的GroupName來分組死信佇列，如果消費端消費失敗，並重試指定次數後，仍然失敗，則發往死信佇列，比如圖中的%DLQ%ConsumerGroupA。
死信佇列（Dead Letter Queue）一般用於存放由於某種原因無法傳遞的訊息，比如處理失敗或者已經過期的訊息。
Consume Queue中儲存單元是一個20位元組定長的二進位制資料，順序寫順序讀，如下圖所示：
consumequeue檔案儲存單元格式
CommitLog Offset是指這條訊息在Commit Log檔案中的實際偏移量
Size儲存中訊息的大小
Message Tag HashCode儲存訊息的Tag的雜湊值：主要用於訂閱時訊息過濾（訂閱時如果指定了Tag，會根據HashCode來快速查詢到訂閱的訊息）
2、Commit Log
CommitLog：訊息存放的物理檔案，每臺broker上的commitlog被本機所有的queue共享，不做任何區分。
檔案的預設位置如下，仍然可通過配置檔案修改：
${user.home} \store\${commitlog}\${fileName}
CommitLog的訊息儲存單元長度不固定，檔案順序寫，隨機讀。訊息的儲存結構如下表所示，按照編號順序以及編號對應的內容依次儲存。
Commit Log儲存單元結構圖
3、訊息儲存實現
訊息儲存實現，比較複雜，也值得大家深入瞭解，後面會單獨成文來分析，這小節只以程式碼說明一下具體的流程。
// Set the storage time
msg.setStoreTimestamp(System.currentTimeMillis());
// Set the message body BODY CRC (consider the most appropriate setting
msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
synchronized (this) {
long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
// Here settings are stored timestamp, in order to ensure an orderly global
msg.setStoreTimestamp(beginLockTimestamp);
// MapedFile：操作物理檔案在記憶體中的對映以及將記憶體資料持久化到物理檔案中
MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile();
// 將Message追加到檔案commitlog
result = mapedFile.appendMessage(msg, this.appendMessageCallback);
switch (result.getStatus()) {
case PUT_OK:break;
case END_OF_FILE:
// Create a new file, re-write the message
mapedFile = this.mapedFileQueue.getLastMapedFile();
result = mapedFile.appendMessage(msg, this.appendMessageCallback);
break;
DispatchRequest dispatchRequest = new DispatchRequest(
topic,// 1
queueId,// 2
result.getWroteOffset(),// 3
result.getWroteBytes(),// 4
tagsCode,// 5
msg.getStoreTimestamp(),// 6
result.getLogicsOffset(),// 7
msg.getKeys(),// 8
/**
* Transaction
*/
msg.getSysFlag(),// 9
msg.getPreparedTransactionOffset());// 10
// 1.分發訊息位置到ConsumeQueue
// 2.分發到IndexService建立索引
this.defaultMessageStore.putDispatchRequest(dispatchRequest);
}
4、訊息的索引檔案
如果一個訊息包含key值的話，會使用IndexFile儲存訊息索引，檔案的內容結構如圖：
訊息索引
索引檔案主要用於根據key來查詢訊息的，流程主要是：
根據查詢的 key 的 hashcode%slotNum 得到具體的槽的位置(slotNum 是一個索引檔案裡面包含的最大槽的數目，例如圖中所示 slotNum=5000000)
根據 slotValue(slot 位置對應的值)查詢到索引項列表的最後一項(倒序排列,slotValue 總是指向最新的一個索引項)
遍歷索引項列表返回查詢時間範圍內的結果集(預設一次最大返回的 32 條記錄)
六、訊息訂閱
RocketMQ訊息訂閱有兩種模式，一種是Push模式，即MQServer主動向消費端推送；另外一種是Pull模式，即消費端在需要時，主動到MQServer拉取。但在具體實現時，Push和Pull模式都是採用消費端主動拉取的方式。
首先看下消費端的負載均衡：
消費端負載均衡
消費端會通過RebalanceService執行緒，10秒鐘做一次基於topic下的所有佇列負載：
遍歷Consumer下的所有topic，然後根據topic訂閱所有的訊息
獲取同一topic和Consumer Group下的所有Consumer
然後根據具體的分配策略來分配消費佇列，分配的策略包含：平均分配、消費端配置等
如同上圖所示：如果有 5 個佇列，2 個 consumer，那麼第一個 Consumer 消費 3 個佇列，第二 consumer 消費 2 個佇列。這裡採用的就是平均分配策略，它類似於我們的分頁，TOPIC下面的所有queue就是記錄，Consumer的個數就相當於總的頁數，那麼每頁有多少條記錄，就類似於某個Consumer會消費哪些佇列。
通過這樣的策略來達到大體上的平均消費，這樣的設計也可以很方面的水平擴充套件Consumer來提高消費能力。
消費端的Push模式是通過長輪詢的模式來實現的，就如同下圖：
Push模式示意圖
Consumer端每隔一段時間主動向broker傳送拉訊息請求，broker在收到Pull請求後，如果有訊息就立即返回資料，Consumer端收到返回的訊息後，再回調消費者設定的Listener方法。如果broker在收到Pull請求時，訊息佇列裡沒有資料，broker端會阻塞請求直到有資料傳遞或超時才返回。
當然，Consumer端是通過一個執行緒將阻塞佇列LinkedBlockingQueue<PullRequest>中的PullRequest傳送到broker拉取訊息，以防止Consumer一致被阻塞。而Broker端，在接收到Consumer的PullRequest時，如果發現沒有訊息，就會把PullRequest扔到ConcurrentHashMap中快取起來。broker在啟動時，會啟動一個執行緒不停的從ConcurrentHashMap取出PullRequest檢查，直到有資料返回。
七、RocketMQ的其他特性
前面的6個特性都是基本上都是點到為止，想要深入瞭解，還需要大家多多檢視原始碼，多多在實際中運用。當然除了已經提到的特性外，RocketMQ還支援：
定時訊息
訊息的刷盤策略
主動同步策略：同步雙寫、非同步複製
海量訊息堆積能力
高效通訊
.......
其中涉及到的很多設計思路和解決方法都值得我們深入研究：
訊息的儲存設計：既要滿足海量訊息的堆積能力，又要滿足極快的查詢效率，還要保證寫入的效率。
高效的通訊元件設計：高吞吐量，毫秒級的訊息投遞能力都離不開高效的通訊。
.......
RocketMQ最佳實踐
一、Producer最佳實踐
1、一個應用盡可能用一個 Topic，訊息子型別用 tags 來標識，tags 可以由應用自由設定。只有傳送訊息設定了tags，消費方在訂閱訊息時，才可以利用 tags 在 broker 做訊息過濾。
2、每個訊息在業務層面的唯一標識碼，要設定到 keys 欄位，方便將來定位訊息丟失問題。由於是雜湊索引，請務必保證 key 儘可能唯一，這樣可以避免潛在的雜湊衝突。
3、訊息傳送成功或者失敗，要列印訊息日誌，務必要列印 sendresult 和 key 欄位。
4、對於訊息不可丟失應用，務必要有訊息重發機制。例如：訊息傳送失敗，儲存到資料庫，能有定時程式嘗試重發或者人工觸發重發。
5、某些應用如果不關注訊息是否傳送成功，請直接使用sendOneWay方法傳送訊息。
二、Consumer最佳實踐
1、消費過程要做到冪等（即消費端去重）
2、儘量使用批量方式消費方式，可以很大程度上提高消費吞吐量。
3、優化每條訊息消費過程
三、其他配置
線上應該關閉autoCreateTopicEnable，即在配置檔案中將其設定為false。
RocketMQ在傳送訊息時，會首先獲取路由資訊。如果是新的訊息，由於MQServer上面還沒有建立對應的Topic，這個時候，如果上面的配置開啟的話，會返回預設TOPIC的（RocketMQ會在每臺broker上面建立名為TBW102的TOPIC）路由資訊，然後Producer會選擇一臺Broker傳送訊息，選中的broker在儲存訊息時，發現訊息的topic還沒有建立，就會自動建立topic。後果就是：以後所有該TOPIC的訊息，都將傳送到這臺broker上，達不到負載均衡的目的。
所以基於目前RocketMQ的設計，建議關閉自動建立TOPIC的功能，然後根據訊息量的大小，手動建立TOPIC。
RocketMQ設計相關
RocketMQ的設計假定：
每臺PC機器都可能宕機不可服務
任意叢集都有可能處理能力不足
最壞的情況一定會發生
內網環境需要低延遲來提供最佳使用者體驗
RocketMQ的關鍵設計：
分散式叢集化
強資料安全
海量資料堆積
毫秒級投遞延遲（推拉模式）
這是RocketMQ在設計時的假定前提以及需要到達的效果。我想這些假定適用於所有的系統設計。隨著我們系統的服務的增多，每位開發者都要注意自己的程式是否存在單點故障，如果掛了應該怎麼恢復、能不能很好的水平擴充套件、對外的介面是否足夠高效、自己管理的資料是否足夠安全...... 多多規範自己的設計，才能開發出高效健壯的程式。
附錄：RocketMQ涉及到的幾個專業術語和整體架構介紹
一、RocketMQ中的專業術語
Topic
topic表示訊息的第一級型別，比如一個電商系統的訊息可以分為：交易訊息、物流訊息...... 一條訊息必須有一個Topic。
Tag
Tag表示訊息的第二級型別，比如交易訊息又可以分為：交易建立訊息，交易完成訊息..... 一條訊息可以沒有Tag。RocketMQ提供2級訊息分類，方便大家靈活控制。
Queue
一個topic下，我們可以設定多個queue(訊息佇列)。當我們傳送訊息時，需要要指定該訊息的topic。RocketMQ會輪詢該topic下的所有佇列，將訊息傳送出去。
Producer 與 Producer Group
Producer表示訊息佇列的生產者。訊息佇列的本質就是實現了publish-subscribe模式，生產者生產訊息，消費者消費訊息。所以這裡的Producer就是用來生產和傳送訊息的，一般指業務系統。
Producer Group是一類Producer的集合名稱，這類Producer通常傳送一類訊息，且傳送邏輯一致。
Consumer 與 Consumer Group
訊息消費者，一般由後臺系統非同步消費訊息。
Push Consumer
Consumer 的一種，應用通常向 Consumer 物件註冊一個 Listener 介面，一旦收到訊息，Consumer 物件立刻回撥 Listener 介面方法。
Pull Consumer
Consumer 的一種，應用通常主動呼叫 Consumer 的拉訊息方法從 Broker 拉訊息，主動權由應用控制。
Consumer Group是一類Consumer的集合名稱，這類Consumer通常消費一類訊息，且消費邏輯一致。
Broker
訊息的中轉者，負責儲存和轉發訊息。可以理解為訊息佇列伺服器，提供了訊息的接收、儲存、拉取和轉發服務。broker是RocketMQ的核心，它不不能掛的，所以需要保證broker的高可用。
廣播消費
一條訊息被多個Consumer消費，即使這些Consumer屬於同一個Consumer Group，訊息也會被Consumer Group中的每個Consumer都消費一次。在廣播消費中的Consumer Group概念可以認為在訊息劃分方面無意義。
叢集消費
一個Consumer Group中的Consumer例項平均分攤消費訊息。例如某個Topic有 9 條訊息，其中一個Consumer Group有 3 個例項(可能是 3 個程序,或者 3 臺機器)，那麼每個例項只消費其中的 3 條訊息。
NameServer
NameServer即名稱服務，兩個功能：
接收broker的請求，註冊broker的路由資訊
介面client的請求，根據某個topic獲取其到broker的路由資訊
NameServer沒有狀態，可以橫向擴充套件。每個broker在啟動的時候會到NameServer註冊；Producer在傳送訊息前會根據topic到NameServer獲取路由(到broker)資訊；Consumer也會定時獲取topic路由資訊。
二、RocketMQ Overview
rocketmq overview
Producer向一些佇列輪流傳送訊息，佇列集合稱為Topic，Consumer如果做廣播消費，則一個consumer例項消費這個Topic對應的所有佇列；如果做叢集消費，則多個Consumer例項平均消費這個Topic對應的佇列集合。
再看下RocketMQ物理部署結構圖：
RocketMQ網路部署圖
RocketMQ網路部署特點：
Name Server 是一個幾乎無狀態節點，可叢集部署，節點之間無任何資訊同步。
Broker部署相對複雜，Broker分為Master與Slave，一個Master可以對應多個Slave，但是一個Slave只能對應一個Master，Master與Slave的對應關係通過指定相同的BrokerName，不同的BrokerId來定義，BrokerId=0表示Master，非0表示Slave。Master也可以部署多個。每個Broker與Name Server叢集中的所有節點建立長連線，定時註冊Topic資訊到所有Name Server。
Producer與Name Server叢集中的其中一個節點（隨機選擇）建立長連線，定期從Name Server取Topic路由資訊，並向提供Topic 服務的Master建立長連線，且定時向Master傳送心跳。Producer 完全無狀態，可叢集部署。
Consumer與Name Server叢集中的其中一個節點（隨機選擇）建立長連線，定期從Name Server取Topic 路由資訊，並向提供Topic服務的Master、Slave建立長連線，且定時向Master、Slave傳送心跳。Consumer既可以從Master訂閱訊息，也可以從Slave訂閱訊息，訂閱規則由Broker配置決定。
三、其他參考資料
RocketMQ使用者指南
RocketMQ原理簡介
RocketMQ最佳實踐
阿里分散式開放訊息服務(ONS)原理與實踐2
阿里分散式開放訊息服務(ONS)原理與實踐3
RocketMQ原理解析
備註：水平有限，難免疏漏，如果問題請留言