RocketMQ在面試中那些常見問題及答案+匯總

0、匯總

RocketMQ入門到入土(一)新手也能看懂的原理和實戰!

RocketMQ入門到入土(二)事務消息&順序消息

從入門到入土(三)RocketMQ 怎麼保證的消息不丟失?

RocketMQ入門到入土(四)producer生產消息源碼剖析

RocketMQ入門到入土(五)消息持久化存儲源碼解析

RocketMQ入門到入土(六)發消息的時候選擇queue的演算法有哪些?

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

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

從入門到入土(九)手摸手教你搭建RocketMQ雙主雙從同步集群,不信學不會!

從入門到入土(十)RocketMQ集群流程以及核心概念

1、說說你們公司線上生產環境用的是什麼消息中間件?

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

2、多個mq如何選型?

MQ 描述
RabbitMQ erlang開發,對消息堆積的支援並不好,當大量消息積壓的時候,會導致 RabbitMQ 的性能急劇下降。每秒鐘可以處理幾萬到十幾萬條消息。
RocketMQ java開發,面向互聯網集群化功能豐富,對在線業務的響應時延做了很多的優化,大多數情況下可以做到毫秒級的響應,每秒鐘大概能處理幾十萬條消息。
Kafka Scala開發,面向日誌功能豐富,性能最高。當你的業務場景中,每秒鐘消息數量沒有那麼多的時候,Kafka 的時延反而會比較高。所以,Kafka 不太適合在線業務場景。
ActiveMQ java開發,簡單,穩定,性能不如前面三個。小型系統用也ok,但是不推薦。推薦用互聯網主流的。

3、為什麼要使用MQ?

因為項目比較大,做了分散式系統,所有遠程服務調用請求都是同步執行經常出問題,所以引入了mq

作用 描述
解耦 系統耦合度降低,沒有強依賴關係
非同步 不需要同步執行的遠程調用可以有效提高響應時間
削峰 請求達到峰值後,後端service還可以保持固定消費速率消費,不會被壓垮

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

角色 作用
Nameserver 無狀態,動態列表;這也是和zookeeper的重要區別之一。zookeeper是有狀態的。
Producer 消息生產者,負責發消息到Broker。
Broker 就是MQ本身,負責收發消息、持久化消息等。
Consumer 消息消費者,負責從Broker上拉取消息進行消費,消費完進行ack。

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上註冊自己的資訊,而不是某一個,是每一個,全部!