可以發(fā)現(xiàn)，RocketMQ是把Topic分片存儲到各個Broker節(jié)點中，然后在把Broker節(jié)點中的Topic繼續(xù)分片為若干等分的ConsumeQueue，從而提高消息的吞吐量。ConsumeQueue是作為負載均衡資源分配的基本單元。

這樣把Topic的消息分區(qū)到了不同的Broker上，從而增加了消息隊列的數(shù)量，從而能夠支持更塊的并發(fā)消費速度(只要有足夠的消費者)。

4.3.2 Broker自動創(chuàng)建Topic會有什么問題？

假設設置為通過Broker自動創(chuàng)建Topic(autoCreateTopicEnable=true)，并且Producer端設置Topic消息隊列數(shù)量設置為4，也就是默認值：

producer.setDefaultTopicQueueNums(4);

嘗試往一個新的 topic itzhai-test-queue-1連續(xù)發(fā)送10條消息，發(fā)送完畢之后，查看Topic狀態(tài)：

我們可以發(fā)現(xiàn)，在兩個broker上面都創(chuàng)建了itzhai-test-queue-a，并且每個broker上的消息隊列數(shù)量都為4。怎么回事，我配置的明明是期望創(chuàng)建4個隊列，為什么加起來會變成了8個？如下圖所示：

由于時間關系，本文我們不會帶大家從源碼方面去解讀為啥會出現(xiàn)這種情況，接下來我們通過一種更加直觀的方式來驗證下這個問題：繼續(xù)做實驗。

我們繼續(xù)嘗試往一個新的 topic itzhai-test-queue-10發(fā)送1條消息，注意，這一次不做并發(fā)發(fā)送了，只發(fā)送一條，發(fā)送完畢之后，查看Topic狀態(tài)：

可以發(fā)現(xiàn)，這次創(chuàng)建的消息隊列數(shù)量又是對的了，并且都是在broker-a上面創(chuàng)建的。接下來，無論怎么并發(fā)發(fā)送消息，消息隊列的數(shù)量都不會繼續(xù)增加了。

其實這也是并發(fā)請求Broker，觸發(fā)自動創(chuàng)建Topic的bug。

為了更加嚴格的管理Topic的創(chuàng)建和分片配置，一般在生產(chǎn)環(huán)境都是配置為手動創(chuàng)建Topic，通過提交運維工單申請創(chuàng)建Topic以及Topic的數(shù)據(jù)分配。

接下來我們來看看RocketMQ的特性。更多其他技術的底層架構內(nèi)幕分析，請訪問我的博客IT宅(itzhai.com)或者關注Java架構雜談公眾號。

5. RocketMQ特性

5.1 生產(chǎn)端

5.1.1 消息發(fā)布

RocketMQ中定義了如下三種消息通信的方式：

public enum CommunicationMode {    SYNC,    ASYNC,    ONEWAY,}

• SYNC：同步發(fā)送，生產(chǎn)端會阻塞等待發(fā)送結果；
  • 應用場景：這種方式應用場景非常廣泛，如重要業(yè)務事件通知。
• ASYNC：異步發(fā)送，生產(chǎn)端調(diào)用發(fā)送API之后，立刻返回，在拿到Broker的響應結果后，觸發(fā)對應的SendCallback回調(diào)；
  • 應用場景：一般用于鏈路耗時較長，對 RT 較為敏感的業(yè)務場景；
• ONEWAY：單向發(fā)送，發(fā)送方只負責發(fā)送消息，不等待服務器回應且沒有回調(diào)函數(shù)觸發(fā)，即只發(fā)送請求不等待應答。 此方式發(fā)送消息的過程耗時非常短，一般在微秒級別；
  • 應用場景：適用于耗時非常短，對可靠性要求不高的場景，如日志收集。

SYNC和ASYNC關注發(fā)送結果，ONEWAY不關注發(fā)送結果。發(fā)送結果如下：

public enum SendStatus {    SEND_OK,    FLUSH_DISK_TIMEOUT,    FLUSH_SLAVE_TIMEOUT,    SLAVE_NOT_AVAILABLE,}

