大家好，我是 yes。

最近我一直扎在消息隊(duì)列實(shí)現(xiàn)細(xì)節(jié)之中無(wú)法自拔，已經(jīng)寫了 3 篇Kafka源碼分析，還剩很多沒肝完。之前還存著RocketMQ源碼分析還沒整理。今兒暫時(shí)先跳出來(lái)盤一盤大方向上的消息隊(duì)列有哪些核心注意點(diǎn)。

核心點(diǎn)有很多，為了更貼合實(shí)際場(chǎng)景，我從常見的面試問題入手：

• 如何保證消息不丟失？
• 如何處理重復(fù)消息？
• 如何保證消息的有序性？
• 如何處理消息堆積？

當(dāng)然在剖析這幾個(gè)問題之前需要簡(jiǎn)單的介紹下什么是消息隊(duì)列，消息隊(duì)列常見的一些基本術(shù)語(yǔ)和概念。

接下來(lái)進(jìn)入正文。

什么是消息隊(duì)列

來(lái)看看維基百科怎么說(shuō)的，順帶學(xué)學(xué)英語(yǔ)這波不虧：

In computer science, message queues and mailboxes are software-engineering components typically used for inter-process communication (IPC), or for inter-thread communication within the same process. They use a queue for messaging – the passing of control or of content. Group communication systems provide similar kinds of functionality.

翻譯一下：在計(jì)算機(jī)科學(xué)領(lǐng)域，消息隊(duì)列和郵箱都是軟件工程組件，通常用于進(jìn)程間或同一進(jìn)程內(nèi)的線程通信。它們通過(guò)隊(duì)列來(lái)傳遞消息-傳遞控制信息或內(nèi)容，群組通信系統(tǒng)提供類似的功能。

簡(jiǎn)單的概括下上面的定義：消息隊(duì)列就是一個(gè)使用隊(duì)列來(lái)通信的組件。

上面的定義沒有錯(cuò)，但就現(xiàn)在而言我們?nèi)粘Ｋf(shuō)的消息隊(duì)列常常指代的是消息中間件，它的存在不僅僅只是為了通信這個(gè)問題。

為什么需要消息隊(duì)列

從本質(zhì)上來(lái)說(shuō)是因?yàn)榛ヂ?lián)網(wǎng)的快速發(fā)展，業(yè)務(wù)不斷擴(kuò)張，促使技術(shù)架構(gòu)需要不斷的演進(jìn)。

從以前的單體架構(gòu)到現(xiàn)在的微服務(wù)架構(gòu)，成百上千的服務(wù)之間相互調(diào)用和依賴。從互聯(lián)網(wǎng)初期一個(gè)服務(wù)器上有 100 個(gè)在線用戶已經(jīng)很了不得，到現(xiàn)在坐擁10億日活的微信。我們需要有一個(gè)「東西」來(lái)解耦服務(wù)之間的關(guān)系、控制資源合理合時(shí)的使用以及緩沖流量洪峰等等。

消息隊(duì)列就應(yīng)運(yùn)而生了。它常用來(lái)實(shí)現(xiàn)：異步處理、服務(wù)解耦、流量控制。

異步處理

隨著公司的發(fā)展你可能會(huì)發(fā)現(xiàn)你項(xiàng)目的請(qǐng)求鏈路越來(lái)越長(zhǎng)，例如剛開始的電商項(xiàng)目，可以就是粗暴的扣庫(kù)存、下單。慢慢地又加上積分服務(wù)、短信服務(wù)等。這一路同步調(diào)用下來(lái)客戶可能等急了，這時(shí)候就是消息隊(duì)列登場(chǎng)的好時(shí)機(jī)。

調(diào)用鏈路長(zhǎng)、響應(yīng)就慢了，并且相對(duì)于扣庫(kù)存和下單，積分和短信沒必要這么的 "及時(shí)"。因此只需要在下單結(jié)束那個(gè)流程，扔個(gè)消息到消息隊(duì)列中就可以直接返回響應(yīng)了。而且積分服務(wù)和短信服務(wù)可以并行的消費(fèi)這條消息。

