繼之前的mysql奪命連環(huán)之后，我發(fā)現(xiàn)我這個標(biāo)題被好多套用的，什么奪命zookeeper，奪命多線程一大堆，這一次，開始面試題系列MQ專題，消息隊列作為日常常見的使用中間件，面試也是必問的點之一，一起來看看MQ的面試題。

你們?yōu)槭裁词褂胢q？具體的使用場景是什么？

mq的作用很簡單，削峰填谷。以電商交易下單的場景來說，正向交易的過程可能涉及到創(chuàng)建訂單、扣減庫存、扣減活動預(yù)算、扣減積分等等。每個接口的耗時如果是100ms，那么理論上整個下單的鏈路就需要耗費400ms，這個時間顯然是太長了。

如果這些操作全部同步處理的話，首先調(diào)用鏈路太長影響接口性能，其次分布式事務(wù)的問題很難處理，這時候像扣減預(yù)算和積分這種對實時一致性要求沒有那么高的請求，完全就可以通過mq異步的方式去處理了。同時，考慮到異步帶來的不一致的問題，我們可以通過job去重試保證接口調(diào)用成功，而且一般公司都會有核對的平臺，比如下單成功但是未扣減積分的這種問題可以通過核對作為兜底的處理方案。

使用mq之后我們的鏈路變簡單了，同時異步發(fā)送消息我們的整個系統(tǒng)的抗壓能力也上升了。

那你們使用什么mq？基于什么做的選型？

我們主要調(diào)研了幾個主流的mq，kafka、rabbitmq、rocketmq、activemq，選型我們主要基于以下幾個點去考慮：

由于我們系統(tǒng)的qps壓力比較大，所以性能是首要考慮的要素。

開發(fā)語言，由于我們的開發(fā)語言是java，主要是為了方便二次開發(fā)。

對于高并發(fā)的業(yè)務(wù)場景是必須的，所以需要支持分布式架構(gòu)的設(shè)計。

功能全面，由于不同的業(yè)務(wù)場景，可能會用到順序消息、事務(wù)消息等。

基于以上幾個考慮，我們最終選擇了RocketMQ。

你上面提到異步發(fā)送，那消息可靠性怎么保證？

消息丟失可能發(fā)生在生產(chǎn)者發(fā)送消息、MQ本身丟失消息、消費者丟失消息3個方面。

生產(chǎn)者丟失

生產(chǎn)者丟失消息的可能點在于程序發(fā)送失敗拋異常了沒有重試處理，或者發(fā)送的過程成功但是過程中網(wǎng)絡(luò)閃斷MQ沒收到，消息就丟失了。

由于同步發(fā)送的一般不會出現(xiàn)這樣使用方式，所以我們就不考慮同步發(fā)送的問題，我們基于異步發(fā)送的場景來說。

異步發(fā)送分為兩個方式：異步有回調(diào)和異步無回調(diào)，無回調(diào)的方式，生產(chǎn)者發(fā)送完后不管結(jié)果可能就會造成消息丟失，而通過異步發(fā)送+回調(diào)通知+本地消息表的形式我們就可以做出一個解決方案。以下單的場景舉例。

下單后先保存本地數(shù)據(jù)和MQ消息表，這時候消息的狀態(tài)是發(fā)送中，如果本地事務(wù)失敗，那么下單失敗，事務(wù)回滾。

下單成功，直接返回客戶端成功，異步發(fā)送MQ消息

MQ回調(diào)通知消息發(fā)送結(jié)果，對應(yīng)更新數(shù)據(jù)庫MQ發(fā)送狀態(tài)

JOB輪詢超過一定時間（時間根據(jù)業(yè)務(wù)配置）還未發(fā)送成功的消息去重試

在監(jiān)控平臺配置或者JOB程序處理超過一定次數(shù)一直發(fā)送不成功的消息，告警，人工介入。

一般而言，對于大部分場景來說異步回調(diào)的形式就可以了，只有那種需要完全保證不能丟失消息的場景我們做一套完整的解決方案。

MQ丟失

如果生產(chǎn)者保證消息發(fā)送到MQ，而MQ收到消息后還在內(nèi)存中，這時候宕機了又沒來得及同步給從節(jié)點，就有可能導(dǎo)致消息丟失。

比如RocketMQ：

RocketMQ分為同步刷盤和異步刷盤兩種方式，默認(rèn)的是異步刷盤，就有可能導(dǎo)致消息還未刷到硬盤上就丟失了，可以通過設(shè)置為同步刷盤的方式來保證消息可靠性，這樣即使MQ掛了，恢復(fù)的時候也可以從磁盤中去恢復(fù)消息。

比如Kafka也可以通過配置做到：

