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