可以看出消息隊(duì)列可以減少請(qǐng)求的等待，還能讓服務(wù)異步并發(fā)處理，提升系統(tǒng)總體性能。

服務(wù)解耦

上面我們說(shuō)到加了積分服務(wù)和短信服務(wù)，這時(shí)候可能又要來(lái)個(gè)營(yíng)銷服務(wù)，之后領(lǐng)導(dǎo)又說(shuō)想做個(gè)大數(shù)據(jù)，又來(lái)個(gè)數(shù)據(jù)分析服務(wù)等等。

可以發(fā)現(xiàn)訂單的下游系統(tǒng)在不斷的擴(kuò)充，為了迎合這些下游系統(tǒng)訂單服務(wù)需要經(jīng)常地修改，任何一個(gè)下游系統(tǒng)接口的變更可能都會(huì)影響到訂單服務(wù)，這訂單服務(wù)組可瘋了，真 ·「核心」項(xiàng)目組。

所以一般會(huì)選用消息隊(duì)列來(lái)解決系統(tǒng)之間耦合的問題，訂單服務(wù)把訂單相關(guān)消息塞到消息隊(duì)列中，下游系統(tǒng)誰(shuí)要誰(shuí)就訂閱這個(gè)主題。這樣訂單服務(wù)就解放啦！

流量控制

想必大家都聽過(guò)「削峰填谷」，后端服務(wù)相對(duì)而言都是比較「弱」的，因?yàn)闃I(yè)務(wù)較重，處理時(shí)間較長(zhǎng)。像一些例如秒殺活動(dòng)爆發(fā)式流量打過(guò)來(lái)可能就頂不住了。因此需要引入一個(gè)中間件來(lái)做緩沖，消息隊(duì)列再適合不過(guò)了。

網(wǎng)關(guān)的請(qǐng)求先放入消息隊(duì)列中，后端服務(wù)盡自己最大能力去消息隊(duì)列中消費(fèi)請(qǐng)求。超時(shí)的請(qǐng)求可以直接返回錯(cuò)誤。

當(dāng)然還有一些服務(wù)特別是某些后臺(tái)任務(wù)，不需要及時(shí)地響應(yīng)，并且業(yè)務(wù)處理復(fù)雜且流程長(zhǎng)，那么過(guò)來(lái)的請(qǐng)求先放入消息隊(duì)列中，后端服務(wù)按照自己的節(jié)奏處理。這也是很 nice 的。

上面兩種情況分別對(duì)應(yīng)著生產(chǎn)者生產(chǎn)過(guò)快和消費(fèi)者消費(fèi)過(guò)慢兩種情況，消息隊(duì)列都能在其中發(fā)揮很好的緩沖效果。

注意

引入消息隊(duì)列固然有以上的好處，但是多引入一個(gè)中間件系統(tǒng)的穩(wěn)定性就下降一層，運(yùn)維的難度抬高一層。因此要權(quán)衡利弊，系統(tǒng)是演進(jìn)的。

消息隊(duì)列基本概念

消息隊(duì)列有兩種模型：隊(duì)列模型和發(fā)布/訂閱模型。

隊(duì)列模型

生產(chǎn)者往某個(gè)隊(duì)列里面發(fā)送消息，一個(gè)隊(duì)列可以存儲(chǔ)多個(gè)生產(chǎn)者的消息，一個(gè)隊(duì)列也可以有多個(gè)消費(fèi)者， 但是消費(fèi)者之間是競(jìng)爭(zhēng)關(guān)系，即每條消息只能被一個(gè)消費(fèi)者消費(fèi)。

發(fā)布/訂閱模型

為了解決一條消息能被多個(gè)消費(fèi)者消費(fèi)的問題，發(fā)布/訂閱模型就來(lái)了。該模型是將消息發(fā)往一個(gè)Topic即主題中，所有訂閱了這個(gè) Topic 的訂閱者都能消費(fèi)這條消息。