acks=all 只有參與復(fù)制的所有節(jié)點全部收到消息，才返回生產(chǎn)者成功。這樣的話除非所有的節(jié)點都掛了，消息才會丟失。 replication.factor=N,設(shè)置大于1的數(shù)，這會要求每個partion至少有2個副本 min.insync.replicas=N，設(shè)置大于1的數(shù)，這會要求leader至少感知到一個follower還保持著連接 retries=N，設(shè)置一個非常大的值，讓生產(chǎn)者發(fā)送失敗一直重試

雖然我們可以通過配置的方式來達(dá)到MQ本身高可用的目的，但是都對性能有損耗，怎樣配置需要根據(jù)業(yè)務(wù)做出權(quán)衡。

消費者丟失

消費者丟失消息的場景：消費者剛收到消息，此時服務(wù)器宕機，MQ認(rèn)為消費者已經(jīng)消費，不會重復(fù)發(fā)送消息，消息丟失。

RocketMQ默認(rèn)是需要消費者回復(fù)ack確認(rèn)，而kafka需要手動開啟配置關(guān)閉自動offset。

消費方不返回ack確認(rèn)，重發(fā)的機制根據(jù)MQ類型的不同發(fā)送時間間隔、次數(shù)都不盡相同，如果重試超過次數(shù)之后會進(jìn)入死信隊列，需要手工來處理了。（Kafka沒有這些）

你說到消費者消費失敗的問題，那么如果一直消費失敗導(dǎo)致消息積壓怎么處理？

因為考慮到時消費者消費一直出錯的問題，那么我們可以從以下幾個角度來考慮：

消費者出錯，肯定是程序或者其他問題導(dǎo)致的，如果容易修復(fù)，先把問題修復(fù)，讓consumer恢復(fù)正常消費

如果時間來不及處理很麻煩，做轉(zhuǎn)發(fā)處理，寫一個臨時的consumer消費方案，先把消息消費，然后再轉(zhuǎn)發(fā)到一個新的topic和MQ資源，這個新的topic的機器資源單獨申請，要能承載住當(dāng)前積壓的消息

處理完積壓數(shù)據(jù)后，修復(fù)consumer，去消費新的MQ和現(xiàn)有的MQ數(shù)據(jù)，新MQ消費完成后恢復(fù)原狀

那如果消息積壓達(dá)到磁盤上限，消息被刪除了怎么辦？

這。。。他媽都刪除了我有啥辦法啊。。。冷靜，再想想。。有了。

最初，我們發(fā)送的消息記錄是落庫保存了的，而轉(zhuǎn)發(fā)發(fā)送的數(shù)據(jù)也保存了，那么我們就可以通過這部分?jǐn)?shù)據(jù)來找到丟失的那部分?jǐn)?shù)據(jù)，再單獨跑個腳本重發(fā)就可以了。如果轉(zhuǎn)發(fā)的程序沒有落庫，那就和消費方的記錄去做對比，只是過程會更艱難一點。

說了這么多，那你說說RocketMQ實現(xiàn)原理吧？

RocketMQ由NameServer注冊中心集群、Producer生產(chǎn)者集群、Consumer消費者集群和若干Broker（RocketMQ進(jìn)程）組成，它的架構(gòu)原理是這樣的：

Broker在啟動的時候去向所有的NameServer注冊，并保持長連接，每30s發(fā)送一次心跳

Producer在發(fā)送消息的時候從NameServer獲取Broker服務(wù)器地址，根據(jù)負(fù)載均衡算法選擇一臺服務(wù)器來發(fā)送消息

Conusmer消費消息的時候同樣從NameServer獲取Broker地址，然后主動拉取消息來消費

為什么RocketMQ不使用Zookeeper作為注冊中心呢？

我認(rèn)為有以下幾個點是不使用zookeeper的原因：

根據(jù)CAP理論，同時最多只能滿足兩個點，而zookeeper滿足的是CP，也就是說zookeeper并不能保證服務(wù)的可用性，zookeeper在進(jìn)行選舉的時候，整個選舉的時間太長，期間整個集群都處于不可用的狀態(tài)，而這對于一個注冊中心來說肯定是不能接受的，作為服務(wù)發(fā)現(xiàn)來說就應(yīng)該是為可用性而設(shè)計。

基于性能的考慮，NameServer本身的實現(xiàn)非常輕量，而且可以通過增加機器的方式水平擴展，增加集群的抗壓能力，而zookeeper的寫是不可擴展的，而zookeeper要解決這個問題只能通過劃分領(lǐng)域，劃分多個zookeeper集群來解決，首先操作起來太復(fù)雜，其次這樣還是又違反了CAP中的A的設(shè)計，導(dǎo)致服務(wù)之間是不連通的。

