0、彙總

RocketMQ入門到入土（一）新手也能看懂的原理和實戰！

RocketMQ入門到入土（二）事務訊息&順序訊息

從入門到入土（三）RocketMQ 怎麼保證的訊息不丟失？

RocketMQ入門到入土（四）producer生產訊息原始碼剖析

RocketMQ入門到入土（五）訊息持久化儲存原始碼解析

RocketMQ入門到入土（六）發訊息的時候選擇queue的演算法有哪些？

RocketMQ入門到入土（七 )為什麼同一個消費組設定不同tag會出現奇怪現象

從入門到入土（八）RocketMQ的Consumer是如何做的負載均衡的

從入門到入土（九）手摸手教你搭建RocketMQ雙主雙從同步叢集，不信學不會！

從入門到入土（十）RocketMQ叢集流程以及核心概念

1、說說你們公司線上生產環境用的是什麼訊息中介軟體?

見【2、多個mq如何選型？】

2、多個mq如何選型？

3、為什麼要使用MQ？

因為專案比較大，做了分散式系統，所有遠端服務呼叫請求都是同步執行經常出問題，所以引入了mq

4、RocketMQ由哪些角色組成，每個角色作用和特點是什麼？

5、RocketMQ中的Topic和JMS的queue有什麼區別？

queue就是來源於資料結構的FIFO佇列。而Topic是個抽象的概念，每個Topic底層對應N個queue，而資料也真實存在queue上的。

6、RocketMQ Broker中的訊息被消費後會立即刪除嗎？

不會，每條訊息都會持久化到CommitLog中，每個Consumer連線到Broker後會維持消費進度資訊，當有訊息消費後只是當前Consumer的消費進度（CommitLog的offset）更新了。

追問：那麼訊息會堆積嗎？什麼時候清理過期訊息？

4.6版本預設48小時後會刪除不再使用的CommitLog檔案

• 檢查這個檔案最後訪問時間
• 判斷是否大於過期時間
• 指定時間刪除，預設凌晨4點

原始碼如下：

/**
 * {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#isTimeToDelete()}
 */
private boolean isTimeToDelete() {
    // when = "04";
    String when = DefaultMessageStore.this.getMessageStoreConfig().getDeleteWhen();
    // 是04點，就返回true
    if (UtilAll.isItTimeToDo(when)) {
        return true;
    }
 // 不是04點，返回false
    return false;
}

/**
 * {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#deleteExpiredFiles()}
 */
private void deleteExpiredFiles() {
    // isTimeToDelete()這個方法是判斷是不是凌晨四點，是的話就執行刪除邏輯。
    if (isTimeToDelete()) {
        // 預設是72，但是broker配置檔案預設改成了48，所以新版本都是48。
        long fileReservedTime = 48 * 60 * 60 * 1000;
        deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(72 * 60 * 60 * 1000, xx, xx, xx);
    }
}
                                                                       
/**
 * {@link org.apache.rocketmq.store.CommitLog#deleteExpiredFile()}
 */
public int deleteExpiredFile(xxx) {
    // 這個方法的主邏輯就是遍歷查詢最後更改時間+過期時間，小於當前系統時間的話就刪了（也就是小於48小時）。
    return this.mappedFileQueue.deleteExpiredFileByTime(72 * 60 * 60 * 1000, xx, xx, xx);
}

7、RocketMQ消費模式有幾種？

消費模型由Consumer決定，消費維度為Topic。

• 叢集消費

1.一條訊息只會被同Group中的一個Consumer消費

2.多個Group同時消費一個Topic時，每個Group都會有一個Consumer消費到資料

• 廣播消費

訊息將對一 個Consumer Group 下的各個 Consumer 例項都消費一遍。即即使這些 Consumer 屬於同一個Consumer Group ，訊息也會被 Consumer Group 中的每個 Consumer 都消費一次。

8、消費訊息是push還是pull？

RocketMQ沒有真正意義的push，都是pull，雖然有push類，但實際底層實現採用的是長輪詢機制，即拉取方式