其實(shí)可以這么理解，發(fā)布/訂閱模型等于我們都加入了一個(gè)群聊中，我發(fā)一條消息，加入了這個(gè)群聊的人都能收到這條消息。那么隊(duì)列模型就是一對(duì)一聊天，我發(fā)給你的消息，只能在你的聊天窗口彈出，是不可能彈出到別人的聊天窗口中的。

講到這有人說(shuō)，那我一對(duì)一聊天對(duì)每個(gè)人都發(fā)同樣的消息不就也實(shí)現(xiàn)了一條消息被多個(gè)人消費(fèi)了嘛。

是的，通過(guò)多隊(duì)列全量存儲(chǔ)相同的消息，即數(shù)據(jù)的冗余可以實(shí)現(xiàn)一條消息被多個(gè)消費(fèi)者消費(fèi)。RabbitMQ 就是采用隊(duì)列模型，通過(guò) Exchange 模塊來(lái)將消息發(fā)送至多個(gè)隊(duì)列，解決一條消息需要被多個(gè)消費(fèi)者消費(fèi)問題。

這里還能看到假設(shè)群聊里除我之外只有一個(gè)人，那么此時(shí)的發(fā)布/訂閱模型和隊(duì)列模型其實(shí)就一樣了。

小結(jié)一下

隊(duì)列模型每條消息只能被一個(gè)消費(fèi)者消費(fèi)，而發(fā)布/訂閱模型就是為讓一條消息可以被多個(gè)消費(fèi)者消費(fèi)而生的，當(dāng)然隊(duì)列模型也可以通過(guò)消息全量存儲(chǔ)至多個(gè)隊(duì)列來(lái)解決一條消息被多個(gè)消費(fèi)者消費(fèi)問題，但是會(huì)有數(shù)據(jù)的冗余。

發(fā)布/訂閱模型兼容隊(duì)列模型，即只有一個(gè)消費(fèi)者的情況下和隊(duì)列模型基本一致。

RabbitMQ 采用隊(duì)列模型，RocketMQ和Kafka 采用發(fā)布/訂閱模型。

接下來(lái)的內(nèi)容都基于發(fā)布/訂閱模型。

常用術(shù)語(yǔ)

一般我們稱發(fā)送消息方為生產(chǎn)者 Producer，接受消費(fèi)消息方為消費(fèi)者Consumer，消息隊(duì)列服務(wù)端為Broker。

消息從Producer發(fā)往Broker，Broker將消息存儲(chǔ)至本地，然后Consumer從Broker拉取消息，或者Broker推送消息至Consumer，最后消費(fèi)。

為了提高并發(fā)度，往往發(fā)布/訂閱模型還會(huì)引入隊(duì)列或者分區(qū)的概念。即消息是發(fā)往一個(gè)主題下的某個(gè)隊(duì)列或者某個(gè)分區(qū)中。RocketMQ中叫隊(duì)列，Kafka叫分區(qū)，本質(zhì)一樣。

例如某個(gè)主題下有 5 個(gè)隊(duì)列，那么這個(gè)主題的并發(fā)度就提高為 5 ，同時(shí)可以有 5 個(gè)消費(fèi)者并行消費(fèi)該主題的消息。一般可以采用輪詢或者 key hash 取余等策略來(lái)將同一個(gè)主題的消息分配到不同的隊(duì)列中。

與之對(duì)應(yīng)的消費(fèi)者一般都有組的概念 Consumer Group, 即消費(fèi)者都是屬于某個(gè)消費(fèi)組的。一條消息會(huì)發(fā)往多個(gè)訂閱了這個(gè)主題的消費(fèi)組。

假設(shè)現(xiàn)在有兩個(gè)消費(fèi)組分別是Group 1 和 Group 2，它們都訂閱了Topic-a。此時(shí)有一條消息發(fā)往Topic-a，那么這兩個(gè)消費(fèi)組都能接收到這條消息。

