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