broker端屬性 longPollingEnable 標記是否開啟長輪詢。預設開啟

原始碼如下：

// {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}
// 看到沒，這是一隻披著羊皮的狼，名字叫PushConsumerImpl，實際乾的確是pull的活。

// 拉取訊息，結果放到pullCallback裡
this.pullAPIWrapper.pullKernelImpl(pullCallback);

追問：為什麼要主動拉取訊息而不使用事件監聽方式？

事件驅動方式是建立好長連線，由事件（傳送資料）的方式來實時推送。

如果broker主動推送訊息的話有可能push速度快，消費速度慢的情況，那麼就會造成訊息在consumer端堆積過多，同時又不能被其他consumer消費的情況。而pull的方式可以根據當前自身情況來pull，不會造成過多的壓力而造成瓶頸。所以採取了pull的方式。

9、broker如何處理拉取請求的？

Consumer首次請求Broker

• Broker中是否有符合條件的訊息
• 有 ->
• 響應Consumer
• 等待下次Consumer的請求
• 沒有
• DefaultMessageStore#ReputMessageService#run方法
• PullRequestHoldService 來Hold連線，每個5s執行一次檢查pullRequestTable有沒有訊息，有的話立即推送
• 每隔1ms檢查commitLog中是否有新訊息，有的話寫入到pullRequestTable
• 當有新訊息的時候返回請求
• 掛起consumer的請求，即不斷開連線，也不返回資料
• 使用consumer的offset，

10、RocketMQ如何做負載均衡？

通過Topic在多Broker中分散式儲存實現。

producer端

傳送端指定message queue傳送訊息到相應的broker，來達到寫入時的負載均衡

• 提升寫入吞吐量，當多個producer同時向一個broker寫入資料的時候，效能會下降
• 訊息分佈在多broker中，為負載消費做準備

預設策略是隨機選擇：

• producer維護一個index
• 每次取節點會自增
• index向所有broker個數取餘
• 自帶容錯策略

其他實現：

• SelectMessageQueueByHash
• hash的是傳入的args
• SelectMessageQueueByRandom
• SelectMessageQueueByMachineRoom 沒有實現

也可以自定義實現MessageQueueSelector介面中的select方法

MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);

consumer端

採用的是平均分配演算法來進行負載均衡。

其他負載均衡演算法

平均分配策略(預設)(AllocateMessageQueueAveragely) 環形分配策略(AllocateMessageQueueAveragelyByCircle) 手動配置分配策略(AllocateMessageQueueByConfig) 機房分配策略(AllocateMessageQueueByMachineRoom) 一致性雜湊分配策略(AllocateMessageQueueConsistentHash) 靠近機房策略(AllocateMachineRoomNearby)

追問：當消費負載均衡consumer和queue不對等的時候會發生什麼？

Consumer和queue會優先平均分配，如果Consumer少於queue的個數，則會存在部分Consumer消費多個queue的情況，如果Consumer等於queue的個數，那就是一個Consumer消費一個queue，如果Consumer個數大於queue的個數，那麼會有部分Consumer空餘出來，白白的浪費了。

11、訊息重複消費

影響訊息正常傳送和消費的重要原因是網路的不確定性。

引起重複消費的原因

• ACK

正常情況下在consumer真正消費完訊息後應該傳送ack，通知broker該訊息已正常消費，從queue中剔除

當ack因為網路原因無法傳送到broker，broker會認為詞條訊息沒有被消費，此後會開啟訊息重投機制把訊息再次投遞到consumer

• 消費模式

在CLUSTERING模式下，訊息在broker中會保證相同group的consumer消費一次，但是針對不同group的consumer會推送多次

解決方案

• 資料庫表

處理訊息前，使用訊息主鍵在表中帶有約束的欄位中insert

• Map

單機時可以使用map ConcurrentHashMap -> putIfAbsent   guava cache

• Redis

分散式鎖搞起來。

12、如何讓RocketMQ保證訊息的順序消費

你們線上業務用訊息中介軟體的時候，是否需要保證訊息的順序性?