然后這條消息實(shí)際是寫入Topic某個(gè)隊(duì)列中，消費(fèi)組中的某個(gè)消費(fèi)者對(duì)應(yīng)消費(fèi)一個(gè)隊(duì)列的消息。

在物理上除了副本拷貝之外，一條消息在Broker中只會(huì)有一份，每個(gè)消費(fèi)組會(huì)有自己的offset即消費(fèi)點(diǎn)位來(lái)標(biāo)識(shí)消費(fèi)到的位置。在消費(fèi)點(diǎn)位之前的消息表明已經(jīng)消費(fèi)過(guò)了。當(dāng)然這個(gè)offset是隊(duì)列級(jí)別的。每個(gè)消費(fèi)組都會(huì)維護(hù)訂閱的Topic下的每個(gè)隊(duì)列的offset。

來(lái)個(gè)圖看看應(yīng)該就很清晰了。

基本上熟悉了消息隊(duì)列常見的術(shù)語(yǔ)和一些概念之后，咱們?cè)賮?lái)看看消息隊(duì)列常見的核心面試點(diǎn)。

如何保證消息不丟失

就我們市面上常見的消息隊(duì)列而言，只要配置得當(dāng)，我們的消息就不會(huì)丟。

先來(lái)看看這個(gè)圖，

可以看到一共有三個(gè)階段，分別是生產(chǎn)消息、存儲(chǔ)消息和消費(fèi)消息。我們從這三個(gè)階段分別入手來(lái)看看如何確保消息不會(huì)丟失。

生產(chǎn)消息

生產(chǎn)者發(fā)送消息至Broker，需要處理Broker的響應(yīng)，不論是同步還是異步發(fā)送消息，同步和異步回調(diào)都需要做好try-catch，妥善的處理響應(yīng)，如果Broker返回寫入失敗等錯(cuò)誤消息，需要重試發(fā)送。當(dāng)多次發(fā)送失敗需要作報(bào)警，日志記錄等。

這樣就能保證在生產(chǎn)消息階段消息不會(huì)丟失。

存儲(chǔ)消息

存儲(chǔ)消息階段需要在消息刷盤之后再給生產(chǎn)者響應(yīng)，假設(shè)消息寫入緩存中就返回響應(yīng)，那么機(jī)器突然斷電這消息就沒了，而生產(chǎn)者以為已經(jīng)發(fā)送成功了。

如果Broker是集群部署，有多副本機(jī)制，即消息不僅僅要寫入當(dāng)前Broker,還需要寫入副本機(jī)中。那配置成至少寫入兩臺(tái)機(jī)子后再給生產(chǎn)者響應(yīng)。這樣基本上就能保證存儲(chǔ)的可靠了。一臺(tái)掛了還有一臺(tái)還在呢（假如怕兩臺(tái)都掛了..那就再多些）。

那假如來(lái)個(gè)地震機(jī)房機(jī)子都掛了呢？emmmmmm...大公司基本上都有異地多活。

那要是這幾個(gè)地都地震了呢？emmmmmm...這時(shí)候還是先關(guān)心關(guān)心人吧。

消費(fèi)消息

這里經(jīng)常會(huì)有同學(xué)犯錯(cuò)，有些同學(xué)當(dāng)消費(fèi)者拿到消息之后直接存入內(nèi)存隊(duì)列中就直接返回給Broker消費(fèi)成功，這是不對(duì)的。

你需要考慮拿到消息放在內(nèi)存之后消費(fèi)者就宕機(jī)了怎么辦。所以我們應(yīng)該在消費(fèi)者真正執(zhí)行完業(yè)務(wù)邏輯之后，再發(fā)送給Broker消費(fèi)成功，這才是真正的消費(fèi)了。

所以只要我們?cè)谙I(yè)務(wù)邏輯處理完成之后再給Broker響應(yīng)，那么消費(fèi)階段消息就不會(huì)丟失。

小結(jié)一下

可以看出，保證消息的可靠性需要三方配合。

