面試官問： 如何保證 MQ消息是有序的？

大致過程：

• 生產(chǎn)端，創(chuàng)建一條消息，通過網(wǎng)絡發(fā)送到MQ Server

• MQ將 消息存儲在topic 的一個分區(qū)里

• 消費端，從分區(qū)中拉取消息，消費處理

但現(xiàn)實往往不一樣！MQ 架構設計要滿足高并發(fā)、高性能、高可用等指標

單分區(qū)，達不到我們的吞吐量要求，我們考慮采用多分區(qū)架構設計，正所謂 ”三個臭皮匠賽過一個諸葛亮“，多分區(qū)可以有效分攤全局壓力，提升整體系統(tǒng)性能。

兩臺 MQ機器，組成一個集群，原先一個分區(qū)存儲6條消息，現(xiàn)在分攤到兩個分區(qū)，每個分區(qū)各存儲3條消息，性能比上面那個提升一倍。

貌似可以滿足我們的需求，但任何事情都有兩面性！

我們看看下面業(yè)務場景：

一個用戶在電商網(wǎng)站上下訂單到交易完成，中間會經(jīng)歷一系列動作，訂單的狀態(tài)也會隨之變化，一個訂單會產(chǎn)生多條MQ消息，下單、付款、發(fā)貨、買家確認收貨，消費端需要嚴格按照業(yè)務狀態(tài)機的順序處理，否則，就會出現(xiàn)業(yè)務問題。

我們發(fā)現(xiàn)，消息帶上了狀態(tài)，不再是一個個獨立的個體，有了上下文依賴關系！

對于這個問題，突然想到HTTP協(xié)議，其本身也是無狀態(tài)的，也就是說前后兩次請求沒有關聯(lián)，但有些業(yè)務功能有登錄要求，那怎么解決？

引入Cookie機制，每次請求客戶端額外傳輸一些數(shù)據(jù)，來達到上下文關聯(lián)。

回到MQ的消息順序問題，我們要如何解決？

答案：各退一步，保證局部有序。

比如上面的電商例子，只要保證一個訂單的多條狀態(tài)消息在同一個分區(qū)，便可以滿足業(yè)務需求，這個方案可以覆蓋大部分的業(yè)務場景。

這里面只需要有一個路由策略組件，由它決定消息該放到哪個分區(qū)中！

考慮到市面MQ開源框架很多，常見的如：Kafka、Pulsar、RabbitMQ、RocketMQ 等，API方法略有區(qū)別，但設計思路是相通的。

接下來，我們以 RocketMQ 為例：

生產(chǎn)端提供了一個接口 MessageQueueSelector

public interface MessageQueueSelector {
   MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

接口內(nèi)定義一個select方法，具體參數(shù)含義：

• mqs：該Topic下所有的隊列分片

• msg：待發(fā)送的消息

• arg：發(fā)送消息時傳遞的參數(shù)

關于MessageQueueSelector接口，RocketMQ 框架提供了三個默認實現(xiàn)類：

• 1、SelectMessageQueueByHash：

arg參數(shù)的hashcode的絕對值，然后對mqs.size()取余，得到目標隊列在mqs的下標

• 2、SelectMessageQueueByRandom：

對mqs.size()值取隨機數(shù)作為目標隊列在mqs的下標

• 3、SelectMessageQueueByMachineRoom

返回null

特別注意：

雖然保證了單個分片的消息有序，但每個分片的消費者只能是單線程處理，因為多線程無法控制消費順序。這個可能會損失一些性能。

這里又引出另一個問題，如何保證一個隊列只能有一個消費端呢？

1、

org.apache.rocketmq.client.impl.consumer.RebalanceImpl#updateProcessQueueTableInRebalance

• 遍歷一個topic下所有的MessageQueue

• isOrder && !this.lock(mq) 嘗試對它加鎖，確保一個MessageQueue只能被一個消費者處理

2、將PullRequest對象放入PullMessageService的pullRequestQueue隊列中

public void dispatchPullRequest(List<PullRequest> pullRequestList) {
    for (PullRequest pullRequest : pullRequestList) {
        this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
        log.info("doRebalance, {}, add a new pull request {}", consumerGroup, pullRequest);
    }
}

3、org.apache.rocketmq.client.impl.consumer.PullMessageService#run

• PullMessageService 是一個Runnable線程任務

• 無限循環(huán)，從隊列中拉取、處理消息

另一個問題，如何保證一個隊列，只有一個線程在處理消息呢？

1、 DefaultMQPushConsumerImpl#pullMessage

• ConsumeMessageService 中有兩個實現(xiàn)類，因為我們有消費順序要求，會選擇ConsumeMessageOrderlyService來處理業(yè)務

2、 ConsumeMessageOrderlyService.ConsumeRequest