如果不需要保證訊息順序，為什麼不需要?假如我有一個場景要保證訊息的順序，你們應該如何保證?

首先多個queue只能保證單個queue裡的順序，queue是典型的FIFO，天然順序。多個queue同時消費是無法絕對保證訊息的有序性的。所以總結如下：

同一topic，同一個QUEUE，發訊息的時候一個執行緒去傳送訊息，消費的時候 一個執行緒去消費一個queue裡的訊息。

追問：怎麼保證訊息發到同一個queue？

Rocket MQ給我們提供了MessageQueueSelector介面，可以自己重寫裡面的介面，實現自己的演算法，舉個最簡單的例子：判斷i % 2 == 0，那就都放到queue1裡，否則放到queue2裡。

for (int i = 0; i < 5; i++) {
    Message message = new Message("orderTopic", ("hello!" + i).getBytes());
    producer.send(
        // 要發的那條訊息
        message,
        // queue 選擇器 ，向 topic中的哪個queue去寫訊息
        new MessageQueueSelector() {
            // 手動 選擇一個queue
            @Override
            public MessageQueue select(
                // 當前topic 裡面包含的所有queue
                List<MessageQueue> mqs,
                // 具體要發的那條訊息
                Message msg,
                // 對應到 send（） 裡的 args，也就是2000前面的那個0
                Object arg) {
                // 向固定的一個queue裡寫訊息，比如這裡就是向第一個queue裡寫訊息
                if (Integer.parseInt(arg.toString()) % 2 == 0) {
                    return mqs.get(0);
                } else {
                    return mqs.get(1);
                }
            }
        },
        // 自定義引數：0
        // 2000代表2000毫秒超時時間
        i, 2000);
}

13、RocketMQ如何保證訊息不丟失

首先在如下三個部分都可能會出現丟失訊息的情況：

• Producer端
• Broker端
• Consumer端

13.1、Producer端如何保證訊息不丟失

• 採取send()同步發訊息，傳送結果是同步感知的。
• 傳送失敗後可以重試，設定重試次數。預設3次。

producer.setRetryTimesWhenSendFailed(10);

• 叢集部署，比如傳送失敗了的原因可能是當前Broker宕機了，重試的時候會發送到其他Broker上。

13.2、Broker端如何保證訊息不丟失

• 修改刷盤策略為同步刷盤。預設情況下是非同步刷盤的。

flushDiskType = SYNC_FLUSH

• 叢集部署，主從模式，高可用。

13.3、Consumer端如何保證訊息不丟失

• 完全消費正常後在進行手動ack確認。

14、rocketMQ的訊息堆積如何處理

下游消費系統如果宕機了，導致幾百萬條訊息在訊息中介軟體裡積壓，此時怎麼處理?

你們線上是否遇到過訊息積壓的生產故障?如果沒遇到過，你考慮一下如何應對?

首先要找到是什麼原因導致的訊息堆積，是Producer太多了，Consumer太少了導致的還是說其他情況，總之先定位問題。

然後看下訊息消費速度是否正常，正常的話，可以通過上線更多consumer臨時解決訊息堆積問題

追問：如果Consumer和Queue不對等，上線了多臺也在短時間內無法消費完堆積的訊息怎麼辦？

• 準備一個臨時的topic

• queue的數量是堆積的幾倍

• queue分佈到多Broker中

• 上線一臺Consumer做訊息的搬運工，把原來Topic中的訊息挪到新的Topic裡，不做業務邏輯處理，只是挪過去

• 上線N臺Consumer同時消費臨時Topic中的資料

• 改bug

• 恢復原來的Consumer，繼續消費之前的Topic

追問：堆積時間過長訊息超時了？

RocketMQ中的訊息只會在commitLog被刪除的時候才會消失，不會超時。也就是說未被消費的訊息不會存在超時刪除這情況。

追問：堆積的訊息會不會進死信佇列？

不會，訊息在消費失敗後會進入重試佇列（%RETRY%+ConsumerGroup），18次（預設18次，網上所有文章都說是16次，無一例外。但是我沒搞懂為啥是16次，這不是18個時間嗎 ？）才會進入死信佇列（%DLQ%+ConsumerGroup）。