生產(chǎn)者需要處理好Broker的響應(yīng)，出錯(cuò)情況下利用重試、報(bào)警等手段。

Broker需要控制響應(yīng)的時(shí)機(jī)，單機(jī)情況下是消息刷盤后返回響應(yīng)，集群多副本情況下，即發(fā)送至兩個(gè)副本及以上的情況下再返回響應(yīng)。

消費(fèi)者需要在執(zhí)行完真正的業(yè)務(wù)邏輯之后再返回響應(yīng)給Broker。

但是要注意消息可靠性增強(qiáng)了，性能就下降了，等待消息刷盤、多副本同步后返回都會(huì)影響性能。因此還是看業(yè)務(wù)，例如日志的傳輸可能丟那么一兩條關(guān)系不大，因此沒必要等消息刷盤再響應(yīng)。

如果處理重復(fù)消息

我們先來(lái)看看能不能避免消息的重復(fù)。

假設(shè)我們發(fā)送消息，就管發(fā)，不管Broker的響應(yīng)，那么我們發(fā)往Broker是不會(huì)重復(fù)的。

但是一般情況我們是不允許這樣的，這樣消息就完全不可靠了，我們的基本需求是消息至少得發(fā)到Broker上，那就得等Broker的響應(yīng)，那么就可能存在Broker已經(jīng)寫入了，當(dāng)時(shí)響應(yīng)由于網(wǎng)絡(luò)原因生產(chǎn)者沒有收到，然后生產(chǎn)者又重發(fā)了一次，此時(shí)消息就重復(fù)了。

再看消費(fèi)者消費(fèi)的時(shí)候，假設(shè)我們消費(fèi)者拿到消息消費(fèi)了，業(yè)務(wù)邏輯已經(jīng)走完了，事務(wù)提交了，此時(shí)需要更新Consumer offset了，然后這個(gè)消費(fèi)者掛了，另一個(gè)消費(fèi)者頂上，此時(shí)Consumer offset還沒更新，于是又拿到剛才那條消息，業(yè)務(wù)又被執(zhí)行了一遍。于是消息又重復(fù)了。

可以看到正常業(yè)務(wù)而言消息重復(fù)是不可避免的，因此我們只能從另一個(gè)角度來(lái)解決重復(fù)消息的問題。

關(guān)鍵點(diǎn)就是冪等。既然我們不能防止重復(fù)消息的產(chǎn)生，那么我們只能在業(yè)務(wù)上處理重復(fù)消息所帶來(lái)的影響。

冪等處理重復(fù)消息

冪等是數(shù)學(xué)上的概念，我們就理解為同樣的參數(shù)多次調(diào)用同一個(gè)接口和調(diào)用一次產(chǎn)生的結(jié)果是一致的。

例如這條 SQLupdate t1 set money = 150 where id = 1 and money = 100; 執(zhí)行多少遍money都是150，這就叫冪等。

因此需要改造業(yè)務(wù)處理邏輯，使得在重復(fù)消息的情況下也不會(huì)影響最終的結(jié)果。

可以通過(guò)上面我那條 SQL 一樣，做了個(gè)前置條件判斷，即money = 100情況，并且直接修改，更通用的是做個(gè)version即版本號(hào)控制，對(duì)比消息中的版本號(hào)和數(shù)據(jù)庫(kù)中的版本號(hào)。

或者通過(guò)數(shù)據(jù)庫(kù)的約束例如唯一鍵，例如insert into update on duplicate key...。

或者記錄關(guān)鍵的key，比如處理訂單這種，記錄訂單ID，假如有重復(fù)的消息過(guò)來(lái)，先判斷下這個(gè)ID是否已經(jīng)被處理過(guò)了，如果沒處理再進(jìn)行下一步。當(dāng)然也可以用全局唯一ID等等。

基本上就這么幾個(gè)套路，真正應(yīng)用到實(shí)際中還是得看具體業(yè)務(wù)細(xì)節(jié)。

如何保證消息的有序性

有序性分：全局有序和部分有序。

全局有序

