[導(dǎo)讀]大家好，我是 yes。最近我一直扎在消息隊列實現(xiàn)細節(jié)之中無法自拔，已經(jīng)寫了 3 篇Kafka源碼分析，還剩很多沒肝完。之前還存著RocketMQ源碼分析還沒整理。今兒暫時先跳出來盤一盤大方向上的消息隊列有哪些核心注意點。核心點有很多，為了更貼合實際場景，我從

大家好，我是 yes。

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

核心點有很多，為了更貼合實際場景，我從常見的面試問題入手：

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

當然在剖析這幾個問題之前需要簡單的介紹下什么是消息隊列，消息隊列常見的一些基本術(shù)語和概念。

接下來進入正文。

什么是消息隊列

來看看維基百科怎么說的，順帶學(xué)學(xué)英語這波不虧：

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.

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

簡單的概括下上面的定義：消息隊列就是一個使用隊列來通信的組件。

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

為什么需要消息隊列

從本質(zhì)上來說是因為互聯(lián)網(wǎng)的快速發(fā)展，業(yè)務(wù)不斷擴張，促使技術(shù)架構(gòu)需要不斷的演進。

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

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

異步處理

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

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

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

服務(wù)解耦

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

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

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

流量控制

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

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

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

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

注意

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

消息隊列基本概念

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

隊列模型

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

發(fā)布/訂閱模型

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

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

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

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

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

小結(jié)一下

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

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

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

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

常用術(shù)語

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

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

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

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

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

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

然后這條消息實際是寫入Topic某個隊列中，消費組中的某個消費者對應(yīng)消費一個隊列的消息。

在物理上除了副本拷貝之外，一條消息在Broker中只會有一份，每個消費組會有自己的offset即消費點位來標識消費到的位置。在消費點位之前的消息表明已經(jīng)消費過了。當然這個offset是隊列級別的。每個消費組都會維護訂閱的Topic下的每個隊列的offset。

來個圖看看應(yīng)該就很清晰了。

基本上熟悉了消息隊列常見的術(shù)語和一些概念之后，咱們再來看看消息隊列常見的核心面試點。

如何保證消息不丟失

就我們市面上常見的消息隊列而言，只要配置得當，我們的消息就不會丟。

先來看看這個圖，

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

生產(chǎn)消息

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

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

存儲消息

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

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

那假如來個地震機房機子都掛了呢？emmmmmm...大公司基本上都有異地多活。

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

消費消息

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

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

所以只要我們在消息業(yè)務(wù)邏輯處理完成之后再給Broker響應(yīng)，那么消費階段消息就不會丟失。

小結(jié)一下

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

如何保證消息的有序性

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

全局有序

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

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

部分有序

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

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

如果處理消息堆積

消息的堆積往往是因為生產(chǎn)者的生產(chǎn)速度與消費者的消費速度不匹配。有可能是因為消息消費失敗反復(fù)重試造成的，也有可能就是消費者消費能力弱，漸漸地消息就積壓了。

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

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

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

最后

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