RocketMQ在面試中那些常見問題及答案+匯總
點擊上方?好好學java?,選擇?星標?公眾號
重磅資訊、干貨,第一時間送達
重磅資訊、干貨,第一時間送達
今日推薦:硬剛一周,3W字總結(jié),一年的經(jīng)驗告訴你如何準備校招!
個人原創(chuàng)100W+訪問量博客:點擊前往,查看更多
個人原創(chuàng)100W+訪問量博客:點擊前往,查看更多
1、說說你們公司線上生產(chǎn)環(huán)境用的是什么消息中間件?
見【2、多個mq如何選型?】
2、多個mq如何選型?
| MQ | 描述 |
|---|---|
| RabbitMQ | erlang開發(fā),對消息堆積的支持并不好,當大量消息積壓的時候,會導致 RabbitMQ 的性能急劇下降。每秒鐘可以處理幾萬到十幾萬條消息。 |
| RocketMQ | java開發(fā),面向互聯(lián)網(wǎng)集群化功能豐富,對在線業(yè)務(wù)的響應(yīng)時延做了很多的優(yōu)化,大多數(shù)情況下可以做到毫秒級的響應(yīng),每秒鐘大概能處理幾十萬條消息。 |
| Kafka | Scala開發(fā),面向日志功能豐富,性能最高。當你的業(yè)務(wù)場景中,每秒鐘消息數(shù)量沒有那么多的時候,Kafka 的時延反而會比較高。所以,Kafka 不太適合在線業(yè)務(wù)場景。 |
| ActiveMQ | java開發(fā),簡單,穩(wěn)定,性能不如前面三個。小型系統(tǒng)用也ok,但是不推薦。推薦用互聯(lián)網(wǎng)主流的。 |
3、為什么要使用MQ?
因為項目比較大,做了分布式系統(tǒng),所有遠程服務(wù)調(diào)用請求都是同步執(zhí)行經(jīng)常出問題,所以引入了mq
| 作用 | 描述 |
|---|---|
| 解耦 | 系統(tǒng)耦合度降低,沒有強依賴關(guān)系 |
| 異步 | 不需要同步執(zhí)行的遠程調(diào)用可以有效提高響應(yīng)時間 |
| 削峰 | 請求達到峰值后,后端service還可以保持固定消費速率消費,不會被壓垮 |
4、RocketMQ由哪些角色組成,每個角色作用和特點是什么?
| 角色 | 作用 |
|---|---|
| Nameserver | 無狀態(tài),動態(tài)列表;這也是和zookeeper的重要區(qū)別之一。zookeeper是有狀態(tài)的。 |
| Producer | 消息生產(chǎn)者,負責發(fā)消息到Broker。 |
| Broker | 就是MQ本身,負責收發(fā)消息、持久化消息等。 |
| Consumer | 消息消費者,負責從Broker上拉取消息進行消費,消費完進行ack。 |
5、RocketMQ中的Topic和JMS的queue有什么區(qū)別?
queue就是來源于數(shù)據(jù)結(jié)構(gòu)的FIFO隊列。而Topic是個抽象的概念,每個Topic底層對應(yīng)N個queue,而數(shù)據(jù)也真實存在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()這個方法是判斷是不是凌晨四點,是的話就執(zhí)行刪除邏輯。
????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)?{
????//?這個方法的主邏輯就是遍歷查找最后更改時間+過期時間,小于當前系統(tǒng)時間的話就刪了(也就是小于48小時)。
????return?this.mappedFileQueue.deleteExpiredFileByTime(72?*?60?*?60?*?1000,?xx,?xx,?xx);
}
7、RocketMQ消費模式有幾種?
消費模型由Consumer決定,消費維度為Topic。
集群消費
1.一條消息只會被同Group中的一個Consumer消費
2.多個Group同時消費一個Topic時,每個Group都會有一個Consumer消費到數(shù)據(jù)
廣播消費
消息將對一 個Consumer Group 下的各個 Consumer 實例都消費一遍。即即使這些 Consumer 屬于同一個Consumer Group ,消息也會被 Consumer Group 中的每個 Consumer 都消費一次。
8、消費消息是push還是pull?
RocketMQ沒有真正意義的push,都是pull,雖然有push類,但實際底層實現(xiàn)采用的是長輪詢機制,即拉取方式
broker端屬性 longPollingEnable 標記是否開啟長輪詢。默認開啟
源碼如下:
//?{@link?org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}
//?看到?jīng)],這是一只披著羊皮的狼,名字叫PushConsumerImpl,實際干的確是pull的活。
//?拉取消息,結(jié)果放到pullCallback里
this.pullAPIWrapper.pullKernelImpl(pullCallback);
追問:為什么要主動拉取消息而不使用事件監(jiān)聽方式?
事件驅(qū)動方式是建立好長連接,由事件(發(fā)送數(shù)據(jù))的方式來實時推送。
如果broker主動推送消息的話有可能push速度快,消費速度慢的情況,那么就會造成消息在consumer端堆積過多,同時又不能被其他consumer消費的情況。而pull的方式可以根據(jù)當前自身情況來pull,不會造成過多的壓力而造成瓶頸。所以采取了pull的方式。
9、broker如何處理拉取請求的?
Consumer首次請求Broker
Broker中是否有符合條件的消息 有 -> 響應(yīng)Consumer 等待下次Consumer的請求 沒有 DefaultMessageStore#ReputMessageService#run方法 PullRequestHoldService 來Hold連接,每個5s執(zhí)行一次檢查pullRequestTable有沒有消息,有的話立即推送 每隔1ms檢查commitLog中是否有新消息,有的話寫入到pullRequestTable 當有新消息的時候返回請求 掛起consumer的請求,即不斷開連接,也不返回數(shù)據(jù) 使用consumer的offset,
10、RocketMQ如何做負載均衡?
通過Topic在多Broker中分布式存儲實現(xiàn)。
producer端
發(fā)送端指定message queue發(fā)送消息到相應(yīng)的broker,來達到寫入時的負載均衡
提升寫入吞吐量,當多個producer同時向一個broker寫入數(shù)據(jù)的時候,性能會下降 消息分布在多broker中,為負載消費做準備
默認策略是隨機選擇:
producer維護一個index 每次取節(jié)點會自增 index向所有broker個數(shù)取余 自帶容錯策略
其他實現(xiàn):
SelectMessageQueueByHash hash的是傳入的args SelectMessageQueueByRandom SelectMessageQueueByMachineRoom 沒有實現(xiàn)
也可以自定義實現(xiàn)MessageQueueSelector接口中的select方法
MessageQueue?select(final?List?mqs,?final?Message?msg,?final?Object?arg);
consumer端
采用的是平均分配算法來進行負載均衡。
其他負載均衡算法
平均分配策略(默認)(AllocateMessageQueueAveragely) 環(huán)形分配策略(AllocateMessageQueueAveragelyByCircle) 手動配置分配策略(AllocateMessageQueueByConfig) 機房分配策略(AllocateMessageQueueByMachineRoom) 一致性哈希分配策略(AllocateMessageQueueConsistentHash) 靠近機房策略(AllocateMachineRoomNearby)
追問:當消費負載均衡consumer和queue不對等的時候會發(fā)生什么?
Consumer和queue會優(yōu)先平均分配,如果Consumer少于queue的個數(shù),則會存在部分Consumer消費多個queue的情況,如果Consumer等于queue的個數(shù),那就是一個Consumer消費一個queue,如果Consumer個數(shù)大于queue的個數(shù),那么會有部分Consumer空余出來,白白的浪費了。
11、消息重復消費
影響消息正常發(fā)送和消費的重要原因是網(wǎng)絡(luò)的不確定性。
引起重復消費的原因
ACK
正常情況下在consumer真正消費完消息后應(yīng)該發(fā)送ack,通知broker該消息已正常消費,從queue中剔除
當ack因為網(wǎng)絡(luò)原因無法發(fā)送到broker,broker會認為詞條消息沒有被消費,此后會開啟消息重投機制把消息再次投遞到consumer
消費模式
在CLUSTERING模式下,消息在broker中會保證相同group的consumer消費一次,但是針對不同group的consumer會推送多次
解決方案
數(shù)據(jù)庫表
處理消息前,使用消息主鍵在表中帶有約束的字段中insert
Map
單機時可以使用map ConcurrentHashMap -> putIfAbsent ? guava cache
Redis
分布式鎖搞起來。
12、如何讓RocketMQ保證消息的順序消費
你們線上業(yè)務(wù)用消息中間件的時候,是否需要保證消息的順序性?
如果不需要保證消息順序,為什么不需要?假如我有一個場景要保證消息的順序,你們應(yīng)該如何保證?
首先多個queue只能保證單個queue里的順序,queue是典型的FIFO,天然順序。多個queue同時消費是無法絕對保證消息的有序性的。所以總結(jié)如下:
同一topic,同一個QUEUE,發(fā)消息的時候一個線程去發(fā)送消息,消費的時候 一個線程去消費一個queue里的消息。
追問:怎么保證消息發(fā)到同一個queue?
Rocket MQ給我們提供了MessageQueueSelector接口,可以自己重寫里面的接口,實現(xiàn)自己的算法,舉個最簡單的例子:判斷i % 2 == 0,那就都放到queue1里,否則放到queue2里。
for?(int?i?=?0;?i?5;?i++)?{
????Message?message?=?new?Message("orderTopic",?("hello!"?+?i).getBytes());
????producer.send(
????????//?要發(fā)的那條消息
????????message,
????????//?queue?選擇器?,向?topic中的哪個queue去寫消息
????????new?MessageQueueSelector()?{
????????????//?手動?選擇一個queue
????????????@Override
????????????public?MessageQueue?select(
????????????????//?當前topic?里面包含的所有queue
????????????????List?mqs,
????????????????//?具體要發(fā)的那條消息
????????????????Message?msg,
????????????????//?對應(yīng)到?send()?里的?args,也就是2000前面的那個0
????????????????Object?arg) ?{
????????????????//?向固定的一個queue里寫消息,比如這里就是向第一個queue里寫消息
????????????????if?(Integer.parseInt(arg.toString())?%?2?==?0)?{
????????????????????return?mqs.get(0);
????????????????}?else?{
????????????????????return?mqs.get(1);
????????????????}
????????????}
????????},
????????//?自定義參數(shù):0
????????//?2000代表2000毫秒超時時間
????????i,?2000);
}
13、RocketMQ如何保證消息不丟失
首先在如下三個部分都可能會出現(xiàn)丟失消息的情況:
Producer端 Broker端 Consumer端
13.1、Producer端如何保證消息不丟失
采取send()同步發(fā)消息,發(fā)送結(jié)果是同步感知的。 發(fā)送失敗后可以重試,設(shè)置重試次數(shù)。默認3次。
producer.setRetryTimesWhenSendFailed(10);
集群部署,比如發(fā)送失敗了的原因可能是當前Broker宕機了,重試的時候會發(fā)送到其他Broker上。
13.2、Broker端如何保證消息不丟失
修改刷盤策略為同步刷盤。默認情況下是異步刷盤的。
flushDiskType = SYNC_FLUSH
集群部署,主從模式,高可用。
13.3、Consumer端如何保證消息不丟失
完全消費正常后在進行手動ack確認。
14、rocketMQ的消息堆積如何處理
下游消費系統(tǒng)如果宕機了,導致幾百萬條消息在消息中間件里積壓,此時怎么處理?
你們線上是否遇到過消息積壓的生產(chǎn)故障?如果沒遇到過,你考慮一下如何應(yīng)對?
首先要找到是什么原因?qū)е碌南⒍逊e,是Producer太多了,Consumer太少了導致的還是說其他情況,總之先定位問題。
然后看下消息消費速度是否正常,正常的話,可以通過上線更多consumer臨時解決消息堆積問題
追問:如果Consumer和Queue不對等,上線了多臺也在短時間內(nèi)無法消費完堆積的消息怎么辦?
準備一個臨時的topic
queue的數(shù)量是堆積的幾倍
queue分布到多Broker中
上線一臺Consumer做消息的搬運工,把原來Topic中的消息挪到新的Topic里,不做業(yè)務(wù)邏輯處理,只是挪過去
上線N臺Consumer同時消費臨時Topic中的數(shù)據(jù)
改bug
恢復原來的Consumer,繼續(xù)消費之前的Topic
追問:堆積時間過長消息超時了?
RocketMQ中的消息只會在commitLog被刪除的時候才會消失,不會超時。也就是說未被消費的消息不會存在超時刪除這情況。
追問:堆積的消息會不會進死信隊列?
不會,消息在消費失敗后會進入重試隊列(%RETRY%+ConsumerGroup),18次(默認18次,網(wǎng)上所有文章都說是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在分布式事務(wù)支持這塊機制的底層原理?
你們用的是RocketMQ?RocketMQ很大的一個特點是對分布式事務(wù)的支持,你說說他在分布式事務(wù)支持這塊機制的底層原理?
分布式系統(tǒng)中的事務(wù)可以使用TCC(Try、Confirm、Cancel)、2pc來解決分布式系統(tǒng)中的消息原子性
RocketMQ 4.3+提供分布事務(wù)功能,通過 RocketMQ 事務(wù)消息能達到分布式事務(wù)的最終一致
RocketMQ實現(xiàn)方式:
**Half Message:**預處理消息,當broker收到此類消息后,會存儲到RMQ_SYS_TRANS_HALF_TOPIC的消息消費隊列中
**檢查事務(wù)狀態(tài):**Broker會開啟一個定時任務(wù),消費RMQ_SYS_TRANS_HALF_TOPIC隊列中的消息,每次執(zhí)行任務(wù)會向消息發(fā)送者確認事務(wù)執(zhí)行狀態(tài)(提交、回滾、未知),如果是未知,Broker會定時去回調(diào)在重新檢查。
**超時:**如果超過回查次數(shù),默認回滾消息。
也就是他并未真正進入Topic的queue,而是用了臨時queue來放所謂的half message,等提交事務(wù)后才會真正的將half message轉(zhuǎn)移到topic下的queue。
16、如果讓你來動手實現(xiàn)一個分布式消息中間件,整體架構(gòu)你會如何設(shè)計實現(xiàn)?
我個人覺得從以下幾個點回答吧:
需要考慮能快速擴容、天然支持集群 持久化的姿勢 高可用性 數(shù)據(jù)0丟失的考慮 服務(wù)端部署簡單、client端使用簡單
17、看過RocketMQ 的源碼沒有。如果看過,說說你對RocketMQ 源碼的理解?
要真讓我說,我會吐槽蠻爛的,首先沒任何注釋,可能是之前阿里巴巴寫了中文注釋,捐贈給apache后,apache覺得中文注釋不能留,自己又懶得寫英文注釋,就都給刪了。里面比較典型的設(shè)計模式有單例、工廠、策略、門面模式。單例工廠無處不在,策略印象深刻比如發(fā)消息和消費消息的時候queue的負載均衡就是N個策略算法類,有隨機、hash等,這也是能夠快速擴容天然支持集群的必要原因之一。持久化做的也比較完善,采取的CommitLog來落盤,同步異步兩種方式。
18、高吞吐量下如何優(yōu)化生產(chǎn)者和消費者的性能?
開發(fā)
同一group下,多機部署,并行消費
單個Consumer提高消費線程個數(shù)
批量消費
消息批量拉取 業(yè)務(wù)邏輯批量處理
運維
網(wǎng)卡調(diào)優(yōu) jvm調(diào)優(yōu) 多線程與cpu調(diào)優(yōu) Page Cache
19、再說說RocketMQ 是如何保證數(shù)據(jù)的高容錯性的?
在不開啟容錯的情況下,輪詢隊列進行發(fā)送,如果失敗了,重試的時候過濾失敗的Broker 如果開啟了容錯策略,會通過RocketMQ的預測機制來預測一個Broker是否可用 如果上次失敗的Broker可用那么還是會選擇該Broker的隊列 如果上述情況失敗,則隨機選擇一個進行發(fā)送 在發(fā)送消息的時候會記錄一下調(diào)用的時間與是否報錯,根據(jù)該時間去預測broker的可用時間
其實就是send消息的時候queue的選擇。源碼在如下:
org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue()
20、任何一臺Broker突然宕機了怎么辦?
Broker主從架構(gòu)以及多副本策略。Master收到消息后會同步給Slave,這樣一條消息就不止一份了,Master宕機了還有slave中的消息可用,保證了MQ的可靠性和高可用性。而且Rocket MQ4.5.0開始就支持了Dlegder模式,基于raft的,做到了真正意義的HA。
21、Broker把自己的信息注冊到哪個NameServer上?
這么問明顯在坑你,因為Broker會向所有的NameServer上注冊自己的信息,而不是某一個,是每一個,全部!
最后,給大家準備了一套算法學習教程,從小白到大神,都是這樣走過來的,建議學習一下,拿走不謝!
下載方式
1.?首先掃描下方二維碼
2.?后臺回復「A110」即可獲取