如果要保證消息的全局有序，首先只能由一個(gè)生產(chǎn)者往Topic發(fā)送消息，并且一個(gè)Topic內(nèi)部只能有一個(gè)隊(duì)列（分區(qū)）。消費(fèi)者也必須是單線程消費(fèi)這個(gè)隊(duì)列。這樣的消息就是全局有序的！

不過(guò)一般情況下我們都不需要全局有序，即使是同步MySQL Binlog也只需要保證單表消息有序即可。

部分有序

因此絕大部分的有序需求是部分有序，部分有序我們就可以將Topic內(nèi)部劃分成我們需要的隊(duì)列數(shù)，把消息通過(guò)特定的策略發(fā)往固定的隊(duì)列中，然后每個(gè)隊(duì)列對(duì)應(yīng)一個(gè)單線程處理的消費(fèi)者。這樣即完成了部分有序的需求，又可以通過(guò)隊(duì)列數(shù)量的并發(fā)來(lái)提高消息處理效率。

圖中我畫了多個(gè)生產(chǎn)者，一個(gè)生產(chǎn)者也可以，只要同類消息發(fā)往指定的隊(duì)列即可。

如果處理消息堆積

消息的堆積往往是因?yàn)?strong>生產(chǎn)者的生產(chǎn)速度與消費(fèi)者的消費(fèi)速度不匹配。有可能是因?yàn)橄⑾M(fèi)失敗反復(fù)重試造成的，也有可能就是消費(fèi)者消費(fèi)能力弱，漸漸地消息就積壓了。

因此我們需要先定位消費(fèi)慢的原因，如果是bug則處理 bug ，如果是因?yàn)楸旧硐M(fèi)能力較弱，我們可以優(yōu)化下消費(fèi)邏輯，比如之前是一條一條消息消費(fèi)處理的，這次我們批量處理，比如數(shù)據(jù)庫(kù)的插入，一條一條插和批量插效率是不一樣的。

假如邏輯我們已經(jīng)都優(yōu)化了，但還是慢，那就得考慮水平擴(kuò)容了，增加Topic的隊(duì)列數(shù)和消費(fèi)者數(shù)量，注意隊(duì)列數(shù)一定要增加，不然新增加的消費(fèi)者是沒東西消費(fèi)的。一個(gè)Topic中，一個(gè)隊(duì)列只會(huì)分配給一個(gè)消費(fèi)者。

當(dāng)然你消費(fèi)者內(nèi)部是單線程還是多線程消費(fèi)那看具體場(chǎng)景。不過(guò)要注意上面提高的消息丟失的問題，如果你是將接受到的消息寫入內(nèi)存隊(duì)列之后，然后就返回響應(yīng)給Broker，然后多線程向內(nèi)存隊(duì)列消費(fèi)消息，假設(shè)此時(shí)消費(fèi)者宕機(jī)了，內(nèi)存隊(duì)列里面還未消費(fèi)的消息也就丟了。

最后

上面的幾個(gè)問題都是我們?cè)谑褂孟㈥?duì)列的時(shí)候經(jīng)常能遇到的問題，并且也是面試關(guān)于消息隊(duì)列方面的核心考點(diǎn)。今天沒有深入具體消息隊(duì)列的細(xì)節(jié)，但是套路就是這么個(gè)套路，大方向上搞明白很關(guān)鍵。之后再接著寫有關(guān)Kafka的源碼分析文章，有興趣的小伙伴請(qǐng)耐心等待。

往期推薦：

圖解+代碼|常見限流算法以及限流在單機(jī)分布式場(chǎng)景下的思考

Kafka請(qǐng)求處理全流程分析

Kafka索引設(shè)計(jì)的亮點(diǎn)：https://juejin.im/post/5efdeae7f265da22d017e58d

Kafka日志段讀寫分析：https://juejin.im/post/5ef6b94ae51d4534a1236cb0

我是 yes，從一點(diǎn)點(diǎn)到億點(diǎn)點(diǎn)，我們下篇見。