戰(zhàn)：微服務(wù)異步解耦與消息可靠性保障)
從實(shí)際項(xiàng)目里摸爬滾打過來的人對(duì)微服務(wù)之間的通信問題應(yīng)該都深有體會(huì)。一個(gè)電商訂單下來要通知庫存服務(wù)扣庫存、通知積分服務(wù)加積分、通知短信服務(wù)發(fā)通知如果全部用 Feign 同步調(diào)用一個(gè)服務(wù)抖動(dòng)整條鏈路就卡死數(shù)據(jù)庫連接池一滿全站跟著雪崩。我在項(xiàng)目里引入 SpringCloud 集成 RabbitMQ 之后這套問題才算真正從根上解決了。這篇就把微服務(wù)場景下使用 RabbitMQ 的完整實(shí)踐過程講清楚從協(xié)議原理、消息可靠性保障、死信隊(duì)列、手動(dòng)確認(rèn)到和 RocketMQ、Kafka 的選型對(duì)比以及我實(shí)際部署和排查問題時(shí)踩過的坑一次性說透。1. 微服務(wù)為什么需要 RabbitMQ核心場景與選型思考1.1 什么場景一定會(huì)用到 RabbitMQ很多新手看完教程會(huì)問我寫個(gè) CRUD 接口直接 HTTP 調(diào)用不就完了為什么非要多加一個(gè)中間件這個(gè)疑問很正常但當(dāng)你面臨下面幾類問題的時(shí)候單純靠同步接口是無解的。第一類是異步解耦。用戶下單后核心鏈路是扣庫存、生成訂單、返回支付鏈接這些必須在幾百毫秒內(nèi)完成。但下單后還要發(fā)短信、發(fā) APP 推送、更新用戶積分、給推薦系統(tǒng)發(fā)行為數(shù)據(jù)這些操作每增加一個(gè)同步調(diào)用接口耗時(shí)就增加幾百毫秒而且積分服務(wù)掛了訂單服務(wù)也得跟著回滾這不合理。把短信推送、積分更新等操作丟進(jìn) MQ訂單接口只需要等隊(duì)列寫入成功耗時(shí)天然降下來下游服務(wù)慢一點(diǎn)甚至臨時(shí)宕機(jī)都不影響主鏈路。第二類是流量削峰。秒殺場景下瞬時(shí) QPS 可能到上萬但數(shù)據(jù)庫只能扛幾百的并發(fā)寫。直接在業(yè)務(wù)代碼里寫數(shù)據(jù)庫瞬間就被打爆。用 MQ 擋在前面請(qǐng)求先全部塞進(jìn)隊(duì)列后端訂單消費(fèi)服務(wù)按自己數(shù)據(jù)庫能承受的速度比如每秒鐘處理 500 條慢慢消費(fèi)削峰填谷的效果立竿見影。這里有一個(gè)關(guān)鍵的定量經(jīng)驗(yàn)我會(huì)先壓測出數(shù)據(jù)庫的寫 TPS 上限再把消費(fèi)者的 prefetch 值設(shè)置成這個(gè) TPS 的一半留一半余量給其他業(yè)務(wù)查詢實(shí)測下來穩(wěn)定得多。第三類是數(shù)據(jù)廣播。微服務(wù)架構(gòu)里一個(gè)用戶下單事件訂單服務(wù)要通知審計(jì)服務(wù)、風(fēng)控服務(wù)、搜索服務(wù)、大數(shù)據(jù)平臺(tái)。如果用 Feign 逐個(gè)調(diào)用每加一個(gè)下游服務(wù)訂單服務(wù)代碼就要改一版。換成 MQ 的 Fanout 交換機(jī)訂單服務(wù)只管往交換機(jī)發(fā)一條消息所有綁定了這個(gè)交換機(jī)的隊(duì)列都能收到自己的副本下游服務(wù)增減完全不影響上游代碼。這種發(fā)布訂閱模型在架構(gòu)演進(jìn)時(shí)價(jià)值極大。一句話講清楚 RabbitMQ 的定位上游只負(fù)責(zé)把消息可靠地發(fā)出去下游按自己的節(jié)奏和需求來消費(fèi)雙方通過隊(duì)列解耦誰出問題都不影響對(duì)方。1.2 選型對(duì)比RabbitMQ 和 RocketMQ、Kafka 怎么選關(guān)于 RabbitMQ、RocketMQ 和 Kafka 的區(qū)別網(wǎng)上說法很多我結(jié)合自己用過的實(shí)際場景給出一個(gè)比較務(wù)實(shí)的看法。首先明確一個(gè)點(diǎn)沒有任何一個(gè) MQ 是絕對(duì)萬能的。先看 RabbitMQ。它是 Erlang 寫的對(duì) AMQP 協(xié)議支持非常完備路由規(guī)則在三種 MQ 里最靈活。社區(qū)認(rèn)知中它的吞吐量上限通常低于 Kafka但單機(jī)幾萬 QPS 對(duì)絕大多數(shù)業(yè)務(wù)系統(tǒng)完全夠用。而且它有一套很成熟的管理控制臺(tái)消息追蹤、隊(duì)列堆積、連接數(shù)監(jiān)控運(yùn)維成本低到驚人中小團(tuán)隊(duì)首選基本不用養(yǎng)專門的 MQ 運(yùn)維。RocketMQ 是阿里開源給 Java 生態(tài)使用的消息隊(duì)列特點(diǎn)是有事務(wù)消息可以保證本地事務(wù)和發(fā)消息的原子性這在支付、轉(zhuǎn)賬場景里很關(guān)鍵。但帶來的問題是部署運(yùn)維重NameServer、Broker、主從同步、Console 一套下來比 RabbitMQ 復(fù)雜不少。Kafka 的定位是日志采集和流處理吞吐量確實(shí)恐怖每秒幾十萬條毫秒級(jí)延遲。但 Kafka 本身不擅長復(fù)雜路由topic 設(shè)計(jì)簡單而且消費(fèi)端要自己維護(hù) offset對(duì)業(yè)務(wù)開發(fā)者來說心智負(fù)擔(dān)明顯更重。數(shù)據(jù)管道、用戶行為日志這種場景用 Kafka 沒問題拿它做訂單業(yè)務(wù)就有點(diǎn)高射炮打蚊子了。在我參與的項(xiàng)目里選型依據(jù)很簡單業(yè)務(wù)消息量日幾百萬級(jí)以下、看重路由靈活性和開發(fā)效率的用 RabbitMQ涉及嚴(yán)格分布式事務(wù)、必須事務(wù)消息兜底的用 RocketMQ海量日志、指標(biāo)數(shù)據(jù)采集秒級(jí)吞吐要求嚇人的用 Kafka。各干各最擅長的別混用。2. 集成前的環(huán)境準(zhǔn)備安裝部署與核心概念2.1 Windows 和 Docker 環(huán)境下的 RabbitMQ 安裝我最早是直接在 Windows 上裝 RabbitMQ 的。這里必須說新手第一次裝十個(gè)有八個(gè)會(huì)卡在 Erlang 版本號(hào)和 RabbitMQ 版本不匹配上。RabbitMQ 對(duì) Erlang 版本有嚴(yán)格對(duì)應(yīng)關(guān)系看一眼官方版本對(duì)照表再下載能避開一大半的坑。裝的時(shí)候先把 Erlang 裝了配好 ERLANG_HOME 環(huán)境變量再裝 RabbitMQ最后裝 rabbitmq_management 插件。命令行窗口用管理員身份運(yùn)行rabbitmq-plugins enable rabbitmq_management裝完訪問http://localhost:15672默認(rèn)賬號(hào) guest / guest 就能進(jìn)管理界面了。但如果你是在云服務(wù)器上部署guest 默認(rèn)只能 localhost 登錄遠(yuǎn)程訪問必須另建賬號(hào)并配置 vhost 權(quán)限。后來圖省事我直接改用 Docker 了是真省心。一條命令依賴全給你裝好docker run -d --name rabbitmq \ -p 5672:5672 -p 15672:15672 \ -e RABBITMQ_DEFAULT_USERadmin \ -e RABBITMQ_DEFAULT_PASSadmin123 \ rabbitmq:3.12-management注意鏡像 tag 一定要帶 management 后綴否則起來的鏡像沒有管理界面插件。我踩過一次坑用的 rabbitmq:3.12 不帶后綴容器起來了 5672 端口通15672 死活不回應(yīng)后來才發(fā)現(xiàn)是鏡像本身沒帶控制臺(tái)插件。2.2 核心概念掃盲交換機(jī)、隊(duì)列、路由鍵的關(guān)系不把 RabbitMQ 的消息模型搞清楚后面寫代碼就是抄模板出了問題不會(huì)排查。這里用生活里的事打比方來理解。想象一個(gè)快遞中轉(zhuǎn)站。生產(chǎn)者就是寄快遞的商家交換機(jī)Exchange就是中轉(zhuǎn)站的調(diào)度臺(tái)隊(duì)列Queue就是快遞員手上的配送單消費(fèi)者就是收快遞的人。商家寄件時(shí)只需要把包裹交給調(diào)度臺(tái)并說清楚目的地調(diào)度臺(tái)根據(jù)他寫的地址把包裹分配到對(duì)應(yīng)的配送單上快遞員按配送單派送收件人接收。這里商家說的目的地在 RabbitMQ 里叫路由鍵Routing Key調(diào)度臺(tái)按什么規(guī)則分配包裹取決于交換機(jī)類型。有三種常用的交換機(jī)Direct 交換機(jī)完全匹配路由鍵精確投遞一對(duì)一的場景最常用Topic 交換機(jī)按通配符模糊匹配比如order.*能匹配order.create和order.cancel適合需要按消息類型分類消費(fèi)的場景Fanout 交換機(jī)不看路由鍵直接廣播給所有綁定的隊(duì)列就像廣播電臺(tái)所有收音機(jī)都能收到。還有一個(gè) Headers 交換機(jī)按消息頭的 key-value 匹配實(shí)際業(yè)務(wù)里用得少我基本不碰。你還要理解Binding這個(gè)概念就是交換機(jī)和隊(duì)列之間建立的那條配送規(guī)則。生產(chǎn)者在綁定交換機(jī)的時(shí)候要同時(shí)指定隊(duì)列名稱和路由鍵。Bean public Binding binding() { return BindingBuilder .bind(orderQueue()) .to(orderExchange()) .with(order.create); }這段話的含義是消息發(fā)到orderExchange這個(gè)交換機(jī)時(shí)如果路由鍵是order.create就進(jìn)入orderQueue這個(gè)隊(duì)列。很多新手把 Queue 上的 name 和 Binding 的 with 混為一談其實(shí)前者是隊(duì)列名后者是路由鍵兩個(gè)概念別搞混了。3. SpringCloud 項(xiàng)目接入從依賴配置到第一個(gè)可靠消息3.1 依賴引入與核心配置項(xiàng)解析在 SpringCloud 項(xiàng)目里集成 RabbitMQ用的是 Spring Boot 的 starter 封裝。基礎(chǔ)依賴就一個(gè)dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency如果項(xiàng)目用的是 SpringCloud 版本管理 BOM不需要額外指定版本號(hào)它會(huì)自動(dòng)匹配兼容版本。這里我說一個(gè)重要經(jīng)驗(yàn)永遠(yuǎn)不要手工指定 spring-boot-starter-amqp 的版本號(hào)讓 SpringCloud BOM 統(tǒng)一管理。我見過一次事故同事手動(dòng)指定版本結(jié)果和當(dāng)前 SpringCloud 版本不兼容消息發(fā)送時(shí) SerializationException 滿天飛查了半天才發(fā)現(xiàn)是版本沖突。application.yml 里核心配置項(xiàng)是這些spring: rabbitmq: host: 127.0.0.1 port: 5672 username: admin password: admin123 virtual-host: / publisher-confirm-type: correlated publisher-returns: true listener: simple: acknowledge-mode: manual prefetch: 50 retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2 default-requeue-rejected: false逐項(xiàng)講一下這些配置背后的邏輯這是面試官最愛問的點(diǎn)也是實(shí)際線上出問題時(shí)要調(diào)的參數(shù)。publisher-confirm-type: correlated是開啟生產(chǎn)者確認(rèn)發(fā)送端能確知消息是否真正到達(dá)交換機(jī)。消息到達(dá)交換機(jī)后RabbitMQ 會(huì)回調(diào)一個(gè) ConfirmCallback 告訴你發(fā)送成功還是失敗。publisher-returns: true是開啟消息未投遞到隊(duì)列時(shí)的退回機(jī)制也就是消息到了交換機(jī)但沒匹配到任何隊(duì)列RabbitMQ 會(huì)把消息退回給生產(chǎn)者并觸發(fā) ReturnCallback。這兩條加一起才保證從生產(chǎn)者到隊(duì)列這一段不丟消息。acknowledge-mode: manual是手動(dòng) ACK這是線上生產(chǎn)環(huán)境必須的設(shè)置。默認(rèn)的 auto 模式在消費(fèi)者拋出異常時(shí)會(huì)自動(dòng)確認(rèn)消息消息就丟了。手動(dòng)模式把消息確認(rèn)的時(shí)機(jī)交給代碼控制消費(fèi)者處理成功后手動(dòng)調(diào)用 basicAck 告訴 RabbitMQ 可以刪除消息處理失敗時(shí)調(diào)用 basicNack告訴 RabbitMQ 這條消息我沒處理好你看著辦。prefetch是消費(fèi)者每次從隊(duì)列拉取多少條消息到本地緩存。設(shè)太大會(huì)導(dǎo)致消費(fèi)者內(nèi)存被占滿而且某條消息處理時(shí)間過長其他消息一直等在緩存里得不到處理設(shè)太小則吞吐量上不去。我常用的做法是結(jié)合服務(wù)端的線程池大小和業(yè)務(wù)處理耗時(shí)來定處理 50ms 以內(nèi)的消息prefetch 設(shè) 50~100 很合理處理 500ms 以上的消息prefetch 建議壓到 10 以下避免本地積壓太多消息導(dǎo)致內(nèi)存抖動(dòng)。3.2 生產(chǎn)者代碼確認(rèn)回調(diào)與 Return 回退配置只是第一步真正落實(shí)到代碼層面生產(chǎn)者的可靠發(fā)送要這樣寫。我先建一個(gè)配置類注冊(cè)交換機(jī)、隊(duì)列和綁定關(guān)系然后再建發(fā)送消息的 Service。先看交換機(jī)、隊(duì)列與綁定關(guān)系的聲明Configuration public class MqQueueConfig { public static final String ORDER_EXCHANGE order.exchange; public static final String ORDER_QUEUE order.queue; public static final String ORDER_ROUTING_KEY order.create; Bean public DirectExchange orderExchange() { return new DirectExchange(ORDER_EXCHANGE, true, false); } Bean public Queue orderQueue() { return QueueBuilder.durable(ORDER_QUEUE).build(); } Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()) .to(orderExchange()) .with(ORDER_ROUTING_KEY); } }這里要留意的細(xì)節(jié)交換機(jī)、隊(duì)列的 durable 參數(shù)都設(shè)為 true代表持久化。這樣 RabbitMQ 重啟后交換機(jī)和隊(duì)列不會(huì)消失。隊(duì)列聲明的時(shí)候我推薦用QueueBuilder.durable().build()這種鏈?zhǔn)綄懛ê罄m(xù)要加 TTL、死信參數(shù)的時(shí)候直接往這個(gè) Builder 上追加即可可讀性比 new Queue(name, true, false, false) 直觀得多。代碼里最好用常量把交換機(jī)名、隊(duì)列名、路由鍵集中管理別在業(yè)務(wù)方法里隨手寫字符串項(xiàng)目大了你根本找不到誰在消費(fèi)誰的消息。然后是發(fā)送消息的核心代碼。我需要讓 RabbitTemplate 在消息確認(rèn)和消息退回時(shí)都留下日志方便排查Service Slf4j public class OrderMessageProducer { private final RabbitTemplate rabbitTemplate; public OrderMessageProducer(RabbitTemplate rabbitTemplate) { this.rabbitTemplate rabbitTemplate; // 消息到達(dá)交換機(jī)的確認(rèn)回調(diào) rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - { if (ack) { log.info(消息發(fā)送成功correlationId {}, correlationData.getId()); } else { log.error(消息發(fā)送失敗correlationId {}, cause {}, correlationData.getId(), cause); } }); // 消息未投遞到隊(duì)列的退回回調(diào) rabbitTemplate.setReturnsCallback(returned - { log.error(消息被退回exchange {}, routingKey {}, body {}, returned.getExchange(), returned.getRoutingKey(), new String(returned.getMessage().getBody())); }); } public void sendOrderMessage(String orderJson) { CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString()); rabbitTemplate.convertAndSend( MqQueueConfig.ORDER_EXCHANGE, MqQueueConfig.ORDER_ROUTING_KEY, orderJson, correlationData ); } }第二個(gè)方法第一步setConfirmCallback通常在構(gòu)造器里執(zhí)行一次就夠了因?yàn)樗侨旨?jí)別的回調(diào)。實(shí)際操作里很多人是在每次 send 方法前回調(diào)設(shè)置一遍也能跑通但沒有必要會(huì)對(duì)后續(xù)排查日志造成一定干擾。注意這里correlationData的 id 我用 UUID 生成目的就是能在回調(diào)日志里精確定位到某一條消息排查線上問題時(shí)特別有用。還有個(gè)細(xì)節(jié)convertAndSend方法發(fā)送字符串的時(shí)候?qū)嶋H走的是 SimpleMessageConverter 默認(rèn)的序列化方式將字符串轉(zhuǎn)成 UTF-8 字節(jié)數(shù)組。如果你直接發(fā)一個(gè) Java 對(duì)象默認(rèn)會(huì)用 JDK 序列化那么消費(fèi)者端也要用 JDK 反序列化類名還要完全一致這對(duì)微服務(wù)之間類共享有很強(qiáng)約束。我自己項(xiàng)目中通常是發(fā)送 JSON 字符串讓每個(gè)微服務(wù)自己反序列化成自己的 DTO避免模塊間強(qiáng)依賴這是微服務(wù)架構(gòu)的基本原則。3.3 消費(fèi)者代碼手動(dòng) ACK 與重試機(jī)制的配合消費(fèi)者這端是坑最多的地方。很多教程里只寫一個(gè)RabbitListener加上RabbitHandler方法執(zhí)行完就算完事了。但真實(shí)生產(chǎn)環(huán)境必須處理消費(fèi)失敗、消息重試、冪等這幾個(gè)問題。先看一個(gè)標(biāo)準(zhǔn)的手動(dòng) ACK 消費(fèi)示例Component Slf4j public class OrderMessageConsumer { RabbitListener(queues MqQueueConfig.ORDER_QUEUE) public void handleOrderCreate(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); String body new String(message.getBody(), StandardCharsets.UTF_8); try { log.info(收到訂單消息{}, body); OrderDTO orderDTO JSON.parseObject(body, OrderDTO.class); // 核心業(yè)務(wù)邏輯扣庫存、更新訂單狀態(tài)、記錄日志 orderService.handleOrderCreated(orderDTO); // 業(yè)務(wù)處理成功手動(dòng)確認(rèn) channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error(訂單消息處理失敗deliveryTag {}, deliveryTag, e); // 確認(rèn)是否已處理過 if (isProcessed(orderDTO.getId())) { channel.basicAck(deliveryTag, false); return; } // 處理失敗不重回隊(duì)列轉(zhuǎn)為死信 channel.basicNack(deliveryTag, false, false); } } }手動(dòng) ACK 的三個(gè)方法要搞清楚basicAck(deliveryTag, false)確認(rèn)成功RabbitMQ 刪除這條消息。第二個(gè)參數(shù) false 表示只確認(rèn)當(dāng)前這條消息不批量。basicNack(deliveryTag, false, true)第三個(gè)參數(shù) true 表示重回隊(duì)列注意這非常危險(xiǎn)如果消息本身有問題回到隊(duì)列會(huì)被再消費(fèi)一次然后再次 Nack形成無限循環(huán)消費(fèi)把 CPU 打滿日志刷屏。basicNack(deliveryTag, false, false)不重回隊(duì)列直接丟棄或進(jìn)死信。生產(chǎn)環(huán)境的重試機(jī)制我推薦用 Spring AMQP 自帶的retry配置而不是在消費(fèi)者代碼里自己寫 for 循環(huán)。配置文件里的 retry 參數(shù)組合起來是這樣的邏輯消費(fèi)者處理拋出異常后Spring 會(huì)按initial-interval開始重試每次重試間隔按multiplier倍數(shù)遞增最大嘗試次數(shù)是max-attempts。三次都失敗之后配合default-requeue-rejected: false這條消息才會(huì)被拒絕并投遞到死信交換機(jī)。所以在消費(fèi)者方法內(nèi)部我只需要拋異常就行重試過程交給框架處理代碼更干凈。還有一個(gè)容易忽略的關(guān)鍵點(diǎn)消費(fèi)者方法的冪等性。RabbitMQ 的投遞語義是至少一次也就是說極端情況下消息雖然被消費(fèi)成功了但 ACK 在網(wǎng)絡(luò)傳輸中丟失RabbitMQ 端認(rèn)為消息沒被消費(fèi)會(huì)重投。消費(fèi)者必須保證同一筆訂單消息被消費(fèi)兩次和一次結(jié)果一致。我常用的做法是在訂單表上建一個(gè)order_event_id唯一索引消費(fèi)前先查詢或嘗試插入一條冪等記錄能插入就繼續(xù)業(yè)務(wù)處理插入沖突說明已經(jīng)處理過直接投遞成功 ACK。這也解釋了為什么代碼里要有isProcessed(orderDTO.getId())這個(gè)判斷沒有這一層的系統(tǒng)是不完整的。4. 微服務(wù)實(shí)戰(zhàn)拆解死信隊(duì)列、延遲消息與削峰實(shí)踐4.1 訂單超時(shí)關(guān)閉場景TTL 與死信隊(duì)列組合實(shí)現(xiàn)延遲消息很多人剛接觸 RabbitMQ 時(shí)都聽說過它可以做延遲消息。但 RabbitMQ 本身并沒有直接提供延遲隊(duì)列插件支持實(shí)現(xiàn)延遲消息的經(jīng)典方案是利用死信隊(duì)列 消息 TTL。先解釋什么叫死信。消息在隊(duì)列里處于以下幾種狀態(tài)時(shí)就變成了死信消息被消費(fèi)者拒絕且不重回隊(duì)列消息 TTL 到期未被消費(fèi)隊(duì)列長度達(dá)到上限導(dǎo)致消息被丟棄。死信不會(huì)自己消失如果隊(duì)列配置了死信交換機(jī)RabbitMQ 會(huì)把死信重新投遞到指定的死信交換機(jī)再由死信交換機(jī)路由到對(duì)應(yīng)的死信隊(duì)列。下面的代碼聲明了一個(gè) 30 秒 TTL 的訂單超時(shí)隊(duì)列并綁定了死信交換機(jī)Configuration public class OrderTimeoutMqConfig { public static final String ORDER_DELAY_EXCHANGE order.delay.exchange; public static final String ORDER_DELAY_QUEUE order.delay.queue; public static final String ORDER_DEAD_EXCHANGE order.dead.exchange; public static final String ORDER_DEAD_QUEUE order.dead.queue; // 延遲隊(duì)列消息在這里等 30 秒 Bean public DirectExchange orderDelayExchange() { return new DirectExchange(ORDER_DELAY_EXCHANGE, true, false); } Bean public Queue orderDelayQueue() { return QueueBuilder.durable(ORDER_DELAY_QUEUE) .ttl(30000) .deadLetterExchange(ORDER_DEAD_EXCHANGE) .deadLetterRoutingKey(order.timeout) .build(); } // 死信交換機(jī)與死信隊(duì)列消息過期后到這里 Bean public DirectExchange orderDeadExchange() { return new DirectExchange(ORDER_DEAD_EXCHANGE, true, false); } Bean public Queue orderDeadQueue() { return QueueBuilder.durable(ORDER_DEAD_QUEUE).build(); } Bean public Binding delayBinding() { return BindingBuilder.bind(orderDelayQueue()) .to(orderDelayExchange()) .with(order.delay); } Bean public Binding deadBinding() { return BindingBuilder.bind(orderDeadQueue()) .to(orderDeadExchange()) .with(order.timeout); } }這里有一個(gè)需要特別提醒的隊(duì)列級(jí)別的 TTL 一旦設(shè)置隊(duì)列里的所有消息共享這個(gè)過期時(shí)間。如果你需要在同一個(gè)隊(duì)列中放不同延遲時(shí)間的消息比如訂單超時(shí)關(guān)閉是 30 秒自動(dòng)確認(rèn)收貨是 7 天就不能用隊(duì)列級(jí) TTL需要單獨(dú)建隊(duì)列。隊(duì)列級(jí) TTL 還有個(gè)大坑一條消息只在隊(duì)列頭部被檢查是否過期如果隊(duì)列頭部那條消息的 TTL 很長后面的消息即使 TTL 更短也只能等頭部消息先被消費(fèi)或過期。所以實(shí)際業(yè)務(wù)里不同延遲時(shí)間務(wù)必拆分到不同隊(duì)列。下單時(shí)訂單服務(wù)把訂單號(hào)發(fā)到order.delay.exchange路由鍵order.delay消息進(jìn)入order.delay.queue等待 30 秒。30 秒后消息變?yōu)樗佬磐哆f到order.dead.exchange再路由到order.dead.queue。此時(shí)專門負(fù)責(zé)關(guān)閉超時(shí)訂單的消費(fèi)者從死信隊(duì)列拿到訂單號(hào)查詢訂單狀態(tài)如果還是未支付則改成已超時(shí)關(guān)閉Component Slf4j public class OrderTimeoutConsumer { RabbitListener(queues OrderTimeoutMqConfig.ORDER_DEAD_QUEUE) public void handleTimeout(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); String orderId new String(message.getBody(), StandardCharsets.UTF_8); try { log.info(收到訂單超時(shí)消息orderId {}, orderId); orderService.closeTimeoutOrder(orderId); channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error(處理訂單超時(shí)消息失敗orderId {}, orderId, e); channel.basicNack(deliveryTag, false, false); } } }如果你想在消息級(jí)別設(shè)置獨(dú)立的 TTL也可以用MessageProperties的setExpiration方法在發(fā)送時(shí)動(dòng)態(tài)指定。關(guān)于消息級(jí) TTL 有一個(gè)注意點(diǎn)如果同時(shí)設(shè)置了隊(duì)列級(jí) TTL 和消息級(jí) TTL取兩者中較小的值。單條消息過期后并不會(huì)立即變成死信而是要等它排到隊(duì)列頭部時(shí)才會(huì)被判定所以如果你對(duì)延遲精確到秒最好直接建獨(dú)立隊(duì)列不要依賴隊(duì)列里堆多個(gè)不同消息級(jí) TTL 的消息。延遲消息還有更簡單的做法安裝 RabbitMQ 官方延遲消息插件 rabbitmq_delayed_message_exchange但考慮到插件在集群環(huán)境部署的一致性要求團(tuán)隊(duì)需要先聲明是否允許依賴額外插件基于當(dāng)前團(tuán)隊(duì)情況我用 TTL 死信方案零插件、邏輯清晰、可控性強(qiáng)適用于大多數(shù)訂單超時(shí)場景。4.2 秒殺削峰場景限流消費(fèi)與庫存防超賣把 MQ 作為秒殺系統(tǒng)的入口是非常典型的削峰填谷實(shí)踐。用戶點(diǎn)秒殺按鈕后請(qǐng)求直接被網(wǎng)關(guān)丟進(jìn) MQ后端秒殺消費(fèi)者按照自己處理能力慢慢消費(fèi)把瞬時(shí)高并發(fā)轉(zhuǎn)化為均勻的數(shù)據(jù)庫寫入流。這里最核心的一個(gè)坑是庫存防超賣。如果消費(fèi)者每收到一個(gè)秒殺請(qǐng)求就去數(shù)據(jù)庫查一下庫存發(fā)現(xiàn)大于零然后執(zhí)行UPDATE stock SET count count - 1 WHERE goods_id ?。你覺得自己寫對(duì)了但在高并發(fā)下多個(gè)消費(fèi)者線程同時(shí)讀到剩余庫存為 1然后同時(shí)執(zhí)行 UPDATE庫存會(huì)變成負(fù)數(shù)。解決方案是加條件更新讓數(shù)據(jù)庫自己去保證原子性UPDATE stock SET count count - 1 WHERE goods_id #{goodsId} AND count 0;判斷影響行數(shù)如果影響行數(shù)為 1說明扣減成功為 0說明庫存不足或已賣完該請(qǐng)求直接當(dāng)失敗處理。這是最簡單、最不容易出錯(cuò)的高并發(fā)庫存扣減方案比 Java 代碼里加鎖或分布式鎖都可靠得多。Redis 預(yù)扣減當(dāng)然更快但如果第一版還沒引入 Redis用這個(gè) SQL 方案就足夠穩(wěn)妥。MQ 在秒殺場景還有一層重要用途錯(cuò)峰落庫。秒殺請(qǐng)求先寫訂單表創(chuàng)建一條狀態(tài)為待支付的訂單這個(gè)過程是高頻寫支付回調(diào)再更新訂單狀態(tài)為已支付這個(gè)過程是低頻寫。直接把兩層都做成 MQ 異步處理數(shù)據(jù)庫的并發(fā)寫入壓力會(huì)被攤平到整個(gè)秒殺時(shí)段而不是集中在開搶后那幾秒業(yè)務(wù)才能穩(wěn)定運(yùn)行。4.3 消息不丟失從交換機(jī)到隊(duì)列再到消費(fèi)者的三段保障面試?yán)镒畛柕?RabbbitMQ 問題就是怎么保證消息不丟失。這個(gè)問題分三段回答缺一段都不完整。第一段生產(chǎn)者到交換機(jī)。默認(rèn)情況下生產(chǎn)者用convertAndSend發(fā)消息如果發(fā)到了一個(gè)不存在的交換機(jī)消息直接丟失生產(chǎn)者毫無感知。開啟publisher-confirm-type: correlated后RabbitMQ 會(huì)回調(diào)你的 ConfirmCallback告訴你消息有沒有被交換機(jī)接收。需要注意ConfirmCallback 只管到交換機(jī)這一段消息進(jìn)了交換機(jī)但沒找到隊(duì)列它不會(huì)觸發(fā)失敗回調(diào)。第二段交換機(jī)到隊(duì)列。路由器把你的消息匹配到隊(duì)列如果沒匹配上任何隊(duì)列RabbitMQ 默認(rèn)直接丟棄。開啟publisher-returns: true后匹配不到隊(duì)列的消息會(huì)被退回生產(chǎn)者你的 ReturnCallback 會(huì)收到一條 returnedMessage里面能拿到 exchange、routingKey 和消息體適合做告警和重發(fā)。第三段隊(duì)列到消費(fèi)者。隊(duì)列收到消息后如果消費(fèi)者處理完就 ACK 了消息就安全了。但假如消費(fèi)者在 ACK 之前崩潰消息會(huì)重新回到隊(duì)列等待其他消費(fèi)者處理。這種重投的語義就要靠前面說的冪等設(shè)計(jì)來兜底。這三段保障組合起來就是我們常說的消息不丟失的完整鏈路。在實(shí)際工程里我建議消息生產(chǎn)方和消費(fèi)方都以數(shù)據(jù)庫一張消息記錄表作為最終對(duì)賬依據(jù)發(fā)送時(shí)先本地落一條 status PENDING 的記錄收到交換機(jī)確認(rèn)回調(diào)后更新為 SENT然后有個(gè)定時(shí)任務(wù)掃描超過 1 分鐘仍未 SENT 的記錄補(bǔ)發(fā)消費(fèi)方也記錄一條處理狀態(tài)在 T1 或者實(shí)時(shí)對(duì)賬任務(wù)中核對(duì)兩邊的狀態(tài)發(fā)現(xiàn)不一致就走補(bǔ)償處理。這套機(jī)制被稱為對(duì)賬閉環(huán)有了它中間件層面即使出點(diǎn)小問題業(yè)務(wù)數(shù)據(jù)也能最終一致。5. 高頻問題排查實(shí)錄安裝失敗、消費(fèi)阻塞、消息積壓5.1 安裝與啟動(dòng)失敗問題速查我在實(shí)際部署 RabbitMQ 時(shí)遇到的啟動(dòng)失敗問題歸納下來就這么幾類先對(duì)照排查現(xiàn)象常見原因解決辦法Windows 啟動(dòng)服務(wù)后端口 5672 無響應(yīng)Erlang 版本與 RabbitMQ 不兼容對(duì)照官方版本兼容表重新安裝匹配版本Docker 啟動(dòng)后 15672 無法訪問鏡像沒帶 management 插件或端口映射遺漏換用 rabbitmq:3.12-management 鏡像檢查 -p 參數(shù)啟動(dòng)服務(wù)提示 unable to connect to epmdErlang 節(jié)點(diǎn)名或主機(jī)名解析異常檢查 hosts 文件把主機(jī)名映射到 127.0.0.1重新執(zhí)行 rabbitmq-server -detached集群模式下節(jié)點(diǎn)間無法通信cookie 不一致全集群同步 .erlang.cookie 文件并統(tǒng)一權(quán)限Windows 上的坑我最想說的是環(huán)境變量。有一次啟動(dòng)服務(wù)后后臺(tái)日志一直報(bào) VM 啟動(dòng)失敗 的錯(cuò)誤看了半天才發(fā)現(xiàn)是系統(tǒng)環(huán)境變量ERLANG_HOME沒有配置RabbitMQ 服務(wù)管理器拿著空路徑去找 Erlang 運(yùn)行時(shí)自然起不來。另外 Windows 下千萬別裝到帶中文和空格的路徑下也會(huì)遇到莫名的啟動(dòng)問題。5.2 消費(fèi)速度慢與消息積壓排查線上最容易出的一類問題就是管理后臺(tái)看著隊(duì)列瘋狂堆積消息消費(fèi)者服務(wù) CPU 也不高但消息就是消費(fèi)不過來。這塊請(qǐng)求量一大積壓就肉眼可見地增加。先從最常用的手段排查打開 RabbitMQ 管理頁面進(jìn)入 Queue 頁面注意幾個(gè)指標(biāo)Ready是等待消費(fèi)的消息數(shù)量Unacked是已經(jīng)推給消費(fèi)者但還沒確認(rèn)的數(shù)量Consumer count是當(dāng)前消費(fèi)者數(shù)量。如果Unacked很大說明消息已經(jīng)推給消費(fèi)者了但消費(fèi)者程序處理不過來或卡在某個(gè)調(diào)用上如果Unacked很小但Ready很大說明消費(fèi)者拉的速度太慢要考慮 prefetch 和線程數(shù)。程序?qū)用嫦瓤聪M(fèi)者方法有沒有慢調(diào)用。比如扣庫存要調(diào)數(shù)據(jù)庫數(shù)據(jù)庫一條 SQL 慢查詢了 2 秒那么消費(fèi)者線程就被占住 2 秒一分鐘只能處理 30 條隊(duì)列怎么可能不積壓。我一般習(xí)慣先看日志里消費(fèi)者處理單條消息的平均耗時(shí)再查數(shù)據(jù)庫慢 SQL 日志基本能找到問題所在。如果單條消息處理耗時(shí)正常但依然積壓可以調(diào)大concurrency參數(shù)。RabbitListener注解上可以直接指定RabbitListener(queues MqQueueConfig.ORDER_QUEUE, concurrency 10-20)這個(gè)10-20表示初始 10 個(gè)消費(fèi)者線程最大 20 個(gè)根據(jù)隊(duì)列負(fù)載自動(dòng)擴(kuò)容。要注意并發(fā)線程數(shù)不能無限大它受限于數(shù)據(jù)庫連接池大小如果連接池只有 20 個(gè)連接你把消費(fèi)者并發(fā)開到 100數(shù)據(jù)庫連接會(huì)被耗盡連帶其他業(yè)務(wù)接口一起雪崩。我一個(gè)項(xiàng)目里就吃過這個(gè)虧消息積壓太猛我把并發(fā)開到了 50結(jié)果數(shù)據(jù)庫連接池被消費(fèi)者線程全部占滿整個(gè)訂單服務(wù)對(duì)外接口超時(shí)事故級(jí)別從 P2 直接升級(jí)到 P0。后面我定了一條規(guī)矩消費(fèi)者并發(fā)線程數(shù) 數(shù)據(jù)庫連接池最大值的一半留一半給常規(guī) HTTP 業(yè)務(wù)從此再?zèng)]因?yàn)?MQ 消費(fèi)把連接池打爆過。如果消息積壓已經(jīng)非常嚴(yán)重比如積壓了幾百萬條而且這些消息不是最新業(yè)務(wù)產(chǎn)生的我建議用重置隊(duì)列的方式把已有的積壓消息全部清空或者通過死信方式丟棄讓消費(fèi)者只處理新消息避免消費(fèi)者線程長時(shí)間處理舊數(shù)據(jù)而無法及時(shí)響應(yīng)新請(qǐng)求。這在業(yè)務(wù)上要看場景不能一概而論但比死磕把積壓消息全消費(fèi)完往往更符合業(yè)務(wù)連續(xù)性。5.3 手動(dòng)確認(rèn)模式下最容易犯的三個(gè)錯(cuò)誤很多團(tuán)隊(duì)從 auto 模式切到 manual 模式后反而出了更多問題。這里說三個(gè)我見過最多的。第一個(gè)錯(cuò)誤消費(fèi)者方法里鎖沒釋放就拋異常。手動(dòng)確認(rèn)模式下如果你在處理消息時(shí)拿到了分布式鎖然后業(yè)務(wù)邏輯出現(xiàn)異常走了 basicNack但鎖的 finally 釋放代碼沒寫對(duì)鎖就一直被別人持有會(huì)導(dǎo)致后續(xù)所有同類消息都被卡住。處理機(jī)制是始終用 try-finally 包裹鎖釋放邏輯。第二個(gè)錯(cuò)誤對(duì)重試機(jī)制理解不深導(dǎo)致重復(fù)執(zhí)行。配置了 retry 之后Spring 會(huì)在消費(fèi)者方法拋異常時(shí)自動(dòng)重試但同一個(gè)消息的方法會(huì)被執(zhí)行多次。如果你在方法內(nèi)先扣了庫存再拋異常Spring 又會(huì)帶著同一筆訂單重新進(jìn)入方法再扣一次庫存庫存就超賣了。所以只要消費(fèi)者的業(yè)務(wù)操作涉及寫操作就必須先做冪等判斷再執(zhí)行真正的業(yè)務(wù)邏輯。這套先冪等后業(yè)務(wù)的順序是鐵律任何情況下都不能換。第三個(gè)錯(cuò)誤是把 basicNack 的 requeue 參數(shù)一直設(shè)為 true重試兩次后仍失敗繼續(xù)重回隊(duì)列直接死循環(huán)。RabbitMQ 不會(huì)因?yàn)橥粭l消息重投了 N 次就自動(dòng)放棄它如果你在代碼里沒有失敗次數(shù)判斷它會(huì)無限循環(huán)下去。我在消息屬性里加了x-death頭判斷它在消息變成死信時(shí)記錄下被投遞的次數(shù)消費(fèi)的時(shí)候讀取這個(gè)頭一旦大于 3 次直接丟棄不再重投。你可以直接試這個(gè)方案// 在消費(fèi)者中獲取重試次數(shù) Object deathHeader message.getMessageProperties().getHeaders().get(x-death); int retryCount 0; if (deathHeader instanceof List !((List?) deathHeader).isEmpty()) { Map?, ? death (Map?, ?) ((List?) deathHeader).get(0); retryCount ((Number) death.get(count)).intValue(); } if (retryCount 3) { channel.basicAck(deliveryTag, false); log.error(消息重試超過3次丟棄消息{}, body); }5.4 SpringCloud 環(huán)境下的特別注意事項(xiàng)在 SpringCloud 框架下集成 RabbitMQ有一個(gè)配置問題容易被忽略如果項(xiàng)目同時(shí)引入了 SpringCloud Stream那它默認(rèn)會(huì)占用 RabbitMQ 的連接并創(chuàng)建名為binder的交換機(jī)。如果你同時(shí)又用原生RabbitTemplate發(fā)消息會(huì)有兩套連接。配置稍微給錯(cuò)一點(diǎn)就會(huì)出現(xiàn)消息發(fā)到binder交換機(jī)而自己的隊(duì)列收不到的情況。我的經(jīng)驗(yàn)是一個(gè)微服務(wù)內(nèi)要么統(tǒng)一用 SpringCloud Stream 的 StreamListener要么統(tǒng)一用原生 RabbitTemplate RabbitListener不要混用否則排查問題時(shí)要同時(shí)看兩套配置邏輯心智負(fù)擔(dān)翻倍。還有 Nacos 注冊(cè)中心配合使用時(shí)的注意事項(xiàng)。SpringCloud 微服務(wù)通常用 Nacos 做配置中心RabbitMQ 的連接參數(shù)往往也放在 Nacos 配置中動(dòng)態(tài)刷新。如果你把spring.rabbitmq.host、spring.rabbitmq.username等配置寫在 Nacos 里并且開啟了配置動(dòng)態(tài)刷新那么修改密碼后部分情況下需要重建 ConnectionFactory 的實(shí)例否則刷新不生效。我踩過一次配置改了不生效的坑后來排查到是 ConnectionFactory 在應(yīng)用啟動(dòng)時(shí)已經(jīng)緩存了舊連接據(jù)說要顯式調(diào)用 resetConnection于是干脆把 MQ 連接參數(shù)單獨(dú)拆到一個(gè)獨(dú)立的配置文件不經(jīng) Nacos 動(dòng)態(tài)刷新修改時(shí)需要重啟消費(fèi)者服務(wù)這樣反而更可控。最后說一個(gè)比較玄但很實(shí)用的事RabbitMQ 的客戶端連接是有心跳機(jī)制的默認(rèn)心跳時(shí)間是 60 秒。如果你的微服務(wù)部署在不穩(wěn)定的網(wǎng)絡(luò)上經(jīng)常出現(xiàn)消費(fèi)者斷開連接但服務(wù)進(jìn)程也沒有報(bào)錯(cuò)的現(xiàn)象把配置稍微調(diào)一下spring: rabbitmq: requested-heartbeat: 30 connection-timeout: 10000心跳時(shí)間短一點(diǎn)網(wǎng)絡(luò)異常能更快暴露并被客戶端感知到觸發(fā)重連比等到 60 秒超時(shí)再去恢復(fù)業(yè)務(wù)影響要小很多。6. 從入門到落地我的總結(jié)與擴(kuò)展經(jīng)驗(yàn)如果你今天剛接觸 RabbitMQ我的建議是別急著把各種高級(jí)特性堆上去先把一個(gè)最簡單的一對(duì)一隊(duì)列完整跑通生產(chǎn)者發(fā)送一條消息、消費(fèi)者手動(dòng) ACK、關(guān)閉服務(wù)后重啟驗(yàn)證消息還在。然后再一步步加上交換機(jī)路由、確認(rèn)回調(diào)、死信隊(duì)列。這個(gè)遞進(jìn)過程比我見過的任何教程都能幫你建立對(duì) MQ 的整體感知。接下來說幾個(gè)我在真實(shí)項(xiàng)目里積累的細(xì)節(jié)經(jīng)驗(yàn)這些不太容易在官方文檔里看到但很可能會(huì)在關(guān)鍵時(shí)刻幫你少踩一次坑。第一個(gè)經(jīng)驗(yàn)隊(duì)列聲明使用durable但持久化到磁盤的消息存在刷盤延遲。RabbitMQ 默認(rèn)對(duì)持久化消息是異步刷盤極端情況下機(jī)器掉電確實(shí)可能丟極少量數(shù)據(jù)。對(duì)業(yè)務(wù)消息嚴(yán)格丟不起的場景要么對(duì)消息落庫做補(bǔ)償要么用 Lazy Queue。RabbitMQ 3.12 之后新隊(duì)列默認(rèn)就是 Lazy Queue消息直接走磁盤犧牲一點(diǎn)吞吐?lián)Q取接近零丟失的保障對(duì)訂單、支付這類業(yè)務(wù)值得。第二個(gè)經(jīng)驗(yàn)微服務(wù)拆得越細(xì)別讓 MQ 隊(duì)列跟著拆得越碎。我看到有些團(tuán)隊(duì)為了解耦一個(gè)業(yè)務(wù)事件建一個(gè)交換機(jī)、一個(gè)隊(duì)列結(jié)果沒到一個(gè)月控制臺(tái)里幾十個(gè)隊(duì)列沒人說得清誰在用。我的原則是按業(yè)務(wù)域劃分隊(duì)列不按具體方法劃分。訂單域一個(gè)交換機(jī)路由鍵區(qū)分 create、cancel、pay 等動(dòng)作消費(fèi)方各取所需。后續(xù)擴(kuò)展新消費(fèi)者綁定同一個(gè)交換機(jī)即可不用動(dòng)上游代碼。第三個(gè)經(jīng)驗(yàn)死信隊(duì)列不是萬能的垃圾桶。死信隊(duì)列里堆積的消息要設(shè)置監(jiān)控告警。一旦死信隊(duì)列持續(xù)增長說明消費(fèi)者處理業(yè)務(wù)的大量異常未得到解決。我習(xí)慣把死信隊(duì)列的消費(fèi)邏輯做成一個(gè)補(bǔ)償網(wǎng)關(guān)接到消息后先查消息里的業(yè)務(wù)主鍵然后從業(yè)務(wù)庫里查當(dāng)前狀態(tài)能修復(fù)就調(diào)用補(bǔ)償接口確實(shí)無法處理的再落到日志表里人工介入。如果死信隊(duì)列本身沒人管那它最終就變成了一個(gè)消息垃圾桶既占內(nèi)存又掩蓋問題真正的異常被靜靜埋在里面。還有一點(diǎn)是關(guān)于團(tuán)隊(duì)協(xié)作的。MQ 是典型的跨服務(wù)、跨團(tuán)隊(duì)基礎(chǔ)設(shè)施比接口契約更講究規(guī)范。生產(chǎn)者改了路由鍵消費(fèi)者隊(duì)列的綁定不做相應(yīng)修改消息就會(huì)進(jìn)死信。我們項(xiàng)目的做法是在 Git 倉庫里單獨(dú)維護(hù)一份 MQ 消息規(guī)范文檔用表格列出每個(gè)交換機(jī)、每個(gè)路由鍵、每個(gè)隊(duì)列的負(fù)責(zé)人、業(yè)務(wù)含義、消息體格式示例。每次改動(dòng)都需要同步更新這份文檔在代碼評(píng)審上一并檢查。這套流程堅(jiān)持下來之后因?yàn)楦南⒏袷綄?dǎo)致的線上問題驟減。最后想說的是RabbitMQ 雖然上手不難但真正要讓它在微服務(wù)架構(gòu)里穩(wěn)定可靠地跑起來需要把可靠性、冪等性、可觀測性這三件事貫穿始終。消息能發(fā)出去只是工程的下限消息在任何情況下都能不丟不重不積壓才是工程的上限。希望這篇實(shí)踐總結(jié)能幫你在自己的微服務(wù)項(xiàng)目里更穩(wěn)地用好 RabbitMQ。如果后續(xù)有時(shí)間我會(huì)再寫一篇從消費(fèi)端到生產(chǎn)端的鏈路追蹤實(shí)踐講講在消息投遞的全鏈路中如何用 TraceId 串起調(diào)用鏈定位到底是哪一環(huán)拖慢了整個(gè)流程。 SpringCloud集成RabbitMQ實(shí)戰(zhàn)：微服務(wù)異步解耦與消息可靠性保障
從實(shí)際項(xiàng)目里摸爬滾打過來的人對(duì)微服務(wù)之間的通信問題應(yīng)該都深有體會(huì)。一個(gè)電商訂單下來要通知庫存服務(wù)扣庫存、通知積分服務(wù)加積分、通知短信服務(wù)發(fā)通知如果全部用 Feign 同步調(diào)用一個(gè)服務(wù)抖動(dòng)整條鏈路就卡死數(shù)據(jù)庫連接池一滿全站跟著雪崩。我在項(xiàng)目里引入 SpringCloud 集成 RabbitMQ 之后這套問題才算真正從根上解決了。這篇就把微服務(wù)場景下使用 RabbitMQ 的完整實(shí)踐過程講清楚從協(xié)議原理、消息可靠性保障、死信隊(duì)列、手動(dòng)確認(rèn)到和 RocketMQ、Kafka 的選型對(duì)比以及我實(shí)際部署和排查問題時(shí)踩過的坑一次性說透。1. 微服務(wù)為什么需要 RabbitMQ核心場景與選型思考1.1 什么場景一定會(huì)用到 RabbitMQ很多新手看完教程會(huì)問我寫個(gè) CRUD 接口直接 HTTP 調(diào)用不就完了為什么非要多加一個(gè)中間件這個(gè)疑問很正常但當(dāng)你面臨下面幾類問題的時(shí)候單純靠同步接口是無解的。第一類是異步解耦。用戶下單后核心鏈路是扣庫存、生成訂單、返回支付鏈接這些必須在幾百毫秒內(nèi)完成。但下單后還要發(fā)短信、發(fā) APP 推送、更新用戶積分、給推薦系統(tǒng)發(fā)行為數(shù)據(jù)這些操作每增加一個(gè)同步調(diào)用接口耗時(shí)就增加幾百毫秒而且積分服務(wù)掛了訂單服務(wù)也得跟著回滾這不合理。把短信推送、積分更新等操作丟進(jìn) MQ訂單接口只需要等隊(duì)列寫入成功耗時(shí)天然降下來下游服務(wù)慢一點(diǎn)甚至臨時(shí)宕機(jī)都不影響主鏈路。第二類是流量削峰。秒殺場景下瞬時(shí) QPS 可能到上萬但數(shù)據(jù)庫只能扛幾百的并發(fā)寫。直接在業(yè)務(wù)代碼里寫數(shù)據(jù)庫瞬間就被打爆。用 MQ 擋在前面請(qǐng)求先全部塞進(jìn)隊(duì)列后端訂單消費(fèi)服務(wù)按自己數(shù)據(jù)庫能承受的速度比如每秒鐘處理 500 條慢慢消費(fèi)削峰填谷的效果立竿見影。這里有一個(gè)關(guān)鍵的定量經(jīng)驗(yàn)我會(huì)先壓測出數(shù)據(jù)庫的寫 TPS 上限再把消費(fèi)者的 prefetch 值設(shè)置成這個(gè) TPS 的一半留一半余量給其他業(yè)務(wù)查詢實(shí)測下來穩(wěn)定得多。第三類是數(shù)據(jù)廣播。微服務(wù)架構(gòu)里一個(gè)用戶下單事件訂單服務(wù)要通知審計(jì)服務(wù)、風(fēng)控服務(wù)、搜索服務(wù)、大數(shù)據(jù)平臺(tái)。如果用 Feign 逐個(gè)調(diào)用每加一個(gè)下游服務(wù)訂單服務(wù)代碼就要改一版。換成 MQ 的 Fanout 交換機(jī)訂單服務(wù)只管往交換機(jī)發(fā)一條消息所有綁定了這個(gè)交換機(jī)的隊(duì)列都能收到自己的副本下游服務(wù)增減完全不影響上游代碼。這種發(fā)布訂閱模型在架構(gòu)演進(jìn)時(shí)價(jià)值極大。一句話講清楚 RabbitMQ 的定位上游只負(fù)責(zé)把消息可靠地發(fā)出去下游按自己的節(jié)奏和需求來消費(fèi)雙方通過隊(duì)列解耦誰出問題都不影響對(duì)方。1.2 選型對(duì)比RabbitMQ 和 RocketMQ、Kafka 怎么選關(guān)于 RabbitMQ、RocketMQ 和 Kafka 的區(qū)別網(wǎng)上說法很多我結(jié)合自己用過的實(shí)際場景給出一個(gè)比較務(wù)實(shí)的看法。首先明確一個(gè)點(diǎn)沒有任何一個(gè) MQ 是絕對(duì)萬能的。先看 RabbitMQ。它是 Erlang 寫的對(duì) AMQP 協(xié)議支持非常完備路由規(guī)則在三種 MQ 里最靈活。社區(qū)認(rèn)知中它的吞吐量上限通常低于 Kafka但單機(jī)幾萬 QPS 對(duì)絕大多數(shù)業(yè)務(wù)系統(tǒng)完全夠用。而且它有一套很成熟的管理控制臺(tái)消息追蹤、隊(duì)列堆積、連接數(shù)監(jiān)控運(yùn)維成本低到驚人中小團(tuán)隊(duì)首選基本不用養(yǎng)專門的 MQ 運(yùn)維。RocketMQ 是阿里開源給 Java 生態(tài)使用的消息隊(duì)列特點(diǎn)是有事務(wù)消息可以保證本地事務(wù)和發(fā)消息的原子性這在支付、轉(zhuǎn)賬場景里很關(guān)鍵。但帶來的問題是部署運(yùn)維重NameServer、Broker、主從同步、Console 一套下來比 RabbitMQ 復(fù)雜不少。Kafka 的定位是日志采集和流處理吞吐量確實(shí)恐怖每秒幾十萬條毫秒級(jí)延遲。但 Kafka 本身不擅長復(fù)雜路由topic 設(shè)計(jì)簡單而且消費(fèi)端要自己維護(hù) offset對(duì)業(yè)務(wù)開發(fā)者來說心智負(fù)擔(dān)明顯更重。數(shù)據(jù)管道、用戶行為日志這種場景用 Kafka 沒問題拿它做訂單業(yè)務(wù)就有點(diǎn)高射炮打蚊子了。在我參與的項(xiàng)目里選型依據(jù)很簡單業(yè)務(wù)消息量日幾百萬級(jí)以下、看重路由靈活性和開發(fā)效率的用 RabbitMQ涉及嚴(yán)格分布式事務(wù)、必須事務(wù)消息兜底的用 RocketMQ海量日志、指標(biāo)數(shù)據(jù)采集秒級(jí)吞吐要求嚇人的用 Kafka。各干各最擅長的別混用。2. 集成前的環(huán)境準(zhǔn)備安裝部署與核心概念2.1 Windows 和 Docker 環(huán)境下的 RabbitMQ 安裝我最早是直接在 Windows 上裝 RabbitMQ 的。這里必須說新手第一次裝十個(gè)有八個(gè)會(huì)卡在 Erlang 版本號(hào)和 RabbitMQ 版本不匹配上。RabbitMQ 對(duì) Erlang 版本有嚴(yán)格對(duì)應(yīng)關(guān)系看一眼官方版本對(duì)照表再下載能避開一大半的坑。裝的時(shí)候先把 Erlang 裝了配好 ERLANG_HOME 環(huán)境變量再裝 RabbitMQ最后裝 rabbitmq_management 插件。命令行窗口用管理員身份運(yùn)行rabbitmq-plugins enable rabbitmq_management裝完訪問http://localhost:15672默認(rèn)賬號(hào) guest / guest 就能進(jìn)管理界面了。但如果你是在云服務(wù)器上部署guest 默認(rèn)只能 localhost 登錄遠(yuǎn)程訪問必須另建賬號(hào)并配置 vhost 權(quán)限。后來圖省事我直接改用 Docker 了是真省心。一條命令依賴全給你裝好docker run -d --name rabbitmq \ -p 5672:5672 -p 15672:15672 \ -e RABBITMQ_DEFAULT_USERadmin \ -e RABBITMQ_DEFAULT_PASSadmin123 \ rabbitmq:3.12-management注意鏡像 tag 一定要帶 management 后綴否則起來的鏡像沒有管理界面插件。我踩過一次坑用的 rabbitmq:3.12 不帶后綴容器起來了 5672 端口通15672 死活不回應(yīng)后來才發(fā)現(xiàn)是鏡像本身沒帶控制臺(tái)插件。2.2 核心概念掃盲交換機(jī)、隊(duì)列、路由鍵的關(guān)系不把 RabbitMQ 的消息模型搞清楚后面寫代碼就是抄模板出了問題不會(huì)排查。這里用生活里的事打比方來理解。想象一個(gè)快遞中轉(zhuǎn)站。生產(chǎn)者就是寄快遞的商家交換機(jī)Exchange就是中轉(zhuǎn)站的調(diào)度臺(tái)隊(duì)列Queue就是快遞員手上的配送單消費(fèi)者就是收快遞的人。商家寄件時(shí)只需要把包裹交給調(diào)度臺(tái)并說清楚目的地調(diào)度臺(tái)根據(jù)他寫的地址把包裹分配到對(duì)應(yīng)的配送單上快遞員按配送單派送收件人接收。這里商家說的目的地在 RabbitMQ 里叫路由鍵Routing Key調(diào)度臺(tái)按什么規(guī)則分配包裹取決于交換機(jī)類型。有三種常用的交換機(jī)Direct 交換機(jī)完全匹配路由鍵精確投遞一對(duì)一的場景最常用Topic 交換機(jī)按通配符模糊匹配比如order.*能匹配order.create和order.cancel適合需要按消息類型分類消費(fèi)的場景Fanout 交換機(jī)不看路由鍵直接廣播給所有綁定的隊(duì)列就像廣播電臺(tái)所有收音機(jī)都能收到。還有一個(gè) Headers 交換機(jī)按消息頭的 key-value 匹配實(shí)際業(yè)務(wù)里用得少我基本不碰。你還要理解Binding這個(gè)概念就是交換機(jī)和隊(duì)列之間建立的那條配送規(guī)則。生產(chǎn)者在綁定交換機(jī)的時(shí)候要同時(shí)指定隊(duì)列名稱和路由鍵。Bean public Binding binding() { return BindingBuilder .bind(orderQueue()) .to(orderExchange()) .with(order.create); }這段話的含義是消息發(fā)到orderExchange這個(gè)交換機(jī)時(shí)如果路由鍵是order.create就進(jìn)入orderQueue這個(gè)隊(duì)列。很多新手把 Queue 上的 name 和 Binding 的 with 混為一談其實(shí)前者是隊(duì)列名后者是路由鍵兩個(gè)概念別搞混了。3. SpringCloud 項(xiàng)目接入從依賴配置到第一個(gè)可靠消息3.1 依賴引入與核心配置項(xiàng)解析在 SpringCloud 項(xiàng)目里集成 RabbitMQ用的是 Spring Boot 的 starter 封裝。基礎(chǔ)依賴就一個(gè)dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency如果項(xiàng)目用的是 SpringCloud 版本管理 BOM不需要額外指定版本號(hào)它會(huì)自動(dòng)匹配兼容版本。這里我說一個(gè)重要經(jīng)驗(yàn)永遠(yuǎn)不要手工指定 spring-boot-starter-amqp 的版本號(hào)讓 SpringCloud BOM 統(tǒng)一管理。我見過一次事故同事手動(dòng)指定版本結(jié)果和當(dāng)前 SpringCloud 版本不兼容消息發(fā)送時(shí) SerializationException 滿天飛查了半天才發(fā)現(xiàn)是版本沖突。application.yml 里核心配置項(xiàng)是這些spring: rabbitmq: host: 127.0.0.1 port: 5672 username: admin password: admin123 virtual-host: / publisher-confirm-type: correlated publisher-returns: true listener: simple: acknowledge-mode: manual prefetch: 50 retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2 default-requeue-rejected: false逐項(xiàng)講一下這些配置背后的邏輯這是面試官最愛問的點(diǎn)也是實(shí)際線上出問題時(shí)要調(diào)的參數(shù)。publisher-confirm-type: correlated是開啟生產(chǎn)者確認(rèn)發(fā)送端能確知消息是否真正到達(dá)交換機(jī)。消息到達(dá)交換機(jī)后RabbitMQ 會(huì)回調(diào)一個(gè) ConfirmCallback 告訴你發(fā)送成功還是失敗。publisher-returns: true是開啟消息未投遞到隊(duì)列時(shí)的退回機(jī)制也就是消息到了交換機(jī)但沒匹配到任何隊(duì)列RabbitMQ 會(huì)把消息退回給生產(chǎn)者并觸發(fā) ReturnCallback。這兩條加一起才保證從生產(chǎn)者到隊(duì)列這一段不丟消息。acknowledge-mode: manual是手動(dòng) ACK這是線上生產(chǎn)環(huán)境必須的設(shè)置。默認(rèn)的 auto 模式在消費(fèi)者拋出異常時(shí)會(huì)自動(dòng)確認(rèn)消息消息就丟了。手動(dòng)模式把消息確認(rèn)的時(shí)機(jī)交給代碼控制消費(fèi)者處理成功后手動(dòng)調(diào)用 basicAck 告訴 RabbitMQ 可以刪除消息處理失敗時(shí)調(diào)用 basicNack告訴 RabbitMQ 這條消息我沒處理好你看著辦。prefetch是消費(fèi)者每次從隊(duì)列拉取多少條消息到本地緩存。設(shè)太大會(huì)導(dǎo)致消費(fèi)者內(nèi)存被占滿而且某條消息處理時(shí)間過長其他消息一直等在緩存里得不到處理設(shè)太小則吞吐量上不去。我常用的做法是結(jié)合服務(wù)端的線程池大小和業(yè)務(wù)處理耗時(shí)來定處理 50ms 以內(nèi)的消息prefetch 設(shè) 50~100 很合理處理 500ms 以上的消息prefetch 建議壓到 10 以下避免本地積壓太多消息導(dǎo)致內(nèi)存抖動(dòng)。3.2 生產(chǎn)者代碼確認(rèn)回調(diào)與 Return 回退配置只是第一步真正落實(shí)到代碼層面生產(chǎn)者的可靠發(fā)送要這樣寫。我先建一個(gè)配置類注冊(cè)交換機(jī)、隊(duì)列和綁定關(guān)系然后再建發(fā)送消息的 Service。先看交換機(jī)、隊(duì)列與綁定關(guān)系的聲明Configuration public class MqQueueConfig { public static final String ORDER_EXCHANGE order.exchange; public static final String ORDER_QUEUE order.queue; public static final String ORDER_ROUTING_KEY order.create; Bean public DirectExchange orderExchange() { return new DirectExchange(ORDER_EXCHANGE, true, false); } Bean public Queue orderQueue() { return QueueBuilder.durable(ORDER_QUEUE).build(); } Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()) .to(orderExchange()) .with(ORDER_ROUTING_KEY); } }這里要留意的細(xì)節(jié)交換機(jī)、隊(duì)列的 durable 參數(shù)都設(shè)為 true代表持久化。這樣 RabbitMQ 重啟后交換機(jī)和隊(duì)列不會(huì)消失。隊(duì)列聲明的時(shí)候我推薦用QueueBuilder.durable().build()這種鏈?zhǔn)綄懛ê罄m(xù)要加 TTL、死信參數(shù)的時(shí)候直接往這個(gè) Builder 上追加即可可讀性比 new Queue(name, true, false, false) 直觀得多。代碼里最好用常量把交換機(jī)名、隊(duì)列名、路由鍵集中管理別在業(yè)務(wù)方法里隨手寫字符串項(xiàng)目大了你根本找不到誰在消費(fèi)誰的消息。然后是發(fā)送消息的核心代碼。我需要讓 RabbitTemplate 在消息確認(rèn)和消息退回時(shí)都留下日志方便排查Service Slf4j public class OrderMessageProducer { private final RabbitTemplate rabbitTemplate; public OrderMessageProducer(RabbitTemplate rabbitTemplate) { this.rabbitTemplate rabbitTemplate; // 消息到達(dá)交換機(jī)的確認(rèn)回調(diào) rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - { if (ack) { log.info(消息發(fā)送成功correlationId {}, correlationData.getId()); } else { log.error(消息發(fā)送失敗correlationId {}, cause {}, correlationData.getId(), cause); } }); // 消息未投遞到隊(duì)列的退回回調(diào) rabbitTemplate.setReturnsCallback(returned - { log.error(消息被退回exchange {}, routingKey {}, body {}, returned.getExchange(), returned.getRoutingKey(), new String(returned.getMessage().getBody())); }); } public void sendOrderMessage(String orderJson) { CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString()); rabbitTemplate.convertAndSend( MqQueueConfig.ORDER_EXCHANGE, MqQueueConfig.ORDER_ROUTING_KEY, orderJson, correlationData ); } }第二個(gè)方法第一步setConfirmCallback通常在構(gòu)造器里執(zhí)行一次就夠了因?yàn)樗侨旨?jí)別的回調(diào)。實(shí)際操作里很多人是在每次 send 方法前回調(diào)設(shè)置一遍也能跑通但沒有必要會(huì)對(duì)后續(xù)排查日志造成一定干擾。注意這里correlationData的 id 我用 UUID 生成目的就是能在回調(diào)日志里精確定位到某一條消息排查線上問題時(shí)特別有用。還有個(gè)細(xì)節(jié)convertAndSend方法發(fā)送字符串的時(shí)候?qū)嶋H走的是 SimpleMessageConverter 默認(rèn)的序列化方式將字符串轉(zhuǎn)成 UTF-8 字節(jié)數(shù)組。如果你直接發(fā)一個(gè) Java 對(duì)象默認(rèn)會(huì)用 JDK 序列化那么消費(fèi)者端也要用 JDK 反序列化類名還要完全一致這對(duì)微服務(wù)之間類共享有很強(qiáng)約束。我自己項(xiàng)目中通常是發(fā)送 JSON 字符串讓每個(gè)微服務(wù)自己反序列化成自己的 DTO避免模塊間強(qiáng)依賴這是微服務(wù)架構(gòu)的基本原則。3.3 消費(fèi)者代碼手動(dòng) ACK 與重試機(jī)制的配合消費(fèi)者這端是坑最多的地方。很多教程里只寫一個(gè)RabbitListener加上RabbitHandler方法執(zhí)行完就算完事了。但真實(shí)生產(chǎn)環(huán)境必須處理消費(fèi)失敗、消息重試、冪等這幾個(gè)問題。先看一個(gè)標(biāo)準(zhǔn)的手動(dòng) ACK 消費(fèi)示例Component Slf4j public class OrderMessageConsumer { RabbitListener(queues MqQueueConfig.ORDER_QUEUE) public void handleOrderCreate(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); String body new String(message.getBody(), StandardCharsets.UTF_8); try { log.info(收到訂單消息{}, body); OrderDTO orderDTO JSON.parseObject(body, OrderDTO.class); // 核心業(yè)務(wù)邏輯扣庫存、更新訂單狀態(tài)、記錄日志 orderService.handleOrderCreated(orderDTO); // 業(yè)務(wù)處理成功手動(dòng)確認(rèn) channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error(訂單消息處理失敗deliveryTag {}, deliveryTag, e); // 確認(rèn)是否已處理過 if (isProcessed(orderDTO.getId())) { channel.basicAck(deliveryTag, false); return; } // 處理失敗不重回隊(duì)列轉(zhuǎn)為死信 channel.basicNack(deliveryTag, false, false); } } }手動(dòng) ACK 的三個(gè)方法要搞清楚basicAck(deliveryTag, false)確認(rèn)成功RabbitMQ 刪除這條消息。第二個(gè)參數(shù) false 表示只確認(rèn)當(dāng)前這條消息不批量。basicNack(deliveryTag, false, true)第三個(gè)參數(shù) true 表示重回隊(duì)列注意這非常危險(xiǎn)如果消息本身有問題回到隊(duì)列會(huì)被再消費(fèi)一次然后再次 Nack形成無限循環(huán)消費(fèi)把 CPU 打滿日志刷屏。basicNack(deliveryTag, false, false)不重回隊(duì)列直接丟棄或進(jìn)死信。生產(chǎn)環(huán)境的重試機(jī)制我推薦用 Spring AMQP 自帶的retry配置而不是在消費(fèi)者代碼里自己寫 for 循環(huán)。配置文件里的 retry 參數(shù)組合起來是這樣的邏輯消費(fèi)者處理拋出異常后Spring 會(huì)按initial-interval開始重試每次重試間隔按multiplier倍數(shù)遞增最大嘗試次數(shù)是max-attempts。三次都失敗之后配合default-requeue-rejected: false這條消息才會(huì)被拒絕并投遞到死信交換機(jī)。所以在消費(fèi)者方法內(nèi)部我只需要拋異常就行重試過程交給框架處理代碼更干凈。還有一個(gè)容易忽略的關(guān)鍵點(diǎn)消費(fèi)者方法的冪等性。RabbitMQ 的投遞語義是至少一次也就是說極端情況下消息雖然被消費(fèi)成功了但 ACK 在網(wǎng)絡(luò)傳輸中丟失RabbitMQ 端認(rèn)為消息沒被消費(fèi)會(huì)重投。消費(fèi)者必須保證同一筆訂單消息被消費(fèi)兩次和一次結(jié)果一致。我常用的做法是在訂單表上建一個(gè)order_event_id唯一索引消費(fèi)前先查詢或嘗試插入一條冪等記錄能插入就繼續(xù)業(yè)務(wù)處理插入沖突說明已經(jīng)處理過直接投遞成功 ACK。這也解釋了為什么代碼里要有isProcessed(orderDTO.getId())這個(gè)判斷沒有這一層的系統(tǒng)是不完整的。4. 微服務(wù)實(shí)戰(zhàn)拆解死信隊(duì)列、延遲消息與削峰實(shí)踐4.1 訂單超時(shí)關(guān)閉場景TTL 與死信隊(duì)列組合實(shí)現(xiàn)延遲消息很多人剛接觸 RabbitMQ 時(shí)都聽說過它可以做延遲消息。但 RabbitMQ 本身并沒有直接提供延遲隊(duì)列插件支持實(shí)現(xiàn)延遲消息的經(jīng)典方案是利用死信隊(duì)列 消息 TTL。先解釋什么叫死信。消息在隊(duì)列里處于以下幾種狀態(tài)時(shí)就變成了死信消息被消費(fèi)者拒絕且不重回隊(duì)列消息 TTL 到期未被消費(fèi)隊(duì)列長度達(dá)到上限導(dǎo)致消息被丟棄。死信不會(huì)自己消失如果隊(duì)列配置了死信交換機(jī)RabbitMQ 會(huì)把死信重新投遞到指定的死信交換機(jī)再由死信交換機(jī)路由到對(duì)應(yīng)的死信隊(duì)列。下面的代碼聲明了一個(gè) 30 秒 TTL 的訂單超時(shí)隊(duì)列并綁定了死信交換機(jī)Configuration public class OrderTimeoutMqConfig { public static final String ORDER_DELAY_EXCHANGE order.delay.exchange; public static final String ORDER_DELAY_QUEUE order.delay.queue; public static final String ORDER_DEAD_EXCHANGE order.dead.exchange; public static final String ORDER_DEAD_QUEUE order.dead.queue; // 延遲隊(duì)列消息在這里等 30 秒 Bean public DirectExchange orderDelayExchange() { return new DirectExchange(ORDER_DELAY_EXCHANGE, true, false); } Bean public Queue orderDelayQueue() { return QueueBuilder.durable(ORDER_DELAY_QUEUE) .ttl(30000) .deadLetterExchange(ORDER_DEAD_EXCHANGE) .deadLetterRoutingKey(order.timeout) .build(); } // 死信交換機(jī)與死信隊(duì)列消息過期后到這里 Bean public DirectExchange orderDeadExchange() { return new DirectExchange(ORDER_DEAD_EXCHANGE, true, false); } Bean public Queue orderDeadQueue() { return QueueBuilder.durable(ORDER_DEAD_QUEUE).build(); } Bean public Binding delayBinding() { return BindingBuilder.bind(orderDelayQueue()) .to(orderDelayExchange()) .with(order.delay); } Bean public Binding deadBinding() { return BindingBuilder.bind(orderDeadQueue()) .to(orderDeadExchange()) .with(order.timeout); } }這里有一個(gè)需要特別提醒的隊(duì)列級(jí)別的 TTL 一旦設(shè)置隊(duì)列里的所有消息共享這個(gè)過期時(shí)間。如果你需要在同一個(gè)隊(duì)列中放不同延遲時(shí)間的消息比如訂單超時(shí)關(guān)閉是 30 秒自動(dòng)確認(rèn)收貨是 7 天就不能用隊(duì)列級(jí) TTL需要單獨(dú)建隊(duì)列。隊(duì)列級(jí) TTL 還有個(gè)大坑一條消息只在隊(duì)列頭部被檢查是否過期如果隊(duì)列頭部那條消息的 TTL 很長后面的消息即使 TTL 更短也只能等頭部消息先被消費(fèi)或過期。所以實(shí)際業(yè)務(wù)里不同延遲時(shí)間務(wù)必拆分到不同隊(duì)列。下單時(shí)訂單服務(wù)把訂單號(hào)發(fā)到order.delay.exchange路由鍵order.delay消息進(jìn)入order.delay.queue等待 30 秒。30 秒后消息變?yōu)樗佬磐哆f到order.dead.exchange再路由到order.dead.queue。此時(shí)專門負(fù)責(zé)關(guān)閉超時(shí)訂單的消費(fèi)者從死信隊(duì)列拿到訂單號(hào)查詢訂單狀態(tài)如果還是未支付則改成已超時(shí)關(guān)閉Component Slf4j public class OrderTimeoutConsumer { RabbitListener(queues OrderTimeoutMqConfig.ORDER_DEAD_QUEUE) public void handleTimeout(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); String orderId new String(message.getBody(), StandardCharsets.UTF_8); try { log.info(收到訂單超時(shí)消息orderId {}, orderId); orderService.closeTimeoutOrder(orderId); channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error(處理訂單超時(shí)消息失敗orderId {}, orderId, e); channel.basicNack(deliveryTag, false, false); } } }如果你想在消息級(jí)別設(shè)置獨(dú)立的 TTL也可以用MessageProperties的setExpiration方法在發(fā)送時(shí)動(dòng)態(tài)指定。關(guān)于消息級(jí) TTL 有一個(gè)注意點(diǎn)如果同時(shí)設(shè)置了隊(duì)列級(jí) TTL 和消息級(jí) TTL取兩者中較小的值。單條消息過期后并不會(huì)立即變成死信而是要等它排到隊(duì)列頭部時(shí)才會(huì)被判定所以如果你對(duì)延遲精確到秒最好直接建獨(dú)立隊(duì)列不要依賴隊(duì)列里堆多個(gè)不同消息級(jí) TTL 的消息。延遲消息還有更簡單的做法安裝 RabbitMQ 官方延遲消息插件 rabbitmq_delayed_message_exchange但考慮到插件在集群環(huán)境部署的一致性要求團(tuán)隊(duì)需要先聲明是否允許依賴額外插件基于當(dāng)前團(tuán)隊(duì)情況我用 TTL 死信方案零插件、邏輯清晰、可控性強(qiáng)適用于大多數(shù)訂單超時(shí)場景。4.2 秒殺削峰場景限流消費(fèi)與庫存防超賣把 MQ 作為秒殺系統(tǒng)的入口是非常典型的削峰填谷實(shí)踐。用戶點(diǎn)秒殺按鈕后請(qǐng)求直接被網(wǎng)關(guān)丟進(jìn) MQ后端秒殺消費(fèi)者按照自己處理能力慢慢消費(fèi)把瞬時(shí)高并發(fā)轉(zhuǎn)化為均勻的數(shù)據(jù)庫寫入流。這里最核心的一個(gè)坑是庫存防超賣。如果消費(fèi)者每收到一個(gè)秒殺請(qǐng)求就去數(shù)據(jù)庫查一下庫存發(fā)現(xiàn)大于零然后執(zhí)行UPDATE stock SET count count - 1 WHERE goods_id ?。你覺得自己寫對(duì)了但在高并發(fā)下多個(gè)消費(fèi)者線程同時(shí)讀到剩余庫存為 1然后同時(shí)執(zhí)行 UPDATE庫存會(huì)變成負(fù)數(shù)。解決方案是加條件更新讓數(shù)據(jù)庫自己去保證原子性UPDATE stock SET count count - 1 WHERE goods_id #{goodsId} AND count 0;判斷影響行數(shù)如果影響行數(shù)為 1說明扣減成功為 0說明庫存不足或已賣完該請(qǐng)求直接當(dāng)失敗處理。這是最簡單、最不容易出錯(cuò)的高并發(fā)庫存扣減方案比 Java 代碼里加鎖或分布式鎖都可靠得多。Redis 預(yù)扣減當(dāng)然更快但如果第一版還沒引入 Redis用這個(gè) SQL 方案就足夠穩(wěn)妥。MQ 在秒殺場景還有一層重要用途錯(cuò)峰落庫。秒殺請(qǐng)求先寫訂單表創(chuàng)建一條狀態(tài)為待支付的訂單這個(gè)過程是高頻寫支付回調(diào)再更新訂單狀態(tài)為已支付這個(gè)過程是低頻寫。直接把兩層都做成 MQ 異步處理數(shù)據(jù)庫的并發(fā)寫入壓力會(huì)被攤平到整個(gè)秒殺時(shí)段而不是集中在開搶后那幾秒業(yè)務(wù)才能穩(wěn)定運(yùn)行。4.3 消息不丟失從交換機(jī)到隊(duì)列再到消費(fèi)者的三段保障面試?yán)镒畛柕?RabbbitMQ 問題就是怎么保證消息不丟失。這個(gè)問題分三段回答缺一段都不完整。第一段生產(chǎn)者到交換機(jī)。默認(rèn)情況下生產(chǎn)者用convertAndSend發(fā)消息如果發(fā)到了一個(gè)不存在的交換機(jī)消息直接丟失生產(chǎn)者毫無感知。開啟publisher-confirm-type: correlated后RabbitMQ 會(huì)回調(diào)你的 ConfirmCallback告訴你消息有沒有被交換機(jī)接收。需要注意ConfirmCallback 只管到交換機(jī)這一段消息進(jìn)了交換機(jī)但沒找到隊(duì)列它不會(huì)觸發(fā)失敗回調(diào)。第二段交換機(jī)到隊(duì)列。路由器把你的消息匹配到隊(duì)列如果沒匹配上任何隊(duì)列RabbitMQ 默認(rèn)直接丟棄。開啟publisher-returns: true后匹配不到隊(duì)列的消息會(huì)被退回生產(chǎn)者你的 ReturnCallback 會(huì)收到一條 returnedMessage里面能拿到 exchange、routingKey 和消息體適合做告警和重發(fā)。第三段隊(duì)列到消費(fèi)者。隊(duì)列收到消息后如果消費(fèi)者處理完就 ACK 了消息就安全了。但假如消費(fèi)者在 ACK 之前崩潰消息會(huì)重新回到隊(duì)列等待其他消費(fèi)者處理。這種重投的語義就要靠前面說的冪等設(shè)計(jì)來兜底。這三段保障組合起來就是我們常說的消息不丟失的完整鏈路。在實(shí)際工程里我建議消息生產(chǎn)方和消費(fèi)方都以數(shù)據(jù)庫一張消息記錄表作為最終對(duì)賬依據(jù)發(fā)送時(shí)先本地落一條 status PENDING 的記錄收到交換機(jī)確認(rèn)回調(diào)后更新為 SENT然后有個(gè)定時(shí)任務(wù)掃描超過 1 分鐘仍未 SENT 的記錄補(bǔ)發(fā)消費(fèi)方也記錄一條處理狀態(tài)在 T1 或者實(shí)時(shí)對(duì)賬任務(wù)中核對(duì)兩邊的狀態(tài)發(fā)現(xiàn)不一致就走補(bǔ)償處理。這套機(jī)制被稱為對(duì)賬閉環(huán)有了它中間件層面即使出點(diǎn)小問題業(yè)務(wù)數(shù)據(jù)也能最終一致。5. 高頻問題排查實(shí)錄安裝失敗、消費(fèi)阻塞、消息積壓5.1 安裝與啟動(dòng)失敗問題速查我在實(shí)際部署 RabbitMQ 時(shí)遇到的啟動(dòng)失敗問題歸納下來就這么幾類先對(duì)照排查現(xiàn)象常見原因解決辦法Windows 啟動(dòng)服務(wù)后端口 5672 無響應(yīng)Erlang 版本與 RabbitMQ 不兼容對(duì)照官方版本兼容表重新安裝匹配版本Docker 啟動(dòng)后 15672 無法訪問鏡像沒帶 management 插件或端口映射遺漏換用 rabbitmq:3.12-management 鏡像檢查 -p 參數(shù)啟動(dòng)服務(wù)提示 unable to connect to epmdErlang 節(jié)點(diǎn)名或主機(jī)名解析異常檢查 hosts 文件把主機(jī)名映射到 127.0.0.1重新執(zhí)行 rabbitmq-server -detached集群模式下節(jié)點(diǎn)間無法通信cookie 不一致全集群同步 .erlang.cookie 文件并統(tǒng)一權(quán)限Windows 上的坑我最想說的是環(huán)境變量。有一次啟動(dòng)服務(wù)后后臺(tái)日志一直報(bào) VM 啟動(dòng)失敗 的錯(cuò)誤看了半天才發(fā)現(xiàn)是系統(tǒng)環(huán)境變量ERLANG_HOME沒有配置RabbitMQ 服務(wù)管理器拿著空路徑去找 Erlang 運(yùn)行時(shí)自然起不來。另外 Windows 下千萬別裝到帶中文和空格的路徑下也會(huì)遇到莫名的啟動(dòng)問題。5.2 消費(fèi)速度慢與消息積壓排查線上最容易出的一類問題就是管理后臺(tái)看著隊(duì)列瘋狂堆積消息消費(fèi)者服務(wù) CPU 也不高但消息就是消費(fèi)不過來。這塊請(qǐng)求量一大積壓就肉眼可見地增加。先從最常用的手段排查打開 RabbitMQ 管理頁面進(jìn)入 Queue 頁面注意幾個(gè)指標(biāo)Ready是等待消費(fèi)的消息數(shù)量Unacked是已經(jīng)推給消費(fèi)者但還沒確認(rèn)的數(shù)量Consumer count是當(dāng)前消費(fèi)者數(shù)量。如果Unacked很大說明消息已經(jīng)推給消費(fèi)者了但消費(fèi)者程序處理不過來或卡在某個(gè)調(diào)用上如果Unacked很小但Ready很大說明消費(fèi)者拉的速度太慢要考慮 prefetch 和線程數(shù)。程序?qū)用嫦瓤聪M(fèi)者方法有沒有慢調(diào)用。比如扣庫存要調(diào)數(shù)據(jù)庫數(shù)據(jù)庫一條 SQL 慢查詢了 2 秒那么消費(fèi)者線程就被占住 2 秒一分鐘只能處理 30 條隊(duì)列怎么可能不積壓。我一般習(xí)慣先看日志里消費(fèi)者處理單條消息的平均耗時(shí)再查數(shù)據(jù)庫慢 SQL 日志基本能找到問題所在。如果單條消息處理耗時(shí)正常但依然積壓可以調(diào)大concurrency參數(shù)。RabbitListener注解上可以直接指定RabbitListener(queues MqQueueConfig.ORDER_QUEUE, concurrency 10-20)這個(gè)10-20表示初始 10 個(gè)消費(fèi)者線程最大 20 個(gè)根據(jù)隊(duì)列負(fù)載自動(dòng)擴(kuò)容。要注意并發(fā)線程數(shù)不能無限大它受限于數(shù)據(jù)庫連接池大小如果連接池只有 20 個(gè)連接你把消費(fèi)者并發(fā)開到 100數(shù)據(jù)庫連接會(huì)被耗盡連帶其他業(yè)務(wù)接口一起雪崩。我一個(gè)項(xiàng)目里就吃過這個(gè)虧消息積壓太猛我把并發(fā)開到了 50結(jié)果數(shù)據(jù)庫連接池被消費(fèi)者線程全部占滿整個(gè)訂單服務(wù)對(duì)外接口超時(shí)事故級(jí)別從 P2 直接升級(jí)到 P0。后面我定了一條規(guī)矩消費(fèi)者并發(fā)線程數(shù) 數(shù)據(jù)庫連接池最大值的一半留一半給常規(guī) HTTP 業(yè)務(wù)從此再?zèng)]因?yàn)?MQ 消費(fèi)把連接池打爆過。如果消息積壓已經(jīng)非常嚴(yán)重比如積壓了幾百萬條而且這些消息不是最新業(yè)務(wù)產(chǎn)生的我建議用重置隊(duì)列的方式把已有的積壓消息全部清空或者通過死信方式丟棄讓消費(fèi)者只處理新消息避免消費(fèi)者線程長時(shí)間處理舊數(shù)據(jù)而無法及時(shí)響應(yīng)新請(qǐng)求。這在業(yè)務(wù)上要看場景不能一概而論但比死磕把積壓消息全消費(fèi)完往往更符合業(yè)務(wù)連續(xù)性。5.3 手動(dòng)確認(rèn)模式下最容易犯的三個(gè)錯(cuò)誤很多團(tuán)隊(duì)從 auto 模式切到 manual 模式后反而出了更多問題。這里說三個(gè)我見過最多的。第一個(gè)錯(cuò)誤消費(fèi)者方法里鎖沒釋放就拋異常。手動(dòng)確認(rèn)模式下如果你在處理消息時(shí)拿到了分布式鎖然后業(yè)務(wù)邏輯出現(xiàn)異常走了 basicNack但鎖的 finally 釋放代碼沒寫對(duì)鎖就一直被別人持有會(huì)導(dǎo)致后續(xù)所有同類消息都被卡住。處理機(jī)制是始終用 try-finally 包裹鎖釋放邏輯。第二個(gè)錯(cuò)誤對(duì)重試機(jī)制理解不深導(dǎo)致重復(fù)執(zhí)行。配置了 retry 之后Spring 會(huì)在消費(fèi)者方法拋異常時(shí)自動(dòng)重試但同一個(gè)消息的方法會(huì)被執(zhí)行多次。如果你在方法內(nèi)先扣了庫存再拋異常Spring 又會(huì)帶著同一筆訂單重新進(jìn)入方法再扣一次庫存庫存就超賣了。所以只要消費(fèi)者的業(yè)務(wù)操作涉及寫操作就必須先做冪等判斷再執(zhí)行真正的業(yè)務(wù)邏輯。這套先冪等后業(yè)務(wù)的順序是鐵律任何情況下都不能換。第三個(gè)錯(cuò)誤是把 basicNack 的 requeue 參數(shù)一直設(shè)為 true重試兩次后仍失敗繼續(xù)重回隊(duì)列直接死循環(huán)。RabbitMQ 不會(huì)因?yàn)橥粭l消息重投了 N 次就自動(dòng)放棄它如果你在代碼里沒有失敗次數(shù)判斷它會(huì)無限循環(huán)下去。我在消息屬性里加了x-death頭判斷它在消息變成死信時(shí)記錄下被投遞的次數(shù)消費(fèi)的時(shí)候讀取這個(gè)頭一旦大于 3 次直接丟棄不再重投。你可以直接試這個(gè)方案// 在消費(fèi)者中獲取重試次數(shù) Object deathHeader message.getMessageProperties().getHeaders().get(x-death); int retryCount 0; if (deathHeader instanceof List !((List?) deathHeader).isEmpty()) { Map?, ? death (Map?, ?) ((List?) deathHeader).get(0); retryCount ((Number) death.get(count)).intValue(); } if (retryCount 3) { channel.basicAck(deliveryTag, false); log.error(消息重試超過3次丟棄消息{}, body); }5.4 SpringCloud 環(huán)境下的特別注意事項(xiàng)在 SpringCloud 框架下集成 RabbitMQ有一個(gè)配置問題容易被忽略如果項(xiàng)目同時(shí)引入了 SpringCloud Stream那它默認(rèn)會(huì)占用 RabbitMQ 的連接并創(chuàng)建名為binder的交換機(jī)。如果你同時(shí)又用原生RabbitTemplate發(fā)消息會(huì)有兩套連接。配置稍微給錯(cuò)一點(diǎn)就會(huì)出現(xiàn)消息發(fā)到binder交換機(jī)而自己的隊(duì)列收不到的情況。我的經(jīng)驗(yàn)是一個(gè)微服務(wù)內(nèi)要么統(tǒng)一用 SpringCloud Stream 的 StreamListener要么統(tǒng)一用原生 RabbitTemplate RabbitListener不要混用否則排查問題時(shí)要同時(shí)看兩套配置邏輯心智負(fù)擔(dān)翻倍。還有 Nacos 注冊(cè)中心配合使用時(shí)的注意事項(xiàng)。SpringCloud 微服務(wù)通常用 Nacos 做配置中心RabbitMQ 的連接參數(shù)往往也放在 Nacos 配置中動(dòng)態(tài)刷新。如果你把spring.rabbitmq.host、spring.rabbitmq.username等配置寫在 Nacos 里并且開啟了配置動(dòng)態(tài)刷新那么修改密碼后部分情況下需要重建 ConnectionFactory 的實(shí)例否則刷新不生效。我踩過一次配置改了不生效的坑后來排查到是 ConnectionFactory 在應(yīng)用啟動(dòng)時(shí)已經(jīng)緩存了舊連接據(jù)說要顯式調(diào)用 resetConnection于是干脆把 MQ 連接參數(shù)單獨(dú)拆到一個(gè)獨(dú)立的配置文件不經(jīng) Nacos 動(dòng)態(tài)刷新修改時(shí)需要重啟消費(fèi)者服務(wù)這樣反而更可控。最后說一個(gè)比較玄但很實(shí)用的事RabbitMQ 的客戶端連接是有心跳機(jī)制的默認(rèn)心跳時(shí)間是 60 秒。如果你的微服務(wù)部署在不穩(wěn)定的網(wǎng)絡(luò)上經(jīng)常出現(xiàn)消費(fèi)者斷開連接但服務(wù)進(jìn)程也沒有報(bào)錯(cuò)的現(xiàn)象把配置稍微調(diào)一下spring: rabbitmq: requested-heartbeat: 30 connection-timeout: 10000心跳時(shí)間短一點(diǎn)網(wǎng)絡(luò)異常能更快暴露并被客戶端感知到觸發(fā)重連比等到 60 秒超時(shí)再去恢復(fù)業(yè)務(wù)影響要小很多。6. 從入門到落地我的總結(jié)與擴(kuò)展經(jīng)驗(yàn)如果你今天剛接觸 RabbitMQ我的建議是別急著把各種高級(jí)特性堆上去先把一個(gè)最簡單的一對(duì)一隊(duì)列完整跑通生產(chǎn)者發(fā)送一條消息、消費(fèi)者手動(dòng) ACK、關(guān)閉服務(wù)后重啟驗(yàn)證消息還在。然后再一步步加上交換機(jī)路由、確認(rèn)回調(diào)、死信隊(duì)列。這個(gè)遞進(jìn)過程比我見過的任何教程都能幫你建立對(duì) MQ 的整體感知。接下來說幾個(gè)我在真實(shí)項(xiàng)目里積累的細(xì)節(jié)經(jīng)驗(yàn)這些不太容易在官方文檔里看到但很可能會(huì)在關(guān)鍵時(shí)刻幫你少踩一次坑。第一個(gè)經(jīng)驗(yàn)隊(duì)列聲明使用durable但持久化到磁盤的消息存在刷盤延遲。RabbitMQ 默認(rèn)對(duì)持久化消息是異步刷盤極端情況下機(jī)器掉電確實(shí)可能丟極少量數(shù)據(jù)。對(duì)業(yè)務(wù)消息嚴(yán)格丟不起的場景要么對(duì)消息落庫做補(bǔ)償要么用 Lazy Queue。RabbitMQ 3.12 之后新隊(duì)列默認(rèn)就是 Lazy Queue消息直接走磁盤犧牲一點(diǎn)吞吐?lián)Q取接近零丟失的保障對(duì)訂單、支付這類業(yè)務(wù)值得。第二個(gè)經(jīng)驗(yàn)微服務(wù)拆得越細(xì)別讓 MQ 隊(duì)列跟著拆得越碎。我看到有些團(tuán)隊(duì)為了解耦一個(gè)業(yè)務(wù)事件建一個(gè)交換機(jī)、一個(gè)隊(duì)列結(jié)果沒到一個(gè)月控制臺(tái)里幾十個(gè)隊(duì)列沒人說得清誰在用。我的原則是按業(yè)務(wù)域劃分隊(duì)列不按具體方法劃分。訂單域一個(gè)交換機(jī)路由鍵區(qū)分 create、cancel、pay 等動(dòng)作消費(fèi)方各取所需。后續(xù)擴(kuò)展新消費(fèi)者綁定同一個(gè)交換機(jī)即可不用動(dòng)上游代碼。第三個(gè)經(jīng)驗(yàn)死信隊(duì)列不是萬能的垃圾桶。死信隊(duì)列里堆積的消息要設(shè)置監(jiān)控告警。一旦死信隊(duì)列持續(xù)增長說明消費(fèi)者處理業(yè)務(wù)的大量異常未得到解決。我習(xí)慣把死信隊(duì)列的消費(fèi)邏輯做成一個(gè)補(bǔ)償網(wǎng)關(guān)接到消息后先查消息里的業(yè)務(wù)主鍵然后從業(yè)務(wù)庫里查當(dāng)前狀態(tài)能修復(fù)就調(diào)用補(bǔ)償接口確實(shí)無法處理的再落到日志表里人工介入。如果死信隊(duì)列本身沒人管那它最終就變成了一個(gè)消息垃圾桶既占內(nèi)存又掩蓋問題真正的異常被靜靜埋在里面。還有一點(diǎn)是關(guān)于團(tuán)隊(duì)協(xié)作的。MQ 是典型的跨服務(wù)、跨團(tuán)隊(duì)基礎(chǔ)設(shè)施比接口契約更講究規(guī)范。生產(chǎn)者改了路由鍵消費(fèi)者隊(duì)列的綁定不做相應(yīng)修改消息就會(huì)進(jìn)死信。我們項(xiàng)目的做法是在 Git 倉庫里單獨(dú)維護(hù)一份 MQ 消息規(guī)范文檔用表格列出每個(gè)交換機(jī)、每個(gè)路由鍵、每個(gè)隊(duì)列的負(fù)責(zé)人、業(yè)務(wù)含義、消息體格式示例。每次改動(dòng)都需要同步更新這份文檔在代碼評(píng)審上一并檢查。這套流程堅(jiān)持下來之后因?yàn)楦南⒏袷綄?dǎo)致的線上問題驟減。最后想說的是RabbitMQ 雖然上手不難但真正要讓它在微服務(wù)架構(gòu)里穩(wěn)定可靠地跑起來需要把可靠性、冪等性、可觀測性這三件事貫穿始終。消息能發(fā)出去只是工程的下限消息在任何情況下都能不丟不重不積壓才是工程的上限。希望這篇實(shí)踐總結(jié)能幫你在自己的微服務(wù)項(xiàng)目里更穩(wěn)地用好 RabbitMQ。如果后續(xù)有時(shí)間我會(huì)再寫一篇從消費(fèi)端到生產(chǎn)端的鏈路追蹤實(shí)踐講講在消息投遞的全鏈路中如何用 TraceId 串起調(diào)用鏈定位到底是哪一環(huán)拖慢了整個(gè)流程。