• SEND_OK：消息發(fā)送成功。SEND_OK并不意味著投遞是可靠的，要確保消息不丟失，需要開啟SYNC_MASTER同步或者SYNC_FLUSH同步寫；
• FLUSH_DISK_TIMEOUT：消息發(fā)送成功，但是刷盤超時。如果Broker的flushDiskType=SYNC_FLUSH，并且5秒內(nèi)沒有完成消息的刷盤，則會返回這個狀態(tài)；
• FLUSH_SLAVE_TIMEOUT：消息發(fā)送成功，但是服務器同步到Slave時超時。如果Broker的brokerRole=SYNC_MASTER，并且5秒內(nèi)沒有完成同步，則會返回這個狀態(tài)；
• SLAVE_NOT_AVAILABLE：消息發(fā)送成功，但是無可用的Slave節(jié)點。如果Broker的brokerRole=SYNC_MASTER，但是沒有發(fā)現(xiàn)SLAVE節(jié)點或者SLAVE節(jié)點掛掉了，那么會返回這個狀態(tài)。

源碼內(nèi)容更精彩，歡迎大家進一步閱讀源碼詳細了解消息發(fā)送的內(nèi)幕：

• 同步發(fā)送：org.apache.rocketmq.client.producer.DefaultMQProducer#send(org.apache.rocketmq.common.message.Message)
• 異步發(fā)送：org.apache.rocketmq.client.producer.DefaultMQProducer#send(org.apache.rocketmq.common.message.Message, org.apache.rocketmq.client.producer.SendCallback)
• 單向發(fā)送：org.apache.rocketmq.client.producer.DefaultMQProducer#sendOneway(org.apache.rocketmq.common.message.Message)

5.1.2 順序消費

消息的有序性指的是一類消息消費的時候，可以按照發(fā)送順序來消費，比如：在Java架構雜談茶餐廳吃飯產(chǎn)生的消息：進入餐廳、點餐、下單、上菜、付款，消息要按照這個順序消費才有意義，但是多個顧客產(chǎn)生的消息是可以并行消費的。順序消費又分為全局順序消費和分區(qū)順序消費：

• 全局順序：同一個Topic下的消息，所有消息按照嚴格的FIFO順序進行發(fā)布和消費。適用于：性能要求不高，所有消息嚴格按照FIFO進行發(fā)布和消費的場景；
• 分區(qū)順序：同一個Topic下，根據(jù)消息的特定業(yè)務ID進行sharding key分區(qū)，同一個分區(qū)內(nèi)的消息按照嚴格的FIFO順序進行發(fā)布和消費。適用于：性能要求高，在同一個分區(qū)中嚴格按照FIFO進行發(fā)布和消費的場景。

一般情況下，生產(chǎn)者是會以輪訓的方式把消息發(fā)送到Topic的消息隊列中的：

在同一個Queue里面，消息的順序性是可以得到保證的，但是如果一個Topic有多個Queue，以輪訓的方式投遞消息，那么就會導致消息亂序了。

為了保證消息的順序性，需要把保持順序性的消息投遞到同一個Queue中。

5.1.2.1 如何保證消息投遞的順序性

RocketMQ提供了MessageQueueSelector接口，可以用來實現(xiàn)自定義的選擇投遞的消息隊列的算法：

