不知道你有沒(méi)有過(guò)這種經(jīng)歷某天線上服務(wù)突然變慢查日志發(fā)現(xiàn)數(shù)據(jù)庫(kù)連接被一堆抄送任務(wù)占滿又或者兩個(gè)系統(tǒng)之間的數(shù)據(jù)同步一遇到對(duì)方重啟就整段丟失。這類(lèi)問(wèn)題在微服務(wù)拆分之后幾乎必然出現(xiàn)而解決方案里出現(xiàn)頻率最高、落地成本又比較溫和的就是RabbitMQ。RabbitMQ可以說(shuō)是消息中間件里的萬(wàn)金油了。它基于AMQP協(xié)議核心思路是生產(chǎn)者不直接把消息丟給消費(fèi)者而是通過(guò)中間層按規(guī)則路由、暫存、轉(zhuǎn)發(fā)。這種架構(gòu)天然地把誰(shuí)產(chǎn)生數(shù)據(jù)和誰(shuí)消費(fèi)數(shù)據(jù)解耦開(kāi)。這篇筆記是我自己整理的基礎(chǔ)階段超詳細(xì)版本適合剛接觸消息隊(duì)列的同學(xué)也適合那些已經(jīng)在項(xiàng)目里用了RabbitMQ但一直沒(méi)搞懂Exchange和Queue之間關(guān)系的開(kāi)發(fā)者。我會(huì)把概念拆開(kāi)、把路由規(guī)則講透、把可靠性的坑填上最后再給出一套可以直接上手的Spring Boot接入方案。1. 先用一個(gè)真實(shí)場(chǎng)景回答很多中間件都能收發(fā)消息為什么選 RabbitMQ1.1 消息中間件到底解決了什么問(wèn)題先說(shuō)一個(gè)最簡(jiǎn)單的場(chǎng)景。訂單服務(wù)創(chuàng)建訂單成功后需要通知積分服務(wù)加積分、通知短信服務(wù)發(fā)短信、通知搜索服務(wù)更新索引。同步調(diào)用的寫(xiě)法很簡(jiǎn)單但每次新增一個(gè)下游就得多寫(xiě)一段代碼而且只要其中一個(gè)下游慢整個(gè)創(chuàng)建訂單接口就被拖住。后端系統(tǒng)用同步處理的壞處還在于下游短暫不可用時(shí)訂單主流程也會(huì)失敗。消息隊(duì)列的引入方式很直接訂單服務(wù)只負(fù)責(zé)往隊(duì)列里扔一條事件消息然后立刻返回成功。其他服務(wù)根據(jù)自己的訂閱去消費(fèi)互不影響。這就是削峰填谷、異步解耦、流量緩沖三個(gè)核心價(jià)值落地的樣子。1.2 路由模式差異為什么 RabbitMQ 的精確路由更舒服同樣是隊(duì)列不同中間件的定位差別挺大的。Kafka 擅長(zhǎng)海量日志的追加式讀寫(xiě)分區(qū)內(nèi)消息有序消費(fèi)者用拉模式順序消費(fèi)適合做大數(shù)據(jù)管道。而 RabbitMQ 更強(qiáng)調(diào)靈活的路由同一份數(shù)據(jù)可以根據(jù)規(guī)則分發(fā)給不同的隊(duì)列每個(gè)消費(fèi)者各取所需。舉個(gè)具體例子。一條訂單消息下游可能有這些需求所有系統(tǒng)都想知道訂單創(chuàng)建了這件事所以需要廣播物流服務(wù)只想消費(fèi)已發(fā)貨事件風(fēng)控服務(wù)只關(guān)心金額超過(guò) 5000 的訂單。如果在 Kafka 里搞通常要建多個(gè) topic 或者讓消費(fèi)者自己過(guò)濾。但在 RabbitMQ 里這屬于消息路由的基本操作通過(guò) Exchange 類(lèi)型和 Routing Key 的組合規(guī)則就能精確控制。這個(gè)設(shè)計(jì)讓 RabbitMQ 在業(yè)務(wù)系統(tǒng)內(nèi)部的系統(tǒng)間集成任務(wù)上特別順手所以我說(shuō)它是萬(wàn)金油。1.3 什么時(shí)候不應(yīng)該選 RabbitMQRabbitMQ 也不是完美的。單隊(duì)列的消息堆積能力遠(yuǎn)不如 Kafka如果某天突然涌進(jìn)幾千萬(wàn)條消息積壓RabbitMQ 的吞吐量會(huì)明顯下滑內(nèi)存和磁盤(pán)壓力都很大。在需要千萬(wàn)級(jí)吞吐、海量日志采集、消息長(zhǎng)時(shí)間積壓回放的場(chǎng)景Kafka 或同類(lèi)產(chǎn)品更合適。還有一個(gè)容易忽略的點(diǎn)RabbitMQ 的消息順序問(wèn)題。它在單個(gè)隊(duì)列、單消費(fèi)者的情況下能保證順序但一旦多個(gè)消費(fèi)者并發(fā)消費(fèi)同一個(gè)隊(duì)列順序就不可能保證了。對(duì)于強(qiáng)順序依賴的業(yè)務(wù)要多考慮一個(gè)排序字段 落庫(kù)后重排的補(bǔ)償方案不能把寶全押在中間件上。2. 核心概念一次吃透Producer、Channel、Exchange、Queue、Binding 之間的協(xié)作關(guān)系2.1 別把 Queue 當(dāng)成核心Exchange 才是理解 RabbitMQ 的鑰匙很多初學(xué)者一上來(lái)就以為消息隊(duì)列的核心是隊(duì)列其實(shí)在 RabbitMQ 里消息從生產(chǎn)者發(fā)出后并不會(huì)直接進(jìn)入隊(duì)列而是先到達(dá)一個(gè) Exchange交換機(jī)。Exchange 根據(jù)消息攜帶的路由鍵和自身的類(lèi)型決定這條消息應(yīng)該投遞到哪個(gè)隊(duì)列、哪幾個(gè)隊(duì)列或者直接丟棄。這個(gè)概念很像快遞分揀中心。你寄件時(shí)不會(huì)把包裹直接塞進(jìn)某個(gè)貨運(yùn)車(chē)而是交給分揀中心分揀中心根據(jù)面單上的地址把包裹分配到不同的運(yùn)輸線。Exchange 就是分揀中心Queue 就是運(yùn)輸線Routing Key 就是面單地址。Exchange 和 Queue 之間需要建立 Binding綁定關(guān)系。一條綁定的核心內(nèi)容是Exchange 名稱 Queue 名稱 Binding Key。消息的 Routing Key 只有和某個(gè)綁定的 Binding Key 匹配上才會(huì)被路由進(jìn)對(duì)應(yīng)的隊(duì)列。2.2 Channel 和 vhost多路復(fù)用與邏輯隔離先說(shuō) Connection 和 Channel 的關(guān)系。一個(gè) Connection 就是一條 TCP 連接但它比較重不能每個(gè)線程單獨(dú)建一條。所以要在一個(gè) Connection 上創(chuàng)建多個(gè) Channel信道每個(gè) Channel 有獨(dú)立的 channel number可以理解為 TCP 連接里的虛擬通道。生產(chǎn)者和消費(fèi)者通常共用一個(gè) Connection但每個(gè)線程持有獨(dú)立的 Channel。官方也明確說(shuō) Channel 不是線程安全的不要在多線程里共享同一個(gè) Channel這是大家初期比較容易犯的錯(cuò)誤。再說(shuō) vhost虛擬主機(jī)。它是權(quán)限隔離和邏輯隔離的單位一個(gè) RabbitMQ 服務(wù)上可以建多個(gè) vhost每個(gè) vhost 擁有自己獨(dú)立的 Exchange、Queue、Binding互相看不到對(duì)方。實(shí)際項(xiàng)目里可以一個(gè)環(huán)境一個(gè) vhost比如 dev、test、prod或者一個(gè)事業(yè)部一個(gè) vhost。網(wǎng)上很多帖子建議統(tǒng)一使用默認(rèn) vhost /我的習(xí)慣是永遠(yuǎn)給項(xiàng)目單獨(dú)建 vhost不只是為了安全更是為了避免不同項(xiàng)目隊(duì)列名撞車(chē)后互相干擾。2.3 一條消息在服務(wù)端長(zhǎng)什么樣消息本身除了 body業(yè)務(wù)數(shù)據(jù)之外還帶著一堆屬性Delivery Mode持久化標(biāo)志、Headers頭信息表、Timestamp、Message ID、Reply To 等等。理解這些屬性的最好方式就是打開(kāi)管理臺(tái)的隊(duì)列頁(yè)面點(diǎn)進(jìn)去看一條消息的 Properties 信息比看文檔記得牢。有一點(diǎn)容易被忽略消息的 Content-Type 和編碼問(wèn)題。生產(chǎn)端寫(xiě)入 JSON 字符串時(shí)如果用 UTF-8消費(fèi)端解析時(shí)也必須按 UTF-8 處理否則中文會(huì)有亂碼排查起來(lái)非常繞。建議在 Producer 設(shè)置消息屬性時(shí)顯式帶上contentType: application/json和contentEncoding: UTF-8。3. 從安裝到管理臺(tái)手把手搭一套本地環(huán)境順便把坑填平3.1 最快的方式是用容器跑起來(lái)本地開(kāi)發(fā)我基本只用 Docker 方式一條命令就能起服務(wù)端不用操心 Erlang 版本依賴問(wèn)題。注意 RabbitMQ 的鏡像需要包含管理插件對(duì)應(yīng) tag 是management比如rabbitmq:3.13-management。只拉默認(rèn)版鏡像的話進(jìn)去之后是沒(méi)有管理界面的要自己 enable 插件很多人第一步就卡在這里。docker run -d \ --name rabbitmq-dev \ -p 5672:5672 \ -p 15672:15672 \ -v rabbitmq-data:/var/lib/rabbitmq \ rabbitmq:3.13-management5672 是 AMQP 協(xié)議端口15672 是管理界面端口默認(rèn)賬號(hào)密碼都是guest。數(shù)據(jù)目錄掛載到宿主機(jī)卷重啟容器鏡像更新時(shí)數(shù)據(jù)不會(huì)丟。有個(gè)細(xì)節(jié)值得說(shuō)明guest賬號(hào)默認(rèn)只允許 localhost 訪問(wèn)如果你是在遠(yuǎn)程服務(wù)器上部署直接用 guest 登錄管理界面會(huì)被拒絕。解決辦法是創(chuàng)建新用戶并設(shè)置權(quán)限這是一定會(huì)遇到的問(wèn)題。3.2 管理界面里每個(gè)區(qū)域?qū)?yīng)哪個(gè)概念登錄管理界面后會(huì)看到 Overview總覽、Connections、Channels、Exchanges、Queues、Admin 等模塊。建議養(yǎng)成習(xí)慣在 Exchanges 頁(yè)面點(diǎn)擊某個(gè)交換機(jī)往下拉能看到 Binding 列表然后跳轉(zhuǎn)到對(duì)應(yīng) Queue再把Binding 和 Routing Key 的對(duì)應(yīng)關(guān)系和代碼里的聲明語(yǔ)句逐行對(duì)照一遍很多路由問(wèn)題當(dāng)場(chǎng)就能發(fā)現(xiàn)是哪里配置錯(cuò)位了。每個(gè) Queue 頁(yè)面還有 Get Messages 功能可以手動(dòng)拉取消息預(yù)覽。注意這個(gè)操作只是把消息抓出來(lái)看默認(rèn)不會(huì)重新入隊(duì)選Requeue會(huì)放回隊(duì)列選Reject會(huì)丟棄或進(jìn)入死信手滑點(diǎn)錯(cuò)可能丟消息。在測(cè)試環(huán)境無(wú)所謂生產(chǎn)環(huán)境不要隨意點(diǎn)這個(gè)按鈕要預(yù)覽消息可以用 UI 的 Quorum Queue 消息瀏覽器普通隊(duì)列沒(méi)有這個(gè)能力。3.3 必會(huì)的 rabbitmqctl 命令比起圖形界面我更喜歡命令行因?yàn)樗桥佩e(cuò)時(shí)最可靠的入口。幾條高頻命令# 查看服務(wù)與節(jié)點(diǎn)狀態(tài) rabbitmqctl status # 查看用戶列表 rabbitmqctl list_users # 新建用戶并授權(quán)實(shí)際項(xiàng)目里的標(biāo)準(zhǔn)操作 rabbitmqctl add_user dev_user dev_password rabbitmqctl set_permissions -p / dev_user .* .* .* # 查看所有隊(duì)列及堆積數(shù)量、消費(fèi)者數(shù) rabbitmqctl list_queues name messages consumers # 查看交換機(jī)對(duì)應(yīng)綁定關(guān)系 rabbitmqctl list_bindingslist_bindings這個(gè)命令的信息量很大它會(huì)把所有 Exchange、Queue、Binding Key 一次性輸出。當(dāng)路由詭異失效的時(shí)候我用它比對(duì)代碼聲明和實(shí)際綁定通常十分鐘內(nèi)就能定位問(wèn)題。4. 消息路由親手驗(yàn)證四種 Exchange 類(lèi)型的選擇邏輯4.1 Fanout廣播路由配置最簡(jiǎn)單Fanout 交換機(jī)不看 Routing Key只要消息進(jìn)入交換機(jī)就復(fù)制到所有綁定它的隊(duì)列。典型的業(yè)務(wù)場(chǎng)景是廣播通知用戶狀態(tài)變更后所有關(guān)心此事件的系統(tǒng)都收到通知。聲明方式如下我用 Java 客戶端示例其它語(yǔ)言思路一致Channel channel connection.createChannel(); channel.exchangeDeclare(event.user.change, BuiltinExchangeType.FANOUT, true); String queueA queue.user.sms; String queueB queue.user.search; // 交換機(jī)與隊(duì)列建立綁定Binding Key 傳空即可 channel.queueDeclare(queueA, true, false, false, null); channel.queueDeclare(queueB, true, false, false, null); channel.queueBind(queueA, event.user.change, ); channel.queueBind(queueB, event.user.change, ); channel.basicPublish(event.user.change, , null, messageBody.getBytes());有一點(diǎn)要注意如果消費(fèi)者不在線消息會(huì)一直存在隊(duì)列里等消費(fèi)者恢復(fù)后再消費(fèi)。這算是好消息也是壞消息——好消息是不丟壞消息是隊(duì)列堆積會(huì)持續(xù)占內(nèi)存。廣播場(chǎng)景尤其要關(guān)注消費(fèi)者處理速度和隊(duì)列堆積水位。4.2 Direct精確匹配的點(diǎn)對(duì)點(diǎn)業(yè)務(wù)Direct 交換機(jī)的路由規(guī)則非常簡(jiǎn)單消息的 Routing Key 和 Binding Key 完全相等才算匹配。處理訂單這種有點(diǎn)對(duì)點(diǎn)性質(zhì)的業(yè)務(wù)時(shí)最常用。channel.exchangeDeclare(exchange.order, BuiltinExchangeType.DIRECT, true); channel.queueDeclare(queue.order.pay, true, false, false, null); channel.queueBind(queue.order.pay, exchange.order, order.pay.success); // 發(fā)消息時(shí)帶上精確路由鍵 channel.basicPublish(exchange.order, order.pay.success, null, payload.getBytes());如果消息的 Routing Key 沒(méi)有匹配到任何綁定Direct 交換機(jī)會(huì)直接把消息丟掉。生產(chǎn)環(huán)境最怕這種靜默丟失所以我建議在發(fā)布端接上確認(rèn)機(jī)制后面可靠性部分細(xì)說(shuō)發(fā)布返回值能明確告訴你這條消息是否被路由到了隊(duì)列。4.3 Topic用通配符做靈活的細(xì)粒度路由Topic 交換機(jī)的匹配規(guī)則是Routing Key 用點(diǎn)號(hào)分割*匹配一個(gè)詞#匹配零個(gè)或多個(gè)詞。它處于 Fanout 和 Direct 中間適合復(fù)雜業(yè)務(wù)規(guī)則的訂閱需求。假設(shè)有股票行情系統(tǒng)# 路由鍵結(jié)構(gòu)market.usd.price queue.usd.all 綁定 market.usd.# queue.price.all 綁定 market.#.price queue.usd.price 綁定 market.usd.price發(fā)一條market.usd.price消息三個(gè)隊(duì)列都能收到。如果只發(fā)market.usd那么#可以匹配零段但*必須匹配一個(gè)精確詞具體表現(xiàn)要看綁定寫(xiě)法。正因?yàn)橐?guī)則比 Direct 復(fù)雜配錯(cuò) Binding Key 是 Topic 模式最常見(jiàn)的線上事故我的經(jīng)驗(yàn)是先寫(xiě)一個(gè)小腳本或直接在管理臺(tái)手動(dòng)發(fā)測(cè)試消息驗(yàn)證路由再上代碼。4.4 Headers什么時(shí)候才值得用Headers 交換機(jī)不看 Routing Key而是根據(jù)消息 Header 里的鍵值對(duì)做匹配配合x(chóng)-match參數(shù)all表示所有指定 Header 都要匹配any表示任意一個(gè)匹配即可。它能實(shí)現(xiàn)按消息屬性路由這種超靈活的需求。實(shí)際項(xiàng)目里用 Headers 的場(chǎng)景真的很少。一是它的路由復(fù)雜度讓調(diào)用方很難直觀理解二是 Header 匹配性能不如前三種三是在管理臺(tái)上排查路由關(guān)系也更麻煩。如果需求只是根據(jù)幾個(gè)條件組合分發(fā)我更建議在生產(chǎn)者里把條件組合成標(biāo)準(zhǔn) Routing Key然后交給 Topic 交換機(jī)處理。除非是歷史遺留代碼已經(jīng)這么寫(xiě)了否則新設(shè)計(jì)我基本不推薦 Headers。5. 如何保證消息不丟從一條消息的生命周期拆解可靠性5.1 消息會(huì)死在哪個(gè)環(huán)節(jié)要討論可靠性先得把一條消息從生產(chǎn)到消費(fèi)經(jīng)歷的路程畫(huà)出來(lái)生產(chǎn)者客戶端發(fā)送消息經(jīng)過(guò)網(wǎng)絡(luò)到 BrokerBroker 寫(xiě)到 Exchange路由到 QueueQueue 落盤(pán)/內(nèi)存存儲(chǔ)消費(fèi)者拉取消息處理業(yè)務(wù)后返回確認(rèn)。丟消息的位置就四個(gè)生產(chǎn)端發(fā)出前網(wǎng)絡(luò)閃斷導(dǎo)致根本沒(méi)送到 Broker生產(chǎn)者發(fā)出后消息被路由到不存在的隊(duì)列且交換機(jī)類(lèi)型不匹配路由鍵消息被丟棄Broker 側(cè)收到消息后沒(méi)做持久化服務(wù)重啟數(shù)據(jù)丟失消費(fèi)者拉取消息后在處理完成之前崩潰服務(wù)端認(rèn)為消息已被消費(fèi)每個(gè)環(huán)節(jié)都有對(duì)應(yīng)的防御手段生產(chǎn)者確認(rèn)、Exchange 與 Queue 持久化、消費(fèi)者手動(dòng) ACK。下面逐項(xiàng)展開(kāi)。5.2 持久化配置三層都要設(shè)缺一層都不保險(xiǎn)消息持久化不是一個(gè)開(kāi)關(guān)而是三條獨(dú)立的持久化設(shè)置同時(shí)生效Exchange 聲明時(shí)設(shè)置durabletrueQueue 聲明時(shí)設(shè)置durabletrue發(fā)布消息時(shí)設(shè)置MessageProperties.PERSISTENT_TEXT_PLAINDelivery Mode 2只把 Queue 設(shè)成持久化但發(fā)布消息時(shí)是默認(rèn)非持久化服務(wù)重啟后 Queue 還在里面的消息還是丟了。只把消息設(shè)成持久化但 Queue 不是持久化那 Queue 本身重啟就沒(méi)了。三個(gè)條件必須同時(shí)滿足消息才能真正扛住 Broker 重啟。這個(gè)細(xì)節(jié)我見(jiàn)很多人踩過(guò)包括當(dāng)年的我自己。還有一個(gè)邊界情況即使三層持久化都做了也只能保證Broker 不丟已確認(rèn)的消息不能保證磁盤(pán)損壞這種極端故障。真要跑到金融級(jí)可靠性得靠后面的鏡像隊(duì)列或 Quorum 隊(duì)列做主從副本。5.3 生產(chǎn)端確認(rèn)與消費(fèi)端確認(rèn)雙向都要簽字生產(chǎn)端確認(rèn)也就是 Publisher Confirm 機(jī)制。開(kāi)啟后Broker 每收到一條消息并成功落盤(pán)后會(huì)回一個(gè) Confirm 應(yīng)答。在 Java 客戶端里開(kāi)啟方式很簡(jiǎn)單channel.confirmSelect(); channel.basicPublish(exchange.order, order.pay.success, null, payload.getBytes()); // 同步等待確認(rèn)會(huì)阻塞直到 Broker 應(yīng)答 if (channel.waitForConfirms()) { // 消息確認(rèn)成功 } else { // 消息確認(rèn)失敗需要記錄并重發(fā) }生產(chǎn)端確認(rèn)能解決消息發(fā)出但沒(méi)送達(dá)到 Broker 的問(wèn)題。一旦沒(méi)有收到確認(rèn)業(yè)務(wù)側(cè)要負(fù)責(zé)重發(fā)。另外這里經(jīng)常有人把 Confirm 和事務(wù)混為一談RabbitMQ 的txSelect()事務(wù)機(jī)制也能做但性能損耗太大一個(gè)事務(wù)要額外幾次 RTT現(xiàn)代項(xiàng)目基本不用它用 Confirm 就好。消費(fèi)端確認(rèn)是在消費(fèi)者接收消息之后處理完業(yè)務(wù)邏輯后顯式basicAck而不是使用自動(dòng)確認(rèn)模式。自動(dòng)確認(rèn)本質(zhì)是服務(wù)端發(fā)消息就算消費(fèi)成功一旦處理過(guò)程拋異常消息就沒(méi)了。改成手動(dòng)確認(rèn)才有重試和死信的機(jī)會(huì)public void handleMessage(String message, Channel channel, long deliveryTag) throws IOException { try { // 業(yè)務(wù)處理 channel.basicAck(deliveryTag, false); } catch (Exception e) { // 第三個(gè)參數(shù)是否重新回到隊(duì)列 channel.basicNack(deliveryTag, false, false); } }basicNack最后一個(gè)參數(shù)值得你認(rèn)真想清楚requeuetrue會(huì)把消息放回原隊(duì)列如果消費(fèi)者總是處理失敗這條消息就會(huì)無(wú)限循環(huán)刷爆日志也影響后續(xù)正常消息。更穩(wěn)妥的做法是requeuefalse讓它進(jìn)死信隊(duì)列后面單獨(dú)處理。5.4 死信隊(duì)列和延遲消息最實(shí)用的進(jìn)階玩法死信隊(duì)列DLX是 RabbitMQ 能力組合的經(jīng)典產(chǎn)物。一個(gè)隊(duì)列可以被指定死信交換機(jī)當(dāng)隊(duì)列里的消息出現(xiàn)以下情況時(shí)消息會(huì)被投遞到死信交換機(jī)消費(fèi)者調(diào)用basicReject/basicNack且requeuefalse消息設(shè)置了存活時(shí)間 TTL到期未被消費(fèi)隊(duì)列已滿無(wú)法容納新消息有了死信隊(duì)列業(yè)務(wù)代碼可以專門(mén)監(jiān)聽(tīng)處理失敗的消息做告警、人工介入或補(bǔ)償。這個(gè)機(jī)制通常也是實(shí)現(xiàn)延遲消息的基礎(chǔ)方案先把消息發(fā)到帶 TTL 的隊(duì)列等到消息過(guò)期后投遞到死信交換機(jī)再由死信交換機(jī)路由到實(shí)際處理隊(duì)列。這樣不需要額外中間件只靠 RabbitMQ 自帶屬性就能實(shí)現(xiàn)定時(shí)延遲效果。實(shí)現(xiàn)要點(diǎn)是 TTL 放在隊(duì)列上比放在單條消息上更可控。每個(gè)隊(duì)列通過(guò)x-message-ttl參數(shù)設(shè)置過(guò)期時(shí)間再綁定x-dead-letter-exchange和x-dead-letter-routing-key。我實(shí)際項(xiàng)目中用這個(gè)方案做過(guò)訂單超時(shí)關(guān)閉延遲 30 分鐘、1 小時(shí)、1 天分別建了不同 TTL 的隊(duì)列雖然隊(duì)列數(shù)量多一些但延遲粒度清楚可控。6. Spring Boot 集成生產(chǎn)級(jí)監(jiān)聽(tīng)的正確姿勢(shì)6.1 起步依賴和最小配置Spring Boot 接入 RabbitMQ 的主要依賴是dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency隨后在配置文件中聲明連接信息spring: rabbitmq: host: 127.0.0.1 port: 5672 username: dev_user password: dev_password virtual-host: / publisher-confirm-type: correlated publisher-returns: truepublisher-confirm-type: correlated和publisher-returns: true必須要開(kāi)。前者讓RabbitTemplate在消息送達(dá) Broker 后觸發(fā)確認(rèn)回調(diào)后者在消息無(wú)法路由時(shí)觸發(fā) ReturnedMessage 回調(diào)這樣生產(chǎn)端才能知道消息到底有沒(méi)有路由成功。默認(rèn)關(guān)閉的話消息發(fā)出去就是聽(tīng)天由命。6.2 RabbitListener 的并發(fā)與手動(dòng)確認(rèn)監(jiān)聽(tīng)器寫(xiě)法很簡(jiǎn)潔Component public class OrderPayConsumer { RabbitListener(queues queue.order.pay, ackMode MANUAL) public void onMessage(String message, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { // 執(zhí)行業(yè)務(wù)邏輯 System.out.println(received: message); channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, false); // 記錄業(yè)務(wù)失敗次數(shù)超過(guò)閾值走死信 } } }ackMode MANUAL對(duì)應(yīng)前面講的手動(dòng)確認(rèn)。Spring Boot 默認(rèn)其實(shí)是自動(dòng)確認(rèn)對(duì)大多數(shù)內(nèi)部接口夠用但一旦業(yè)務(wù)處理中間出現(xiàn)異常消息就靜默丟失了。要達(dá)到我在第 5 節(jié)說(shuō)的可靠性這里必須改成手動(dòng)。并發(fā)參數(shù)也很關(guān)鍵。RabbitListener默認(rèn)單線程消費(fèi)處理慢時(shí)消息容易堆積?？梢栽谂渲妙?lèi)里定制容器工廠Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setPrefetchCount(50); factory.setConcurrentConsumers(5); factory.setMaxConcurrentConsumers(15); return factory; }prefetchCount表示每個(gè)消費(fèi)者從隊(duì)列預(yù)取多少條消息。設(shè)得太大消息全部給到當(dāng)前消費(fèi)者其它消費(fèi)者閑置設(shè)得太小性能上不去。我一般從 50 起步配合concurrentConsumers觀察積壓量再調(diào)整。這個(gè)參數(shù)不是越大越好關(guān)鍵看單個(gè)消息的處理時(shí)長(zhǎng)。6.3 重試和異常處理別讓消息陷入死循環(huán)Spring Boot 默認(rèn)的消息重試策略是defaultRequeueRejectedfalse3.1 之前是 false更早版本行為不同也就是說(shuō)處理失敗的消費(fèi)會(huì)被直接拒絕并放回隊(duì)列。放回隊(duì)列后若依然失敗就無(wú)限循環(huán)。所以我會(huì)在配置里顯式關(guān)掉無(wú)限重試改用有限重試加死信spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 3 initial-interval: 2s multiplier: 2這樣做的問(wèn)題在于Spring 重試只是重新調(diào)用監(jiān)聽(tīng)器方法如果最終還是失敗默認(rèn)行為還是會(huì)把消息確認(rèn)掉等于重試 3 次后放棄。所以生產(chǎn)環(huán)境我通常會(huì)配合RabbitTemplate的 ReturnCallback 和死信隊(duì)列在最后一次失敗后做告警或進(jìn)入死信由人工處理。性能方面有一個(gè)很實(shí)際的坑如果監(jiān)聽(tīng)的隊(duì)列有多個(gè)消費(fèi)者實(shí)例部署在不同節(jié)點(diǎn)消息會(huì)被負(fù)載均衡分發(fā)這時(shí)候冪等設(shè)計(jì)是硬性要求。同一個(gè)訂單可能被重試多次或分給不同實(shí)例消費(fèi)端一定要保證同一條消息重復(fù)執(zhí)行和執(zhí)行一次結(jié)果一致通常用唯一業(yè)務(wù) ID 去重表來(lái)實(shí)現(xiàn)。7. 集群、鏡像隊(duì)列與 Quorum 隊(duì)列高可用方案的兩種思路7.1 普通集群到底解決了什么問(wèn)題RabbitMQ 的默認(rèn)集群模式多個(gè)節(jié)點(diǎn)組成邏輯上的一個(gè) Broker。元數(shù)據(jù)Exchange、Queue、Binding、權(quán)限會(huì)在集群內(nèi)同步但隊(duì)列內(nèi)容和消息數(shù)據(jù)只存在于隊(duì)列所在節(jié)點(diǎn)其它節(jié)點(diǎn)不復(fù)制消息數(shù)據(jù)。也就是說(shuō)一個(gè)節(jié)點(diǎn)宕機(jī)它負(fù)責(zé)的隊(duì)列短暫不可用除非客戶端連到了持有數(shù)據(jù)的那臺(tái)節(jié)點(diǎn)否則無(wú)法消費(fèi)。這種模式的價(jià)值是橫向擴(kuò)展吞吐量不是高可用。三個(gè)節(jié)點(diǎn)分擔(dān)大量隊(duì)列和連接單點(diǎn)壓力變小。但如果你追求節(jié)點(diǎn)掛掉消息不丟、服務(wù)不斷普通集群是不夠的。7.2 鏡像隊(duì)列與腦裂風(fēng)險(xiǎn)鏡像隊(duì)列Mirrored Queue把隊(duì)列的完整數(shù)據(jù)復(fù)制到指定的多個(gè)節(jié)點(diǎn)上。方式是在隊(duì)列聲明時(shí)加x-ha-policy: all隊(duì)列在每個(gè)節(jié)點(diǎn)都會(huì)有一份完整數(shù)據(jù)其中一個(gè)節(jié)點(diǎn)為主節(jié)點(diǎn)其它為鏡像。主節(jié)點(diǎn)宕機(jī)后從鏡像中選一個(gè)提升為主節(jié)點(diǎn)業(yè)務(wù)繼續(xù)。鏡像隊(duì)列的代價(jià)是寫(xiě)放大每條消息要同步到所有鏡像節(jié)點(diǎn)后才算確認(rèn)完成延遲和吞吐都會(huì)有損耗。還有更頭疼的腦裂問(wèn)題網(wǎng)絡(luò)分區(qū)時(shí)節(jié)點(diǎn)之間無(wú)法確認(rèn)對(duì)方狀態(tài)可能各自選出新的主節(jié)點(diǎn)導(dǎo)致隊(duì)列數(shù)據(jù)出現(xiàn)分裂。RabbitMQ 的默認(rèn)處理策略是少數(shù)派停止服務(wù)避免出現(xiàn)腦裂后數(shù)據(jù)不一致但這意味著集群的可用性會(huì)暫時(shí)降低。用鏡像隊(duì)列時(shí)我的建議是集群節(jié)點(diǎn)數(shù)至少 3 個(gè)奇數(shù)個(gè)節(jié)點(diǎn)更方便仲裁不要跨機(jī)房部署同一個(gè)鏡像組跨機(jī)房網(wǎng)絡(luò)延遲會(huì)把這個(gè)系統(tǒng)的性能拖垮網(wǎng)絡(luò)分區(qū)策略要提前想好到底是用autoheal、pause_minority還是pause_if_all_down這個(gè)配置在線上出問(wèn)題時(shí)如果沒(méi)定清楚人會(huì)非?；?。7.3 Quorum 隊(duì)列數(shù)據(jù)安全優(yōu)先的新默認(rèn)選擇RabbitMQ 3.8 之后引入的 Quorum 隊(duì)列基于 Raft 協(xié)議做復(fù)制。每條消息獲得多數(shù)節(jié)點(diǎn)確認(rèn)后才算寫(xiě)入成功同樣支持故障轉(zhuǎn)移。和鏡像隊(duì)列比它在網(wǎng)絡(luò)分區(qū)時(shí)表現(xiàn)更穩(wěn)定遵循 Raft 的選主規(guī)則不容易出現(xiàn)腦裂而且性能損耗更可控?，F(xiàn)在的項(xiàng)目如果是從零開(kāi)始建我會(huì)直接考慮 Quorum 隊(duì)列而不是鏡像隊(duì)列。它還能配合管理界面直接瀏覽消息內(nèi)容排查問(wèn)題比普通隊(duì)列方便很多。如果你接手的是老項(xiàng)目暫時(shí)不需要立刻遷移但新隊(duì)列盡量用 Quorum。唯一要注意的是 Quorum 隊(duì)列的特性跟普通隊(duì)列有差異比如不支持immediate標(biāo)志、消息不一定嚴(yán)格 FIFO但有確定順序這些在遷移時(shí)要提前排查。8. 性能調(diào)優(yōu)與踩坑復(fù)盤(pán)我常用的參數(shù)和排錯(cuò)路徑8.1 幾個(gè)想重點(diǎn)標(biāo)記的參數(shù)下面是我不論項(xiàng)目多小都會(huì)設(shè)置的清單整理成一張表方便對(duì)照配置項(xiàng)推薦起點(diǎn)說(shuō)明prefetchCount50消費(fèi)者預(yù)取數(shù)量處理快的任務(wù)可以加大到 100concurrentConsumers5監(jiān)聽(tīng)器初始并發(fā)數(shù)maxConcurrentConsumers15超過(guò)后不再增加消費(fèi)者receiveTimeout1000隊(duì)列為空時(shí)拉取超時(shí)避免空轉(zhuǎn)channel-cache-size100緩存 Channel 數(shù)量publisher-confirm-typecorrelated生產(chǎn)確認(rèn)必須開(kāi)publisher-returnstrue路由失敗必須通知還有一個(gè)被反復(fù)忽略的問(wèn)題消息體大小。RabbitMQ 對(duì)單條消息沒(méi)有硬性限制但實(shí)際斷言過(guò)大的消息會(huì)拖慢整個(gè)集群。我在項(xiàng)目里碰過(guò)有人把整個(gè) Excel 轉(zhuǎn)成 JSON 塞消息一條幾 MB結(jié)果隊(duì)列吞吐直接趴窩。建議業(yè)務(wù)消息體控制在幾十 KB 以內(nèi)超過(guò)這個(gè)規(guī)模的走文件存儲(chǔ)消息里放文件路徑即可。8.2 消息堆積、消費(fèi)停滯和連接異常三條排查鏈路消息堆積是最常見(jiàn)的告警項(xiàng)。排查思路第一步用rabbitmqctl list_queues name messages consumers看堆積量再對(duì)比消費(fèi)者數(shù)量如果消費(fèi)者數(shù)是 0說(shuō)明消費(fèi)端沒(méi)連上或監(jiān)聽(tīng)器沒(méi)起來(lái)如果消費(fèi)者數(shù)正常但消息不減說(shuō)明消費(fèi)者在處理時(shí)被卡住或處理速度跟不上生產(chǎn)速度。消費(fèi)停滯時(shí)我先看rabbitmqctl list_channels | grep找到對(duì)應(yīng)信道確認(rèn)prefetch_count和unacked數(shù)量。unacked數(shù)量高說(shuō)明消費(fèi)者已經(jīng)拉走了消息但遲遲沒(méi)有 ACK大概率處理邏輯出問(wèn)題或線程阻塞了。這時(shí)候去消費(fèi)者服務(wù)的線程棧通常能一眼看到是被什么調(diào)用卡住了。連接異常時(shí)有個(gè)坑客戶端默認(rèn)的心跳超時(shí)如果網(wǎng)絡(luò)質(zhì)量不好服務(wù)端認(rèn)為客戶端死了會(huì)主動(dòng)關(guān)閉連接。連接頻繁斷開(kāi)-重連又會(huì)產(chǎn)生一批未確認(rèn)消息重復(fù)消費(fèi)。癥狀是消息偶爾重復(fù)、日志滿屏連接重置。解決辦法是調(diào)整心跳間隔比如在連接工廠里把requestedHeartbeat設(shè)置為 30 秒而不是用默認(rèn)的 60 秒或自動(dòng)協(xié)商值。另外客戶端所有連接都應(yīng)該設(shè)置automaticRecoveryEnabledtrue否則一次網(wǎng)絡(luò)抖動(dòng)就可能永久斷連。8.3 關(guān)于重試冪等和監(jiān)控體系的經(jīng)驗(yàn)把重試次數(shù)、消費(fèi)耗時(shí)、隊(duì)列堆積、未確認(rèn)數(shù)這些指標(biāo)接入監(jiān)控比平時(shí)多花半天時(shí)間但線上出問(wèn)題時(shí)能省下好幾個(gè)不眠夜。RabbitMQ 的 management 插件自帶 Prometheus 端點(diǎn)只需在配置文件里開(kāi)啟management.enable_prometheus_apitrueGrafana 里用現(xiàn)成的 RabbitMQ 面板就能快速搭起來(lái)。最后再分享一個(gè)經(jīng)驗(yàn)任何 RabbitMQ 相關(guān)的代碼改動(dòng)最好先寫(xiě)一個(gè)小型路由自測(cè)測(cè)試類(lèi)用臨時(shí)隊(duì)列驗(yàn)證 Exchange、Routing Key、Binding 是否符合預(yù)期。我每次排查線上路由問(wèn)題都覺(jué)得當(dāng)初如果沒(méi)有這套自測(cè)腳本定位時(shí)間至少翻倍。消息中間件的好處是很多問(wèn)題能在線下環(huán)境復(fù)現(xiàn)把基礎(chǔ)的路由規(guī)則吃透再多線上疑難案例也能還原成某個(gè)概念沒(méi)理解透這一個(gè)原因。