原始碼如下：

public class MessageStoreConfig {
    // 每隔如下時間會進行重試，到最後一次時間重試失敗的話就進入死信隊列了。
 private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
}

15、RocketMQ在分散式事務支援這塊機制的底層原理?

你們用的是RocketMQ?RocketMQ很大的一個特點是對分散式事務的支援，你說說他在分散式事務支援這塊機制的底層原理?

分散式系統中的事務可以使用TCC（Try、Confirm、Cancel）、2pc來解決分散式系統中的訊息原子性

RocketMQ 4.3+提供分佈事務功能，通過 RocketMQ 事務訊息能達到分散式事務的最終一致

RocketMQ實現方式：

**Half Message：**預處理訊息，當broker收到此類訊息後，會儲存到RMQ_SYS_TRANS_HALF_TOPIC的訊息消費佇列中

**檢查事務狀態：**Broker會開啟一個定時任務，消費RMQ_SYS_TRANS_HALF_TOPIC佇列中的訊息，每次執行任務會向訊息傳送者確認事務執行狀態（提交、回滾、未知），如果是未知，Broker會定時去回撥在重新檢查。

**超時：**如果超過回查次數，預設回滾訊息。

也就是他並未真正進入Topic的queue，而是用了臨時queue來放所謂的half message，等提交事務後才會真正的將half message轉移到topic下的queue。

16、如果讓你來動手實現一個分散式訊息中介軟體，整體架構你會如何設計實現?

我個人覺得從以下幾個點回答吧：

• 需要考慮能快速擴容、天然支援叢集
• 持久化的姿勢
• 高可用性
• 資料0丟失的考慮
• 服務端部署簡單、client端使用簡單

17、看過RocketMQ 的原始碼沒有。如果看過，說說你對RocketMQ 原始碼的理解?

要真讓我說，我會吐槽蠻爛的，首先沒任何註釋，可能是之前阿里巴巴寫了中文註釋，捐贈給apache後，apache覺得中文註釋不能留，自己又懶得寫英文註釋，就都給刪了。裡面比較典型的設計模式有單例、工廠、策略、門面模式。單例工廠無處不在，策略印象深刻比如發訊息和消費訊息的時候queue的負載均衡就是N個策略演算法類，有隨機、hash等，這也是能夠快速擴容天然支援叢集的必要原因之一。持久化做的也比較完善，採取的CommitLog來落盤，同步非同步兩種方式。

18、高吞吐量下如何優化生產者和消費者的效能?

開發

• 同一group下，多機部署，並行消費

• 單個Consumer提高消費執行緒個數

• 批量消費

• 訊息批量拉取
• 業務邏輯批量處理

運維

• 網絡卡調優
• jvm調優
• 多執行緒與cpu調優
• Page Cache

19、再說說RocketMQ 是如何保證資料的高容錯性的?

• 在不開啟容錯的情況下，輪詢佇列進行傳送，如果失敗了，重試的時候過濾失敗的Broker
• 如果開啟了容錯策略，會通過RocketMQ的預測機制來預測一個Broker是否可用
• 如果上次失敗的Broker可用那麼還是會選擇該Broker的佇列
• 如果上述情況失敗，則隨機選擇一個進行傳送
• 在傳送訊息的時候會記錄一下呼叫的時間與是否報錯，根據該時間去預測broker的可用時間

其實就是send訊息的時候queue的選擇。原始碼在如下：

org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue()

20、任何一臺Broker突然宕機了怎麼辦？

Broker主從架構以及多副本策略。Master收到訊息後會同步給Slave，這樣一條訊息就不止一份了，Master宕機了還有slave中的訊息可用，保證了MQ的可靠性和高可用性。而且Rocket MQ4.5.0開始就支援了Dlegder模式，基於raft的，做到了真正意義的HA。

21、Broker把自己的資訊註冊到哪個NameServer上？

這麼問明顯在坑你，因為Broker會向所有的NameServer上註冊自己的資訊，而不是某一個，是每一個，全