for (int i = 0; i < orderList.size(); i++) {    String content = "Hello itzhai.com. Java架構雜談," + new Date();    Message msg = new Message("topic-itzhai-com", tags[i % tags.length], "KEY" + i,            content.getBytes(RemotingHelper.DEFAULT_CHARSET));    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {        @Override        public MessageQueue select(List mqs, Message msg, Object arg) {            Long orderId = (Long) arg;            // 訂單號與消息隊列個數(shù)取模，保證讓同一個訂單號的消息落入同一個消息隊列            long index = orderId % mqs.size();            return mqs.get((int) index);        }    }, orderList.get(i).getOrderId());    System.out.printf("content: %s, sendResult: %s%n", content, sendResult);}

如上圖，我們實現(xiàn)了MessageQueueSelector接口，并在實現(xiàn)的select方法里面，指定了選擇消息隊列的算法：訂單號與消息隊列個數(shù)取模，保證讓同一個訂單號的消息落入同一個消息隊列：

有個異常場景需要考慮：假設某一個Master節(jié)點掛掉了，導致Topic的消息隊列數(shù)量發(fā)生了變化，那么繼續(xù)使用以上的選擇算法，就會導致在這個過程中同一個訂單的消息會分散到不同的消息隊列里面，最終導致消息不能順序消費。

為了避免這種情況，只能選擇犧牲failover特性了。

現(xiàn)在投遞到消息隊列中的消息保證了順序，那如何保證消費也是順序的呢？

5.1.2.2 如何保證消息消費的順序性？

RocketMQ中提供了MessageListenerOrderly，該對象用于有順序收異步傳遞的消息，一個隊列對應一個消費線程，使用方法如下：

consumer.registerMessageListener(new MessageListenerOrderly() {    // 消費次數(shù)，用于輔助模擬各種消費結果    AtomicLong consumeTimes = new AtomicLong(0);    @Override    public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {        context.setAutoCommit(true);        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);        this.consumeTimes.incrementAndGet();        if ((this.consumeTimes.get() % 2) == 0) {            return ConsumeOrderlyStatus.SUCCESS;        } else if ((this.consumeTimes.get() % 3) == 0) {            return ConsumeOrderlyStatus.ROLLBACK;        } else if ((this.consumeTimes.get() % 4) == 0) {            return ConsumeOrderlyStatus.COMMIT;        } else if ((this.consumeTimes.get() % 5) == 0) {            context.setSuspendCurrentQueueTimeMillis(3000);            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;        }        return ConsumeOrderlyStatus.SUCCESS;    }});

如果您使用的是MessageListenerConcurrently，表示并發(fā)消費，為了保證消息消費的順序性，需要設置為單線程模式。

使用MessageListenerOrderly的問題：如果遇到某條消息消費失敗，并且無法跳過，那么消息隊列的消費進度就會停滯。

5.1.3 延遲隊列(定時消息)

定時消費是指消息發(fā)送到Broker之后不會立即被消費，而是等待特定的時間之后才投遞到Topic中。定時消息會暫存在名為SCHEDULE_TOPIC_XXXX的topic中，并根據(jù)delayTimeLevel存入特定的queue，queueId=delayTimeLevel-1，一個queue只存相同延遲的消息，保證具有相同延遲的消息能夠順序消費。比如，我們設置1秒后把消息投遞到topic-itzhai-comtopic，則存儲的文件目錄如下所示：

Broker會調(diào)度地消費SCHEDULE_TOPIC_XXXX，將消息寫入真實的topic。

定時消息的副作用：定時消息會在第一次寫入Topic和調(diào)度寫入實際的topic都會進行計數(shù)，因此發(fā)送數(shù)量，tps都會變高。

使用延遲隊列的場景：提交了訂單之后，如果等待超過約定的時間還未支付，則把訂單設置為超時狀態(tài)。

RocketMQ提供了以下幾個固定的延遲級別：

public class MessageStoreConfig {    ...    // 10個level，level：1~18    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";    ...}

level = 0 表示不使用延遲消息。

另外，消息消費失敗也會進入延遲隊列，消息發(fā)送時間與設置的延遲級別和重試次數(shù)有關。

以下是發(fā)送延遲消息的代碼：

public class ScheduledMessageProducer {    public static void main(String[] args) throws Exception {        DefaultMQProducer producer = new DefaultMQProducer("TestProducerGroup");        producer.start();        int totalMessagesToSend = 100;        for (int i = 0; i < totalMessagesToSend; i++) {            Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());            // 指定該消息在10秒后被消費者消費            message.setDelayTimeLevel(3);            producer.send(message);        }        producer.shutdown();    }}

5.1.4 數(shù)據(jù)完整性與事務消息

通過消息對系統(tǒng)進行解耦之后，勢必會遇到分布式系統(tǒng)數(shù)據(jù)完整性的問題。

5.1.4.1 實現(xiàn)分布式事務的手段有哪些？

我們可以通過以下手段解決分布式系統(tǒng)數(shù)據(jù)最終一致性問題：

• 數(shù)據(jù)庫層面的2PC(Two-phase commit protocol)，二階段提交，同步阻塞，效率低下，存在協(xié)調(diào)者單點故障問題，極端情況下存在數(shù)據(jù)不一致的風險。對應技術上的XA、JTA/JTS。這是分布式環(huán)境下事務處理的典型模式；
• 數(shù)據(jù)庫層面的3PC，三階段提交，引入了參與者超時機制，增加了預提交階段，使得故障恢復之后協(xié)調(diào)者的決策復雜度降低，但整體的交互過程變得更長了，性能有所下降，仍舊會存在數(shù)據(jù)不一致的問題；
• 業(yè)務層面的TCC ，Try - Confirm - Cancel。對業(yè)務的侵入較大，和業(yè)務緊耦合，對于每一個操作都需要定義三個動作分別對應：Try - Confirm - Cancel，將資源層的兩階段提交協(xié)議轉(zhuǎn)換到業(yè)務層，成為業(yè)務模型中的一部分；
• 本地消息表；
• 事務消息；

RocketMQ事務消息（Transactional Message）則是通過事務消息來實現(xiàn)分布式事務的最終一致性。下面看看RocketMQ是如何實現(xiàn)事務消息的。

5.1.4.2 RocketMQ如何實現(xiàn)事務消息？

如下圖：

事務消息有兩個流程：

1. 事務消息發(fā)送及提交：
   1. 發(fā)送half消息；
   2. 服務端響應half消息寫入結果；
   3. 根據(jù)half消息的發(fā)送結果執(zhí)行本地事務。如果發(fā)送失敗，此時half消息對業(yè)務不可見，本地事務不執(zhí)行；
   4. 根據(jù)本地事務狀態(tài)執(zhí)行Commit或者Rollback。Commit操作會觸發(fā)生成ConsumeQueue索引，此時消息對消費者可見；
2. 補償流程：
   5. 對于沒有Commit/Rollback的事務消息，會處于pending狀態(tài)，這對這些消息，MQ Server發(fā)起一次回查；
   6. Producer收到回查消息，檢查回查消息對應的本地事務的轉(zhuǎn)塔體；
   7. 根據(jù)本地事務狀態(tài)，重新執(zhí)行Commit或者Rollback。

補償階段主要用于解決消息的Commit或者Rollback發(fā)生超時或者失敗的情況。

half消息：并不是發(fā)送了一半的消息，而是指消息已經(jīng)發(fā)送到了MQ Server，但是該消息未收到生產(chǎn)者的二次確認，此時該消息暫時不能投遞到具體的ConsumeQueue中，這種狀態(tài)的消息稱為half消息。

5.1.4.3 RocketMQ事務消息是如何存儲的？

發(fā)送到MQ Server的half消息對消費者是不可見的，為此，RocketMQ會先把half消息的Topic和Queue信息存儲到消息的屬性中，然后把該half消息投遞到一個專門的處理事務消息的隊列中：RMQ_SYS_TRANS_HALF_TOPIC，由于消費者沒有訂閱該Topic，所以無法消息half類型的消息。

生產(chǎn)者執(zhí)行Commit half消息的時候，會存儲一條專門的Op消息，用于標識事務消息已確定的狀態(tài)，如果一條事務消息還沒有對應的Op消息，說明這個事務的狀態(tài)還無法確定。RocketMQ會開啟一個定時任務，對于pending狀態(tài)的消息，會先向生產(chǎn)者發(fā)送回查事務狀態(tài)請求，根據(jù)事務狀態(tài)來決定是否提交或者回滾消息。

當消息被標記為Commit狀態(tài)之后，會把half消息的Topic和Queue相關屬性還原為原來的值，最終構建實際的消費索引(ConsumeQueue)。

RocketMQ并不會無休止的嘗試消息事務狀態(tài)回查，默認查找15次，超過了15次還是無法獲取事務狀態(tài)，RocketMQ默認回滾該消息。并打印錯誤日志，可以通過重寫AbstractTransactionalMessageCheckListener類修改這個行為。

可以通過Broker的配置參數(shù)：transactionCheckMax來修改此值。

5.1.5 消息重投

如果消息發(fā)布方式是同步發(fā)送會重投，如果是異步發(fā)送會重試。

消息重投可以盡可能保證消息投遞成功，但是可能會造成消息重復。

什么情況會造成重復消費消息？

• 出現(xiàn)消息量大，網(wǎng)絡抖動的時候；
• 生產(chǎn)者主動重發(fā)；
• 消費負載發(fā)生變化。

可以使用的消息重試策略：

• retryTimesWhenSendFailed：設置同步發(fā)送失敗的重投次數(shù)，默認為2。所以生產(chǎn)者最多會嘗試發(fā)送retryTimesWhenSendFailed+1次。
  • 為了最大程度保證消息不丟失，重投的時候會嘗試向其他broker發(fā)送消息；
  • 超過重投次數(shù)，拋出異常，讓客戶端自行處理；
  • 觸發(fā)重投的異常：RemotingException、MQClientException和部分MQBrokerException；
• retryTimesWhenSendAsyncFailed：設置異步發(fā)送失敗重試次數(shù)，異步重試不會選擇其他Broker，不保證消息不丟失；
• retryAnotherBrokerWhenNotStoreOK：消息刷盤(主或備)超時或slave不可用(返回狀態(tài)非SEND_OK)，是否嘗試發(fā)送到其他broker，默認false。重要的消息可以開啟此選項。

oneway發(fā)布方式不支持重投。

5.1.6 批量消息

為了提高系統(tǒng)的吞吐量，提高發(fā)送效率，可以使用批量發(fā)送消息。

批量發(fā)送消息的限制：

• 同一批批量消息的topic，waitStoreMsgOK屬性必須保持一致；
• 批量消息不支持延遲隊列；
• 批量消息一次課發(fā)送的上限是4MB。

發(fā)送批量消息的例子：

String topic = "itzhai-test-topic";List messages = new ArrayList<>();messages.add(new Message(topic, "TagA", "OrderID001", "Hello world itzhai.com 0".getBytes()));messages.add(new Message(topic, "TagA", "OrderID002", "Hello world itzhai.com 1".getBytes()));messages.add(new Message(topic, "TagA", "OrderID003", "Hello world itzhai.com 2".getBytes()));producer.send(messages);

如果發(fā)送的消息比較多，會增加復雜性，為此，可以對大消息進行拆分。以下是拆分的例子：

public class ListSplitter implements Iterator> {     // 限制最大大小    private final int SIZE_LIMIT = 1024 * 1024 * 4;    private final List messages;    private int currIndex;    public ListSplitter(List messages) {         this.messages = messages;    }    @Override public boolean hasNext() {        return currIndex < messages.size();     }    @Override public List next() {         int startIndex = getStartIndex();        int nextIndex = startIndex;        int totalSize = 0;        for (; nextIndex < messages.size(); nextIndex++) {            Message message = messages.get(nextIndex);             int tmpSize = calcMessageSize(message);            if (tmpSize + totalSize > SIZE_LIMIT) {                break;             } else {                totalSize += tmpSize;             }        }        List subList = messages.subList(startIndex, nextIndex);         currIndex = nextIndex;        return subList;    }    private int getStartIndex() {        Message currMessage = messages.get(currIndex);         int tmpSize = calcMessageSize(currMessage);         while(tmpSize > SIZE_LIMIT) {            currIndex += 1;            Message message = messages.get(curIndex);             tmpSize = calcMessageSize(message);        }        return currIndex;     }    private int calcMessageSize(Message message) {        int tmpSize = message.getTopic().length() + message.getBody().length();         Map properties = message.getProperties();        for (Map.Entry entry : properties.entrySet()) {            tmpSize += entry.getKey().length() + entry.getValue().length();         }        tmpSize = tmpSize + 20; // Increase the log overhead by 20 bytes        return tmpSize;     }}// then you could split the large list into small ones:ListSplitter splitter = new ListSplitter(messages);while (splitter.hasNext()) {   try {       List  listItem = splitter.next();       producer.send(listItem);   } catch (Exception e) {       e.printStackTrace();       // handle the error   }}

5.1.7 消息過濾

RocketMQ的消費者可以根據(jù)Tag進行消息過濾來獲取自己感興趣的消息，也支持自定義屬性過濾。

Tags是Topic下的次級消息類型/二級類型（注：Tags也支持TagA || TagB這樣的表達式），可以在同一個Topic下基于Tags進行消息過濾。

消息過濾是在Broker端實現(xiàn)的，減少了對Consumer無用消息的網(wǎng)絡傳輸，缺點是增加了Broker負擔，實現(xiàn)相對復雜。

5.2 消費端

5.2.1 消費模型

消費端有兩周消費模型：集群消費和廣播消費。

集群消費

集群消費模式下,相同Consumer Group的每個Consumer實例平均分攤消息。

廣播消費

廣播消費模式下，相同Consumer Group的每個Consumer實例都接收全量的消息。

5.2.2 消息重試

RocketMQ會為每個消費組都設置一個Topic名稱為%RETRY%consumerGroupName的重試隊列（這里需要注意的是，這個Topic的重試隊列是針對消費組，而不是針對每個Topic設置的），用于暫時保存因為各種異常而導致Consumer端無法消費的消息。

考慮到異?；謴推饋硇枰恍r間，會為重試隊列設置多個重試級別，每個重試級別都有與之對應的重新投遞延時，重試次數(shù)越多投遞延時就越大。

RocketMQ對于重試消息的處理是先保存至Topic名稱為SCHEDULE_TOPIC_XXXX的延遲隊列中，后臺定時任務按照對應的時間進行Delay后重新保存至%RETRY%consumerGroupName的重試隊列中。

比如，我們設置1秒后把消息投遞到topic-itzhai-comtopic，則存儲的文件目錄如下所示：

5.2.3 死信隊列

當一條消息初次消費失敗，消息隊列會自動進行消息重試；達到最大重試次數(shù)后，若消費依然失敗，則表明消費者在正常情況下無法正確地消費該消息，此時，消息隊列不會立刻將消息丟棄，而是將其發(fā)送到該消費者對應的特殊隊列中。

RocketMQ將這種正常情況下無法被消費的消息稱為死信消息（Dead-Letter Message），將存儲死信消息的特殊隊列稱為死信隊列（Dead-Letter Queue）。

在RocketMQ中，可以通過使用console控制臺對死信隊列中的消息進行重發(fā)來使得消費者實例再次進行消費。

由于RocketMQ是使用Java寫的，所以它的代碼特別適合拿來閱讀消遣，我們繼續(xù)來看看RocketMQ的源碼結構...

不不，還是算了，一下子又到周末晚上了，時間差不多了，今天就寫到這里了。有空再聊。

我精心整理了一份Redis寶典給大家，涵蓋了Redis的方方面面，面試官懂的里面有，面試官不懂的里面也有，有了它，不怕面試官連環(huán)問，就怕面試官一上來就問你Redis的Redo Log是干啥的？畢竟這種問題我也不會。

在Java架構雜談公眾號發(fā)送Redis關鍵字獲取pdf文件：

本文作者： arthinking

博客鏈接： https://www.itzhai.com/articles/deep-understanding-of-rocketmq.html

高并發(fā)異步解耦利器：RocketMQ究竟強在哪里？

版權聲明： 版權歸作者所有，未經(jīng)許可不得轉(zhuǎn)載，侵權必究！聯(lián)系作者請加公眾號。

References

apache/rocketmq. Retrieved from https://github.com/apache/rocketmq