持久化的機制來帶的問題，ZooKeeper 的 ZAB 協(xié)議對每一個寫請求，會在每個 ZooKeeper 節(jié)點上保持寫一個事務(wù)日志，同時再加上定期的將內(nèi)存數(shù)據(jù)鏡像（Snapshot）到磁盤來保證數(shù)據(jù)的一致性和持久性，而對于一個簡單的服務(wù)發(fā)現(xiàn)的場景來說，這其實沒有太大的必要，這個實現(xiàn)方案太重了。而且本身存儲的數(shù)據(jù)應(yīng)該是高度定制化的。

消息發(fā)送應(yīng)該弱依賴注冊中心，而RocketMQ的設(shè)計理念也正是基于此，生產(chǎn)者在第一次發(fā)送消息的時候從NameServer獲取到Broker地址后緩存到本地，如果NameServer整個集群不可用，短時間內(nèi)對于生產(chǎn)者和消費者并不會產(chǎn)生太大影響。

那Broker是怎么保存數(shù)據(jù)的呢？

RocketMQ主要的存儲文件包括commitlog文件、consumequeue文件、indexfile文件。

Broker在收到消息之后，會把消息保存到commitlog的文件當(dāng)中，而同時在分布式的存儲當(dāng)中，每個broker都會保存一部分topic的數(shù)據(jù)，同時，每個topic對應(yīng)的messagequeue下都會生成consumequeue文件用于保存commitlog的物理位置偏移量offset，indexfile中會保存key和offset的對應(yīng)關(guān)系。

CommitLog文件保存于${Rocket_Home}/store/commitlog目錄中，從圖中我們可以明顯看出來文件名的偏移量，每個文件默認(rèn)1G，寫滿后自動生成一個新的文件。

由于同一個topic的消息并不是連續(xù)的存儲在commitlog中，消費者如果直接從commitlog獲取消息效率非常低，所以通過consumequeue保存commitlog中消息的偏移量的物理地址，這樣消費者在消費的時候先從consumequeue中根據(jù)偏移量定位到具體的commitlog物理文件，然后根據(jù)一定的規(guī)則（offset和文件大小取模）在commitlog中快速定位。

Master和Slave之間是怎么同步數(shù)據(jù)的呢？

而消息在master和slave之間的同步是根據(jù)raft協(xié)議來進(jìn)行的：

在broker收到消息后，會被標(biāo)記為uncommitted狀態(tài)

然后會把消息發(fā)送給所有的slave

slave在收到消息之后返回ack響應(yīng)給master

master在收到超過半數(shù)的ack之后，把消息標(biāo)記為committed

發(fā)送committed消息給所有slave，slave也修改狀態(tài)為committed

你知道RocketMQ為什么速度快嗎？

是因為使用了順序存儲、Page Cache和異步刷盤。

我們在寫入commitlog的時候是順序?qū)懭氲?#xff0c;這樣比隨機寫入的性能就會提高很多

寫入commitlog的時候并不是直接寫入磁盤，而是先寫入操作系統(tǒng)的PageCache

最后由操作系統(tǒng)異步將緩存中的數(shù)據(jù)刷到磁盤

什么是事務(wù)、半事務(wù)消息？怎么實現(xiàn)的？

事務(wù)消息就是MQ提供的類似XA的分布式事務(wù)能力，通過事務(wù)消息可以達(dá)到分布式事務(wù)的最終一致性。

半事務(wù)消息就是MQ收到了生產(chǎn)者的消息，但是沒有收到二次確認(rèn)，不能投遞的消息。

實現(xiàn)原理如下：

生產(chǎn)者先發(fā)送一條半事務(wù)消息到MQ

MQ收到消息后返回ack確認(rèn)

生產(chǎn)者開始執(zhí)行本地事務(wù)

如果事務(wù)執(zhí)行成功發(fā)送commit到MQ，失敗發(fā)送rollback

如果MQ長時間未收到生產(chǎn)者的二次確認(rèn)commit或者rollback，MQ對生產(chǎn)者發(fā)起消息回查

生產(chǎn)者查詢事務(wù)執(zhí)行最終狀態(tài)

根據(jù)查詢事務(wù)狀態(tài)再次提交二次確認(rèn)

最終，如果MQ收到二次確認(rèn)commit，就可以把消息投遞給消費者，反之如果是rollback，消息會保存下來并且在3天后被刪除。

關(guān)注公眾號：科技繆繆，獲取更多精彩技術(shù)文章和電子書

- END -

總結(jié)

以上是生活随笔為你收集整理的active mq topic消费后删除_《我想进大厂》之MQ夺命连环11问的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。