做消息中間件這一行最怕的事情不是并發(fā)高不是延遲大而是悄無聲息地丟消息。用戶下單成功結果支付回調(diào)沒觸發(fā)庫存扣減了結果訂單服務沒收到通知積分加了結果異步日志沒落庫。這些看起來“偶發(fā)”的問題背后往往都是消息投遞鏈路中某個環(huán)節(jié)出了問題。而RabbitMQ作為目前使用最廣泛的消息隊列之一雖然本身提供了不少可靠性保障機制但很多人并沒有把這些機制串起來導致消息還是會在特定場景下丟失。這篇內(nèi)容我準備從“可靠性投遞”這個角度完整拆解一套可落地的實現(xiàn)方案從生產(chǎn)端到消費端再從存儲端到兜底補償把每一步的原理、配置和坑都講清楚。如果你正在負責一個基于RabbitMQ的核心鏈路改造或者你的系統(tǒng)已經(jīng)出現(xiàn)過消息丟失、重復消費這類問題這篇文章應該能幫你少走不少彎路。方案涉及生產(chǎn)者Confirm確認機制、Return回退機制、消息持久化、消費者手動ACK、冪等去重、死信隊列、延遲重試以及最終的數(shù)據(jù)對賬屬于一套比較完整的工程化落地思路。1. 先理清楚消息到底是在哪個環(huán)節(jié)丟的1.1 一次消息投遞要經(jīng)過的四個位置在聊實現(xiàn)方案之前我建議先花兩分鐘把消息的完整鏈路在腦子里過一遍。RabbitMQ中一條消息從業(yè)務系統(tǒng)發(fā)出到被最終處理要經(jīng)過四個關鍵位置生產(chǎn)者應用、RabbitMQ Broker、交換機與隊列存儲和消費者應用。流程大致是這樣的生產(chǎn)者把消息發(fā)給交換機交換機根據(jù)路由鍵把消息投遞到綁定的隊列隊列把消息存儲下來然后消費者從隊列拉取或者由Broker推送消息進行消費。你仔細想一下這四個位置中任何一個環(huán)節(jié)出問題消息都可能丟。比較典型的場景包括生產(chǎn)者發(fā)消息時網(wǎng)絡閃斷消息根本沒到Broker消息到了交換機但路由不到任何隊列消息已經(jīng)在隊列里了但Broker宕機并且消息沒有持久化消費者收到消息后處理邏輯報錯或者消費者進程在確認之前崩潰。每一類問題對應的解決方案完全不同如果只是籠統(tǒng)地加一個“手動確認”或者“消息持久化”根本覆蓋不了所有丟消息的場景。1.2 可靠性投遞的完整目標所謂“可靠性投遞”不應該只是某一個開關或者某一項配置而是要從端到端的視角打一套組合拳。我的目標很明確所有發(fā)送成功并完成確認的消息最終一定會被業(yè)務方處理且只被處理一次。注意這里面有兩個關鍵詞一個是“不丟失”另一個是“不重復”。后者在分布式系統(tǒng)中往往比前者更難解決因為消息重復在大多數(shù)情況下難以徹底避免我們能做的只能是讓消費端具備冪等處理能力。這個目標拆解下來對應四層工作生產(chǎn)端的發(fā)送確認、服務端的存儲保障、消費端的確認機制以及最后兜底的補償對賬。下面每一層我都會詳細展開。2. 生產(chǎn)端第一道防線Confirm確認機制和Return回退2.1 為什么默認的發(fā)送方式是不安全的很多剛開始用RabbitMQ的同學最容易犯的一個錯誤就是發(fā)完消息不檢查結果。代碼寫得很簡單聲明一個隊列、一個交換機然后調(diào)用一下基礎發(fā)送方法消息發(fā)出去了就認為萬事大吉。實際上默認情況下生產(chǎn)者是“發(fā)完即忘”的狀態(tài)只要連接不斷消息就算發(fā)送出去了。但問題是發(fā)送操作成功并不代表Broker真的收到了哪怕一個字節(jié)。舉個例子生產(chǎn)者發(fā)送消息時如果RabbitMQ服務端發(fā)生了網(wǎng)絡分區(qū)或者集群中的某個節(jié)點發(fā)生了故障切換客戶端連接斷開那這條消息很可能在TCP緩沖區(qū)里沒有真正送達服務端。在同步發(fā)送場景下客戶端會拋異常但異步發(fā)送場景或者框架封裝了緩沖隊列時這條消息可能就被當成“已發(fā)送”直接吞掉了。所以在實現(xiàn)可靠性投遞方案時第一步就是給生產(chǎn)端加上發(fā)送確認機制。2.2 開啟Publisher Confirm后的核心邏輯RabbitMQ在AMQP 0-9-1協(xié)議基礎上提供了一個擴展能力叫作Publisher Confirm。原理很簡單生產(chǎn)者把信道設置成Confirm模式之后每發(fā)送一條消息Broker在成功接收并持久化如果消息是持久化的之后會給生產(chǎn)者返回一個確認BasicAck如果內(nèi)部發(fā)生了異常導致消息沒有接收成功則返回BasicNack。實際開發(fā)中我建議直接把整個鏈路都開Confirm。使用Spring Boot的RabbitTemplate時可以通過配置文件開啟Confirm回調(diào)核心配置如下spring: rabbitmq: publisher-confirm-type: correlated publisher-returns: true template: mandatory: true這三行配置分別對應什么呢publisher-confirm-type: correlated是說開啟Confirm機制并且采用Correlated模式也就是每條消息會攜帶一個CorrelationData對象用來唯一標識這條消息的確認結果。publisher-returns: true和template.mandatory: true合在一起是讓消息在“無法被路由到任何隊列”時能夠通過Return回調(diào)把消息退回給生產(chǎn)者。否則默認情況下一條發(fā)不到隊列的消息會在交換機這里被靜默丟棄。配置完成后還需要在代碼里注冊對應的回調(diào)比如rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - { if (!ack) { log.error(消息發(fā)送失敗correlationId: {}, cause: {}, correlationData.getId(), cause); // 這里將消息標記為待重發(fā)或直接寫入本地失敗記錄表 } else { log.info(消息發(fā)送成功correlationId: {}, correlationData.getId()); } }); rabbitTemplate.setReturnsCallback(returned - { log.error(消息路由失敗exchange: {}, routingKey: {}, replyText: {}, returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()); // 處理無法路由的消息 });這里有個很重要的細節(jié)Confirm回調(diào)只能證明消息到達了Broker不能證明消息被正確路由到了隊列。也就是說如果交換機找不到匹配的隊列Broker還是會返回Ack給生產(chǎn)者只是額外觸發(fā)一次Return回調(diào)。所以這兩個回調(diào)必須同時處理消息才算是真正安全地“入了隊”。關于這一點我是踩過坑的。早期項目里只做了消息持久化和Confirm確認有一段時間訂單消息偶爾會消失排查了很久后來定位到原因是路由鍵在某個配置中心里被改動了消息發(fā)到交換機后發(fā)現(xiàn)沒有匹配的綁定交換機直接丟棄了消息。因為當時沒有配置mandatory和Return回調(diào)日志里完全看不出來生產(chǎn)端還認為發(fā)出去了實際上消息根本沒進隊列。2.3 Confirm模式下如何選擇同步等待與異步回調(diào)在非Spring Boot環(huán)境下如果你使用的是原生RabbitMQ客戶端也可以通過channel.confirmSelect()開啟Confirm模式。之后有兩種處理方式一種是每條消息發(fā)送后同步調(diào)用waitForConfirms()阻塞等待確認結果另一種是批量發(fā)送后用waitForConfirmsOrDie()統(tǒng)一檢查還有一種是基于監(jiān)聽器的異步處理。我個人比較推薦的方式是在瞬時發(fā)送量大但又不希望丟任何一條消息的場景下采用回調(diào)式異步處理因為同步確認會顯著降低生產(chǎn)者的吞吐量。尤其是在單條消息RTT往返時延在毫秒級以上的情況下同步等待會讓發(fā)送能力出現(xiàn)很大損耗。當然同步等待也有它的好處代碼邏輯簡單不需要維護狀態(tài)。如果是低頻次且對吞吐不敏感的場景完全可以用同步方式。這兩者的本質(zhì)是一樣的都是在發(fā)送端確認Broker已經(jīng)收到消息區(qū)別只在于等待模型不同。實際項目中我需要的是“既要吞吐又要可靠”所以最終選的是Correlated異步確認并且配合本地消息表記錄發(fā)送狀態(tài)。3. 存儲端保障持久化、鏡像隊列與存儲策略3.1 三個持久化缺一不可消息到達Broker之后如果Broker進程突然崩潰或者所在的物理機宕機這時候消息是否安全取決于你做了多少持久化工作。這里有三個必須同時滿足的條件交換機持久化、隊列持久化、消息本身持久化。交換機持久化是在聲明交換機時設置durable參數(shù)為true。隊列持久化同樣是在聲明隊列時設置durable為true。如果交換機或隊列沒有持久化Broker重啟后相關的定義會消失后續(xù)即使消息持久化了存儲的隊列都沒了消息也無處安放。消息持久化是在發(fā)送消息時設置deliveryMode為PERSISTENT值為2在Spring AMQP中對應MessageDeliveryMode.PERSISTENT。比如Message message MessageBuilder.withBody(payload.getBytes()) .setDeliveryMode(MessageDeliveryMode.PERSISTENT) .build(); rabbitTemplate.send(exchange, routingKey, message);需要特別注意這三個地方必須全部持久缺任何一個都可能丟消息。有些同學只設置了durabletrue的隊列然后發(fā)送消息時沒有指定持久化模式默認情況下Spring Boot的RabbitTemplate會根據(jù)消息屬性生成默認的deliveryMode如果你沒有顯式設置它可能是非持久的。非持久消息只要Broker內(nèi)存中還有就不會丟但一重啟就沒了。列出這幾個持久化配置在Spring Boot中的設置位置持久化對象設置方式缺失風險交換機聲明時durabletrue或配置類中設置durable重啟后交換機定義丟失消息無法路由隊列聲明時durabletrue或配置類中設置durable重啟后隊列定義丟失消息無法存儲消息發(fā)送時deliveryModePERSISTENT值為2重啟后消息內(nèi)容丟失3.2 單機持久化夠不夠聊聊鏡像隊列和仲裁隊列持久化解決的是“進程重啟后數(shù)據(jù)還在”的問題但它解決不了“機器徹底宕機”的問題。如果你部署的是單節(jié)點RabbitMQ磁盤損壞這種極端情況仍然可能丟數(shù)據(jù)。所以在生產(chǎn)環(huán)境里高可用部署是可靠性投遞的基本要求。傳統(tǒng)方案是配置鏡像隊列Mirrored Queue用鏡像策略把主隊列的數(shù)據(jù)同步到其他節(jié)點。主節(jié)點負責讀寫從節(jié)點作為熱備份。舊版本中這是比較成熟的模式但在高并發(fā)下鏡像隊列同步存在性能開銷而且主從切換時可能會有短暫的不可用。新版本推薦使用仲裁隊列Quorum Queue它基于Raft協(xié)議實現(xiàn)數(shù)據(jù)在多個節(jié)點上形成副本具備更強的數(shù)據(jù)一致性保證。仲裁隊列要求至少三個節(jié)點才能發(fā)揮出容錯作用適合對數(shù)據(jù)可靠性要求極高的核心鏈路。如果項目已經(jīng)用了5.x版本的RabbitMQ我建議直接切換到仲裁隊列尤其是在處理訂單、支付、庫存這類核心消息時。它的網(wǎng)絡開銷比鏡像隊列更高但換來的是更干凈的故障切換體驗。如果你只是測試環(huán)境或者業(yè)務量不大的系統(tǒng)單節(jié)點加持久化也可以跑但不要把單節(jié)點方案直接照搬到生產(chǎn)環(huán)境這是很多小團隊后來不得不做架構改造的原因。3.3 持久化對性能到底有多大影響聊了這么多可靠性保障肯定有人關心性能問題。先說結論只要開啟持久化和Confirm性能肯定比“發(fā)完即忘”模式差但差得不至于不能用。持久化消息的落盤過程本質(zhì)上就是把消息寫入磁盤日志文件。RabbitMQ的持久化消息會先寫入內(nèi)存隊列同時異步寫入磁盤不會每條消息都同步刷盤所以它是有一個批量寫入機制的。根據(jù)我自己做的壓測結果單節(jié)點環(huán)境下持久化消息加Confirm回調(diào)的模式吞吐量大約是非持久化消息的50%到70%。聽起來降得挺多但在絕大多數(shù)業(yè)務場景下每秒幾十條到幾百條的消息量這點損耗完全可以接受。真正需要關注的不是吞吐量而是延遲。如果你的業(yè)務要求毫秒級響應每條消息都同步等待磁盤刷盤那延遲會變得不穩(wěn)定。對于這種場景建議把消息的可靠性交給對賬和補償機制而不是完全依賴RabbitMQ本身的持久化。這個思路后面我會講到。4. 消費端可靠性手動ACK、死信隊列與冪等消費4.1 自動ACK為什么是萬惡之源消費端丟消息九成以上都是因為默認開啟了自動確認。在Spring AMQP的監(jiān)聽容器中如果把acknowledgeMode設置為AUTO或NONE消費者在收到消息后立即向Broker返回確認不管業(yè)務邏輯是否執(zhí)行成功。這樣一來如果消費方法里拋了異常或者消費者進程在業(yè)務處理到一半時崩潰這條消息已經(jīng)被標記為“已確認”Broker就認為你已經(jīng)成功消費了于是直接把它從隊列中刪除。消息就這樣靜悄悄地丟了。正確做法是使用手動ACK模式也就是acknowledgeMode: MANUAL。在這個模式下業(yè)務邏輯執(zhí)行成功后才主動調(diào)用basicAck方法確認執(zhí)行失敗則調(diào)用basicNack或basicReject根據(jù)參數(shù)決定是丟棄、重新入隊還是進入死信隊列。配置方式如下spring: rabbitmq: listener: simple: acknowledge-mode: manual對應的消費代碼可以這樣寫RabbitListener(queues order.queue) public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { // 業(yè)務邏輯處理 process(message); channel.basicAck(deliveryTag, false); } catch (Exception e) { // 根據(jù)異常類型決定是否重試盡量不要無條件重新入隊 channel.basicNack(deliveryTag, false, false); } }這里有一個非常關鍵的坑basicNack的第三個參數(shù)requeue如果設置為true消息會被重新放回原隊列然后立刻再次投遞給消費者形成一個死循環(huán)。如果消費邏輯里有拋異常的bug消息就會一直在消費者和隊列之間來回彈跳CPU飆高日志刷屏還會把正常消息堵在后面。所以我在處理這個參數(shù)時會非常謹慎普通業(yè)務異常直接basicNack并且requeuefalse讓消息進入死信隊列由專門的補償任務去處理。只有網(wǎng)絡抖動、數(shù)據(jù)庫連接池暫時不可用這類瞬時故障才考慮重新入隊并配合次數(shù)限制。4.2 重試策略的設計不要裸奔式無限重試手動ACK只是解決了“消費失敗不丟失”的問題但接下來會面對另一個問題失敗的這條消息怎么處理最簡單的做法是配合Spring Retry在RabbitListener外層加上重試機制。比如spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 5 initial-interval: 1000 multiplier: 2.0 max-interval: 10000這樣消費失敗后會由Spring框架在內(nèi)存中做指數(shù)退避重試重試次數(shù)超過上限后消息會被丟棄或者進入我們指定的異常處理邏輯。這個方案適合處理那些因為數(shù)據(jù)庫臨時鎖、外部接口瞬時超時等原因導致的短暫不可用。但是要注意Spring Retry的重試是發(fā)生在消費者進程內(nèi)的如果消費者進程在重試過程中重啟內(nèi)存中的重試狀態(tài)就沒了。另一種更可靠的方案是使用RabbitMQ本身的死信隊列機制。核心思路是定義一個業(yè)務隊列綁定一個死信交換機。當消息被拒絕且requeuefalse或消息過期或隊列長度超限時消息會被自動轉發(fā)到死信交換機再由死信交換機路由到死信隊列。專門寫一個消費者監(jiān)聽死信隊列把消息取出來做延遲處理比如保存到數(shù)據(jù)庫、發(fā)送告警、或者轉入定時任務等待后續(xù)重發(fā)。死信隊列的聲明代碼大致是這樣Bean public Queue businessQueue() { return QueueBuilder.durable(business.queue) .deadLetterExchange(dlx.exchange) .deadLetterRoutingKey(dlx.routing.key) .build(); }使用死信隊列帶來的一個額外好處是你有了一個“消息事故現(xiàn)場”可以隨時去查看哪些消息是處理失敗的失敗時的原始內(nèi)容是什么方便排查和審計。這是純內(nèi)存重試不具備的優(yōu)勢。4.3 消費端必須自己做冪等不要指望消息只來一次哪怕我們把前面所有可靠性機制都做對了依然要面對一個現(xiàn)實消息在分布式環(huán)境下是有可能重復投遞的。舉個例子消費者處理完消息后業(yè)務數(shù)據(jù)已經(jīng)寫庫了但在返回確認給Broker的過程中網(wǎng)絡閃斷Broker沒有收到確認于是過一會兒又把同一條消息投遞了過來。這時候如果消費端不做冪等處理就會出現(xiàn)重復扣減庫存、重復加積分、重復發(fā)短信之類的嚴重故障。冪等設計最常用的方案是“唯一業(yè)務鍵已處理記錄表”。在消息體中帶上一個業(yè)務唯一ID比如訂單號、支付流水號。消費者在處理消息前先查一下本地數(shù)據(jù)庫看看這個ID是否已經(jīng)處理過。如果已經(jīng)處理直接確認消息并跳過業(yè)務邏輯。為了避免并發(fā)場景下兩個線程同時查到“未處理”還需要在記錄表中對唯一鍵加數(shù)據(jù)庫唯一約束插入失敗就說明重復了。代碼里一般是這樣一個套路String msgId message.getMessageProperties().getMessageId(); if (processedMsgMapper.exist(msgId)) { channel.basicAck(deliveryTag, false); return; } try { // 在事務內(nèi)同時插入 msgId 與執(zhí)行業(yè)務邏輯利用唯一約束防重 processBusiness(message); processedMsgMapper.insert(msgId); channel.basicAck(deliveryTag, false); } catch (DuplicateKeyException e) { channel.basicAck(deliveryTag, false); }需要說明的是冪等不應該只依賴數(shù)據(jù)庫唯一鍵因為某些業(yè)務場景下消息清理或表歸檔后唯一約束就沒了。更保險的做法是業(yè)務上本身就具備天然冪等性比如“訂單狀態(tài)流轉已支付狀態(tài)下再次收到支付成功回調(diào)直接忽略”這種設計對應的代碼應該寫在業(yè)務邏輯內(nèi)部而不是完全依賴于消息中間件的去重。4.4 手動ACK和冪等是配套的別拆開做我見過一些團隊手動ACK做了但是消費端沒有冪等另一些團隊接口冪等做了但用的是自動ACK。這兩種方案單獨拿出來都有各自的漏洞。手動ACK沒有冪等重復消費時業(yè)務數(shù)據(jù)會出錯自動ACK配合冪等雖然業(yè)務上不會出大問題但消息可能因為消費者進程崩潰而丟失畢竟Broker在你收到消息后立刻就把消息刪掉了。所以我的建議是老老實實把這一整套都配上手動ACK保證“至少一次”投遞消費者業(yè)務冪等保證“最多一次”的業(yè)務影響兩者合在一起才能達到“有效一次”的效果。這也是目前幾乎所有的消息中間件可靠性方案里最核心的共識。5. 兜底補償本地消息表與定時對賬5.1 聊完實時鏈路還得聊聊終極兜底前四節(jié)講的各種機制已經(jīng)可以應對絕大多數(shù)故障場景了。但做過程序的人都知道再完備的鏈路也可能出現(xiàn)“意外中的意外”。比如RabbitMQ集群徹底崩潰且備份也丟了比如消息在某個未知環(huán)節(jié)被手動刪除比如代碼邏輯上存在并發(fā)問題導致部分消息沒有走到發(fā)送邏輯。純靠消息中間件本身的可靠性機制沒法兜住所有問題。因此我一般會在核心業(yè)務鏈路上加一層數(shù)據(jù)最終一致的補償機制。這套機制的核心是本地消息表也叫事務消息表。思路是這樣的在業(yè)務數(shù)據(jù)庫里建一張message_record表業(yè)務操作和消息記錄寫入在同一個本地事務中完成。事務提交之后再通過一個獨立的消息發(fā)送程序把消息從表里撈出來發(fā)送到RabbitMQ。發(fā)送成功后更新消息狀態(tài)為“已發(fā)送”。5.2 表結構和發(fā)送程序怎么設計消息表通常包含這幾個核心字段消息ID、業(yè)務類型、業(yè)務主鍵、消息內(nèi)容JSON格式、狀態(tài)待發(fā)送、已發(fā)送、發(fā)送失敗、已確認、重試次數(shù)、創(chuàng)建時間、最后修改時間。在這里消息ID整個producer端的唯一標識也就是前面Confirm回調(diào)里用到的CorrelationData對應的ID。發(fā)送程序可以做成一個定時任務每隔幾秒掃描一次狀態(tài)為“待發(fā)送”且重試次數(shù)未超上限的消息調(diào)用RabbitTemplate發(fā)送然后等待Confirm回調(diào)更新狀態(tài)。如果回調(diào)里返回失敗就把消息狀態(tài)改成“發(fā)送失敗”等待下一次掃描重試。這個方案的巧妙之處在于消息的可靠性從“依賴RabbitMQ不丟消息”變成了“依賴業(yè)務系統(tǒng)自己有一個可以反復重放的待發(fā)送隊列”。只要數(shù)據(jù)庫不丟消息總能發(fā)出去。偽代碼如下Scheduled(fixedDelay 5000) public void scanAndSend() { ListMessageRecord records messageRecordMapper.scanPending(); for (MessageRecord record : records) { CorrelationData correlationData new CorrelationData(record.getMessageId()); try { rabbitTemplate.convertAndSend(record.getExchange(), record.getRoutingKey(), record.getMessageContent(), correlationData); } catch (Exception e) { record.setRetryCount(record.getRetryCount() 1); messageRecordMapper.updateStatus(record.getId(), SEND_FAILED); } } }定時任務發(fā)送之后還需要按計劃定期清理已經(jīng)完成的消息記錄避免表無限膨脹。根據(jù)消息量的大小可以保留最近7天到30天超期的按照消息ID歸檔或者直接刪除。5.3 補償機制能不能代替Confirm機制這個問題經(jīng)常有人問。既然有了本地消息表和定時任務是不是可以不開Confirm了答案是不行。本地消息表和定時任務解決的是“消息發(fā)送前”的可靠性它保證了消息一定會被發(fā)送程序嘗試發(fā)送但不保證發(fā)送之后Broker是否真的收到。如果發(fā)送程序把消息交給了RabbitTemplate但由于網(wǎng)絡分區(qū)等原因Broker沒有收到RabbitTemplate內(nèi)部已經(jīng)把消息包裝好了而發(fā)送程序又因為邏輯異常退出了消息狀態(tài)仍然停留在“待發(fā)送”。這沒問題下次掃描還會重發(fā)。但如果發(fā)送程序在邏輯上認為“發(fā)送成功”了實際上Broker沒收到那這條消息就會卡在“已發(fā)送”狀態(tài)再也不會被掃描到。所以仍然需要Confirm回調(diào)來更新真正的確認狀態(tài)。就算你不開Confirm也要至少檢查發(fā)送結果或使用同步發(fā)送的異常捕獲機制否則本地消息表就形同虛設了。從可靠性角度排序最完善的是本地消息表負責“一定發(fā)得出去”Confirm機制負責“發(fā)出去了沒”持久化負責“Broker存儲不丟”手動ACK負責“消費者處理不丟”冪等設計負責“重復了也沒事”。五層各司其職缺一層都有漏洞。6. 常見問題排查實錄與避坑技巧6.1 消息丟失的經(jīng)典排查路徑日常運維和開發(fā)過程中遇到消息丟失問題最忌諱的是直接去翻代碼而是應該按鏈路一層層排查。我一般會按照下面這個順序來先看生產(chǎn)端有沒有發(fā)送日志消息是否真的調(diào)用了RabbitTemplate的發(fā)送方法。再看Confirm回調(diào)有沒有觸發(fā)返回的是Ack還是Nack有沒有Return回調(diào)。如果Ack返回了那就看Broker端隊列深度是否增加消息是否在隊列中滯留。如果隊列有消息但消費者沒消費檢查消費者的日志和監(jiān)聽容器狀態(tài)。如果隊列沒消息但確認已經(jīng)消費了看消費端有沒有成功執(zhí)行業(yè)務、有沒有手動Ack。如果手動Ack了但沒有業(yè)務數(shù)據(jù)多半是消費邏輯里異常被吞了這類問題需要仔細看鏈路日志。在實際定位中我會依賴RabbitMQ管理界面的隊列狀態(tài)來輔助判斷。例如一條消息發(fā)出去后隊列深度不變說明消息可能沒有進隊列隊列深度增加了但消費者一直沒消費說明消費端可能在報錯或者網(wǎng)絡不通隊列深度為零但數(shù)據(jù)庫沒有數(shù)據(jù)說明消費端Ack了但業(yè)務邏輯有問題。這種逐層推理的方式比單純看代碼更高效。6.2 高頻踩坑場景速查表下面這些場景在我參與和接觸過的項目中出現(xiàn)的頻率非常高我整理成了一張速查表問題現(xiàn)象可能原因解決思路消息確認回調(diào)返回Nack交換機或隊列不存在或Broker內(nèi)部IO異常檢查交換機隊列是否聲明檢查落盤空間Confirm回調(diào)一直不觸發(fā)CorrelationData未設置或回調(diào)線程異常檢查是否設置publisher-confirm-type回調(diào)異常是否被吞消息進了隊列但消費者不消費消費者應用未啟動或監(jiān)聽容器線程數(shù)耗盡查看消費者連接狀態(tài)擴容并發(fā)線程數(shù)消費端報錯后消息消失使用了自動Ack或手動Ack前拋異常且未進入Nack分支改為手動Ack異常時統(tǒng)一進入Nack/死信邏輯重復消費導致數(shù)據(jù)重復消費端沒有冪等處理增加唯一鍵去重業(yè)務冪等處理消息被無限重新入隊循環(huán)requeuetrue且消費一直失敗設置requeuefalse配合死信隊列隊列一直堆積消費者CPU飆升單條消息處理時間過長或者消費死循環(huán)查看消費者日志優(yōu)化處理耗時必要時拆分隊列消息丟失且發(fā)送日志顯示成功RabbitMQ持久化未開啟或消息未持久化確認三個持久化點尤其是消息deliveryMode確認模式下發(fā)送吞吐量下降明顯使用了同步確認或持久化日志刷盤頻繁改異步Confirm回調(diào)優(yōu)化刷盤策略6.3 每次上線前必做的消息鏈路檢查可靠性方案做得再好如果上線時無意中改錯了一個配置照樣會出事。我自己在項目上線前會強制要求團隊完成幾個檢查項。確認所有核心業(yè)務隊列聲明時都設置了durabletrue不能因為開發(fā)環(huán)境沒開持久化就默認照搬配置。確認消費者監(jiān)聽容器設置的是acknowledge-mode: manual這是代碼配置層最容易漏掉的地方。確認生產(chǎn)端RestTemplate或生產(chǎn)連接工廠上開啟了Confirm和Return生產(chǎn)環(huán)境不能用測試環(huán)境的純內(nèi)存模式。確認死信隊列消費者存在不然消息進入死信后沒人處理堆積起來毫無意義。確認關鍵業(yè)務鏈路有對賬或者告警機制比如隊列積壓超過一定閾值時釘釘或者企業(yè)微信要能收到告警。如果能做到這幾點可靠性投遞方案基本上就比較扎實了。6.4 追查消息軌跡的實踐建議最后再分享一個我自己在項目中用的技巧。RabbitMQ本身沒有內(nèi)置消息軌跡追蹤的效果但我們可以自己給每一條消息打上全局唯一的消息ID同時在所有生產(chǎn)和消費的關鍵節(jié)點打印結構化日志。日志里至少包含消息ID、業(yè)務ID、交換機、路由鍵、隊列、時間戳、操作類型發(fā)送/確認/入隊/消費/Ack/Nack。比如rabbitTemplate.convertAndSend(exchange, routingKey, message, new MessagePostProcessor() { Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setMessageId(UUID.randomUUID().toString()); return message; } });當消息在某個環(huán)節(jié)丟失時拿著這個消息ID去日志平臺搜索就能快速定位它最后出現(xiàn)在哪個節(jié)點。這個辦法比任何監(jiān)控面板都直觀尤其是在排查跨團隊協(xié)作的系統(tǒng)時日志里有沒有消息ID是能否高效扯皮、快速定位問題的關鍵。我強烈建議在核心鏈路中把消息ID做為一個必填字段。7. 我的最后幾點總結寫到這里可靠性投遞方案的核心內(nèi)容差不多講完了。回顧一下最核心的要點發(fā)送端開啟Confirm和Return機制確認消息真的到達了隊列存儲端做好交換機、隊列、消息三重持久化部署盡量使用鏡像隊列或仲裁隊列消費端改手動ACK結合重試機制和死信隊列消費邏輯一定要具備冪等能力兜底層使用本地消息表加定時任務確保“即使RabbitMQ全掛也能通過重放數(shù)據(jù)庫記錄把消息找回來”。這里要特別注意不要覺得“配置多、步驟多”就覺得麻煩而偷懶。在核心業(yè)務上省掉的每一步可靠性工作后面都會以線上故障的形式讓你加倍償還。我見過不少系統(tǒng)初期跑得很順等用戶量和業(yè)務復雜度上來了消息丟失引發(fā)的數(shù)據(jù)不一致問題集中爆發(fā)最后不得不返工加各種補償機制風險和成本都高得多。也別過度設計比如那些對數(shù)據(jù)一致性要求不高的通知類消息比如“用戶瀏覽記錄統(tǒng)計”“推薦位點擊數(shù)據(jù)”這些場景丟了也無所謂就不需要付出額外的性能代價?？煽啃允欠旨墑e的核心鏈路和非核心鏈路要有不同的方案這一點也是很多年在生產(chǎn)環(huán)境里實踐出來的體會。如果你正準備改造消息這塊不要一口吃成胖子先把Confirm、手動ACK、持久化和冪等這幾件事做了再把死信隊列和本地消息表逐步加上一點點往“最終一致”的方向靠攏系統(tǒng)的穩(wěn)定性自然會慢慢提上來。這應該是當前成本最低、見效最快的一條路徑了。