簡(jiǎn)介這是一套面向Java開發(fā)者的RabbitMQ入門到進(jìn)階實(shí)戰(zhàn)代碼案例。案例圍繞AMQP協(xié)議從生產(chǎn)者、消費(fèi)者、交換機(jī)、隊(duì)列等核心概念出發(fā)演示了基于amqp-client庫(kù)的消息發(fā)送與接收流程并擴(kuò)展到direct、topic、fanout等多種交換機(jī)路由策略以及事務(wù)、發(fā)布確認(rèn)、死信隊(duì)列、消息TTL等可靠性機(jī)制。壓縮包大小僅396KB共包含299個(gè)文件其中39個(gè)Java源碼與39個(gè)class可直接對(duì)應(yīng)學(xué)習(xí)另有188個(gè)XML配置和properties資源文件用于工程搭建輔以少量JSP頁(yè)面及Maven相關(guān)文件便于在IDEA等環(huán)境中運(yùn)行調(diào)試。已有8098人學(xué)習(xí)下載適合需要快速掌握消息中間件實(shí)際用法、或正在設(shè)計(jì)分布式異步系統(tǒng)的開發(fā)人員參考。通過(guò)閱讀和運(yùn)行這些示例可以理解RabbitMQ API的調(diào)用細(xì)節(jié)與消息路由原理少走彎路并快速落地到自己的項(xiàng)目中。 搞消息隊(duì)列這幾年RabbitMQ 是我用得最順手的一個(gè)。很多人一提到 RabbitMQ 就只會(huì)發(fā)個(gè) hello world真到了生產(chǎn)環(huán)境廣播、死信、JSON 序列化、多語(yǔ)言對(duì)接全得靠代碼案例一點(diǎn)點(diǎn)趟坑。這篇就用我實(shí)際跑通過(guò)的項(xiàng)目代碼把 RabbitMQ 從安裝、啟動(dòng)、SpringBoot 集成、C# 推送、死信隊(duì)列到 Docker 集群部署完整串一遍。適合剛接觸消息隊(duì)列的開發(fā)者也適合已經(jīng)在用但想補(bǔ)全細(xì)節(jié)的同行。1. 項(xiàng)目背景與整體設(shè)計(jì)思路1.1 為什么選 RabbitMQ而不是其他消息隊(duì)列做技術(shù)選型時(shí)很多人愛糾結(jié) Kafka 還是 RocketMQ但 RabbitMQ 在中小型系統(tǒng)、企業(yè)內(nèi)部服務(wù)間異步通信上優(yōu)勢(shì)非常明顯部署輕量、管理界面友好、路由規(guī)則靈活、多語(yǔ)言客戶端成熟。更重要的是它的 AMQP 協(xié)議模型交換機(jī)、隊(duì)列、綁定幾乎覆蓋了你日常能想到的所有消息場(chǎng)景。我這邊有個(gè)真實(shí)業(yè)務(wù)前端提交工單后系統(tǒng)需要同時(shí)通知多個(gè)服務(wù)去處理比如發(fā)短信、寫日志、更新統(tǒng)計(jì)。如果同步調(diào)用一個(gè)服務(wù)掛了整個(gè)流程就卡住。用 RabbitMQ 做廣播 fanout 模式消息發(fā)到交換機(jī)所有綁定的隊(duì)列都會(huì)收到互不影響。另一個(gè)場(chǎng)景是訂單超時(shí)未支付需要自動(dòng)關(guān)閉這就要用到死信隊(duì)列加延遲路由。1.2 業(yè)務(wù)場(chǎng)景梳理廣播、異步、死信分別解決什么問(wèn)題場(chǎng)景一廣播通知。一條消息要讓所有監(jiān)聽方都收到用 fanout 交換機(jī)最合適。場(chǎng)景二異步解耦。用戶注冊(cè)成功后只需要寫庫(kù)郵件驗(yàn)證碼之類的異步發(fā)出去用戶無(wú)感知。場(chǎng)景三死信兜底。消息消費(fèi)失敗、超時(shí)未被確認(rèn)進(jìn)入死信隊(duì)列后續(xù)做補(bǔ)償處理或人工介入。這三個(gè)場(chǎng)景對(duì)應(yīng)了 RabbitMQ 最常見的三類用法也正好是面試中常被追問(wèn)的“你在項(xiàng)目中哪里用到了 MQ”。能講清楚場(chǎng)景比背概念有用得多。1.3 環(huán)境準(zhǔn)備Windows 安裝、Docker 部署、集群搭建環(huán)境是第一步也是最容易踩坑的一步。Windows 下安裝 RabbitMQ 需要先裝 Erlang而且版本要匹配。我之前試過(guò) Erlang 24 配 RabbitMQ 3.9 沒(méi)問(wèn)題后來(lái)?yè)Q機(jī)器裝成 Erlang 26 就出現(xiàn)啟動(dòng)后管理界面打不開的情況。官方有兼容性對(duì)照表裝之前一定要看一眼。Docker 方式相對(duì)省心一條命令就能跑起來(lái)docker run -d --name rabbitmq \ -p 5672:5672 -p 15672:15672 \ -e RABBITMQ_DEFAULT_USERadmin \ -e RABBITMQ_DEFAULT_PASSadmin123 \ rabbitmq:3.12-management這樣裝完自帶管理插件瀏覽器訪問(wèn) 15672 端口就能看到界面。如果你要搭集群記住幾個(gè)關(guān)鍵點(diǎn)節(jié)點(diǎn) cookie 必須一致、每個(gè)節(jié)點(diǎn)的 hostname 不能被解析成 127.0.0.1、用rabbitmqctl join_cluster把新節(jié)點(diǎn)加入集群。Docker 下做集群我建議用--network host或者 docker-compose 固定容器名否則節(jié)點(diǎn)間通訊經(jīng)常會(huì)出幺蛾子。2. 核心代碼案例一SpringBoot 集成 RabbitMQ廣播 JSON2.1 引入依賴與基礎(chǔ)配置SpringBoot 集成 RabbitMQ 算是 Java 生態(tài)里最省事的方案。首先在pom.xml里加依賴dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency然后在application.yml里配置連接信息spring: rabbitmq: host: localhost port: 5672 username: admin password: admin123 virtual-host: / publisher-confirm-type: correlated publisher-returns: true這里有兩個(gè)容易被忽略的參數(shù)publisher-confirm-type和publisher-returns。前者開啟發(fā)送端確認(rèn)消息到達(dá)交換機(jī)后回調(diào)告訴你結(jié)果后者開啟消息路由不到隊(duì)列時(shí)的返回回調(diào)。生產(chǎn)環(huán)境一定要開不然消息丟失了你根本不知道。2.2 定義交換機(jī)、隊(duì)列和綁定關(guān)系RabbitMQ 的核心是交換機(jī)、隊(duì)列、綁定三者之間的關(guān)系。在我的廣播案例里項(xiàng)目里通常用注解方式直接聲明Configuration public class RabbitConfig { public static final String FANOUT_EXCHANGE exchange.notify; public static final String QUEUE_SMS queue.sms; public static final String QUEUE_LOG queue.log; Bean public FanoutExchange fanoutExchange() { return new FanoutExchange(FANOUT_EXCHANGE); } Bean public Queue smsQueue() { return QueueBuilder.durable(QUEUE_SMS).build(); } Bean public Queue logQueue() { return QueueBuilder.durable(QUEUE_LOG).build(); } Bean public Binding smsBinding() { return BindingBuilder.bind(smsQueue()).to(fanoutExchange()); } Bean public Binding logBinding() { return BindingBuilder.bind(logQueue()).to(fanoutExchange()); } }fanout 交換機(jī)不關(guān)心 routingKey只要消息發(fā)到交換機(jī)上所有綁定的隊(duì)列都能收到。如果你用 direct 或 topic那就要在 bind 的時(shí)候指定 routingKey發(fā)送時(shí)也得帶對(duì)應(yīng) key。曾經(jīng)有個(gè)同事把 direct 當(dāng) fanout 用routingKey 寫錯(cuò)導(dǎo)致消息全部進(jìn)入黑洞排查了一下午其實(shí)就是沒(méi)理解模型。2.3 生產(chǎn)者推送 JSON 消息把 JSON 放入 RabbitMQ是實(shí)際項(xiàng)目里最常見的操作。SpringBoot 里可以手動(dòng)轉(zhuǎn) JSON也可以配置消息轉(zhuǎn)換器讓convertAndSend自動(dòng)把對(duì)象序列化成 JSON。我推薦配置一個(gè)全局 Jackson 轉(zhuǎn)換器Bean public MessageConverter messageConverter() { return new Jackson2JsonMessageConverter(); }生產(chǎn)者推送Service public class NotifyService { Autowired private RabbitTemplate rabbitTemplate; public void sendNotify(NotifyMessage message) { CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString()); rabbitTemplate.convertAndSend( RabbitConfig.FANOUT_EXCHANGE, , message, correlationData ); } }NotifyMessage就是一個(gè)普通 POJO字段包括id、content、createTime。加了Jackson2JsonMessageConverter后RabbitTemplate 會(huì)把對(duì)象轉(zhuǎn)成 JSON 字符串再包裝成消息體發(fā)送。注意如果你在 fanout 模式下把 routingKey 隨便填一個(gè)字符串消息依然能發(fā)出去因?yàn)?fanout 根本不檢查 key這也容易造成“明明綁定了隊(duì)列卻收不到消息”的錯(cuò)覺(jué)。2.4 消費(fèi)者監(jiān)聽與消息確認(rèn)消費(fèi)者就更簡(jiǎn)單了用RabbitListener注解就能監(jiān)聽隊(duì)列Component public class NotifyConsumer { RabbitListener(queues RabbitConfig.QUEUE_SMS) public void handleSms(NotifyMessage message) { System.out.println(發(fā)送短信: message.getContent()); } RabbitListener(queues RabbitConfig.QUEUE_LOG) public void handleLog(NotifyMessage message) { System.out.println(寫日志: message.getContent()); } }這里有個(gè)非常重要的參數(shù)RabbitListener(queues ..., ackMode MANUAL)。默認(rèn)情況下 SpringBoot 的監(jiān)聽器是自動(dòng)確認(rèn)只要方法執(zhí)行完就 ack。但如果你在處理消息時(shí)拋了異常Spring 會(huì)自動(dòng) requeue導(dǎo)致消息一直在隊(duì)列頭部打轉(zhuǎn)形成死循環(huán)。常見做法是改成手動(dòng)確認(rèn)RabbitListener(queues RabbitConfig.QUEUE_SMS, ackMode MANUAL) public void handleSms(NotifyMessage message, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { System.out.println(發(fā)送短信: message.getContent()); channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, true); } }basicNack的第三個(gè)參數(shù)是requeue設(shè)為 true 表示重新放回隊(duì)列。如果確認(rèn)消息確實(shí)有問(wèn)題建議設(shè)為 false配合死信隊(duì)列做兜底而不是無(wú)限重試。3. 核心代碼案例二C# 推送 RabbitMQ3.1 C# 端連接與發(fā)送很多公司是 Java 后端 C# 桌面端或 .NET 服務(wù)并存這時(shí)候 RabbitMQ 跨語(yǔ)言優(yōu)勢(shì)就體現(xiàn)出來(lái)了。C# 用官方 RabbitMQ.Client 庫(kù)NuGet 里直接裝Install-Package RabbitMQ.Client發(fā)送端核心代碼using RabbitMQ.Client; using System.Text; using System.Text.Json; var factory new ConnectionFactory() { HostName localhost, Port 5672, UserName admin, Password admin123 }; using var connection factory.CreateConnection(); using var channel connection.CreateModel(); var notify new { id Guid.NewGuid().ToString(), content hello from csharp, createTime DateTime.Now }; string json JsonSerializer.Serialize(notify); byte[] body Encoding.UTF8.GetBytes(json); channel.BasicPublish( exchange: exchange.notify, routingKey: , basicProperties: null, body: body );注意C# 端聲明交換機(jī)的時(shí)候不能用ExchangeDeclarePassive去檢查但不存在就報(bào)錯(cuò)。通常做法是確保 C# 里也有一樣的交換機(jī)聲明邏輯或者由 Java 服務(wù)統(tǒng)一把交換機(jī)隊(duì)列建好C# 只負(fù)責(zé)發(fā)送。我在實(shí)際項(xiàng)目里是讓 Java 服務(wù)管理所有交換機(jī)隊(duì)列聲明C# 只推送避免兩邊聲明不一致。3.2 讓 JSON 消息被多語(yǔ)言消費(fèi)者正確消費(fèi)跨語(yǔ)言消費(fèi)有個(gè)細(xì)節(jié)消息體里的字段命名風(fēng)格。Java 默認(rèn)字段是駝峰createTimeC# 序列化默認(rèn)也是駝峰但如果 C# 側(cè)反序列化成CreateTime屬性System.Text.Json默認(rèn)區(qū)分大小寫就會(huì)反序列化失敗。解決方法是統(tǒng)一字段命名或在 C# 反序列化時(shí)配置PropertyNameCaseInsensitive truevar options new JsonSerializerOptions { PropertyNameCaseInsensitive true }; var msg JsonSerializer.DeserializeNotifyMessage(json, options);還有一點(diǎn)RabbitMQ 的消息頭里content_type一定要是application/json。Java 的Jackson2JsonMessageConverter會(huì)自動(dòng)設(shè)置但 C# 手動(dòng)BasicPublish的時(shí)候如果不設(shè)置basicProperties默認(rèn)是text/plain。雖然消費(fèi)者按 JSON 解也能成功但有些框架會(huì)根據(jù) content_type 做消息轉(zhuǎn)換建議顯式設(shè)置var props channel.CreateBasicProperties(); props.ContentType application/json; channel.BasicPublish(exchange.notify, , props, body);3.3 C# 消費(fèi)端的常用寫法C# 消費(fèi)端用 EventingBasicConsumer 訂閱using var channel connection.CreateModel(); channel.QueueDeclare(queue.result, durable: true, exclusive: false, autoDelete: false); var consumer new EventingBasicConsumer(channel); consumer.Received (model, ea) { string json Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($收到消息: {json}); channel.BasicAck(ea.DeliveryTag, false); }; channel.BasicConsume(queue: queue.result, autoAck: false, consumer: consumer);這里我特別強(qiáng)調(diào)autoAck: false。C# 消費(fèi)端如果設(shè)成 true消息一收到就確認(rèn)萬(wàn)一處理邏輯崩了消息就丟了。設(shè)成 false 配合手動(dòng)BasicAck至少在異常時(shí)能通過(guò)BasicNack把消息放回隊(duì)列或進(jìn)死信。4. 死信隊(duì)列實(shí)戰(zhàn)延遲消息與異常兜底4.1 死信隊(duì)列的原理死信隊(duì)列Dead Letter Queue本質(zhì)上就是普通的 RabbitMQ 隊(duì)列只不過(guò)它專門接收“沒(méi)有正常被消費(fèi)”的消息。消息變成死信的條件有三個(gè)消費(fèi)者basicNack且requeuefalse、消息過(guò)期未被消費(fèi)、隊(duì)列長(zhǎng)度達(dá)到上限。利用這個(gè)機(jī)制可以模擬延遲隊(duì)列給普通隊(duì)列設(shè)置消息過(guò)期時(shí)間 TTL消息過(guò)期后自動(dòng)轉(zhuǎn)到死信交換機(jī)死信交換機(jī)再把消息路由到真正的消費(fèi)隊(duì)列。RabbitMQ 本身沒(méi)有延遲隊(duì)列插件時(shí)這是最經(jīng)典的延遲消息實(shí)現(xiàn)方案。4.2 代碼實(shí)現(xiàn)普通隊(duì)列綁定死信交換機(jī)SpringBoot 里聲明一個(gè)帶死信屬性的隊(duì)列Configuration public class DelayConfig { public static final String DELAY_EXCHANGE exchange.delay; public static final String DELAY_QUEUE queue.delay; public static final String DEAD_EXCHANGE exchange.dead; public static final String DEAD_QUEUE queue.dead; Bean public DirectExchange delayExchange() { return new DirectExchange(DELAY_EXCHANGE); } Bean public DirectExchange deadExchange() { return new DirectExchange(DEAD_EXCHANGE); } Bean public Queue delayQueue() { return QueueBuilder.durable(DELAY_QUEUE) .deadLetterExchange(DEAD_EXCHANGE) .deadLetterRoutingKey(dead) .build(); } Bean public Queue deadQueue() { return QueueBuilder.durable(DEAD_QUEUE).build(); } Bean public Binding delayBinding() { return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(delay); } Bean public Binding deadBinding() { return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(dead); } }生產(chǎn)者設(shè)置消息過(guò)期時(shí)間MessagePostProcessor processor message - { message.getMessageProperties().setExpiration(1800000); return message; }; rabbitTemplate.convertAndSend(DELAY_EXCHANGE, delay, order, processor);setExpiration的單位是毫秒1800000 就是 30 分鐘。這條消息發(fā)到 delay 隊(duì)列后不會(huì)被立即消費(fèi)而是要等 30 分鐘過(guò)期然后轉(zhuǎn)進(jìn)死信交換機(jī)最終到達(dá)真正的消費(fèi)隊(duì)列。4.3 30分鐘積壓場(chǎng)景分析與參數(shù)調(diào)優(yōu)熱詞里有“rabbitmq 死信30分種會(huì)壓多少”我覺(jué)得這個(gè)問(wèn)題的本質(zhì)是延遲消息設(shè)為 30 分鐘后同時(shí)會(huì)有多少消息積壓在延遲隊(duì)列里。要估算積壓量核心就兩個(gè)指標(biāo)每秒進(jìn)入延遲隊(duì)列的消息數(shù)和延遲時(shí)間。比如系統(tǒng)每秒產(chǎn)生 100 條超時(shí)訂單延遲 30 分鐘1800 秒那累積在隊(duì)列里的消息數(shù)大約就是 100 * 1800 180000 條。這是理論積壓量。這數(shù)字影響什么影響隊(duì)列內(nèi)存、磁盤持久化、以及消息過(guò)期后瞬間轉(zhuǎn)入死信隊(duì)列的沖擊力。如果 18 萬(wàn)條消息同一時(shí)間點(diǎn)過(guò)期那死信隊(duì)列會(huì)瞬間收到大量消息消費(fèi)者可能扛不住。解決辦法是把過(guò)期時(shí)間做離散化比如在業(yè)務(wù)上將延遲時(shí)間增加隨機(jī)偏移量讓過(guò)期時(shí)間分散。另一個(gè)辦法是給死信隊(duì)列配置多個(gè)消費(fèi)者加并發(fā)或者用 RabbitMQ 的x-max-priority來(lái)控制某些緊急消息優(yōu)先消費(fèi)。實(shí)際測(cè)試中普通單機(jī) RabbitMQ 幾萬(wàn)條消息沒(méi)問(wèn)題但 18 萬(wàn)條消息同時(shí)觸發(fā)流轉(zhuǎn)管理界面會(huì)明顯卡頓消費(fèi)者若不加prefetch限制內(nèi)存占用直接飆升。所以我建議給消費(fèi)者設(shè)置prefetchCount用 SimpleMessageListenerContainer 的時(shí)候Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setPrefetchCount(100); factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); return factory; }5. 常見問(wèn)題與排查技巧5.1 啟動(dòng)失敗、端口占用與管理界面打不開Windows 下 RabbitMQ 啟動(dòng)失敗八成都出在 Erlang 版本不匹配或主機(jī)名解析上。解決辦法用管理員身份打開命令行先運(yùn)行rabbitmq-service stop再rabbitmq-service uninstall清掉 Erlang 的 cookie 緩存重裝匹配版本。端口方面5672 是 AMQP 端口15672 是管理界面端口。用 Docker 時(shí)如果 15672 被占用改映射端口即可但 5672 必須跟客戶端配置保持一致。排查命令rabbitmqctl status rabbitmq-diagnostics -q ping rabbitmq-plugins enable rabbitmq_management5.2 消息堆積、重復(fù)消費(fèi)與確認(rèn)機(jī)制消息堆積常見原因有兩個(gè)生產(chǎn)者速度遠(yuǎn)大于消費(fèi)者速度、消費(fèi)者處理邏輯太慢。前者可以增加消費(fèi)者實(shí)例或提高 prefetch后者需要優(yōu)化業(yè)務(wù)代碼甚至把下游寫庫(kù)改成批量寫。還有一種隱蔽情況消費(fèi)者一直接收消息但每一條都拋異常 requeue看起來(lái)隊(duì)列長(zhǎng)度不變實(shí)際是在死循環(huán)這個(gè)問(wèn)題只能靠日志定位。重復(fù)消費(fèi)是消息系統(tǒng)永遠(yuǎn)繞不開的話題。RabbitMQ 不保證 exactly once所以消費(fèi)者必須具備冪等性。最簡(jiǎn)單的方式是消息里帶上業(yè)務(wù)唯一 ID消費(fèi)前查一下 Redis 或數(shù)據(jù)庫(kù)已經(jīng)處理過(guò)就直接 ack。5.3 Docker 集群部署的常見坑Docker 下搭 RabbitMQ 集群我踩過(guò)的坑有幾個(gè)。第一是容器重啟后 hostname 變化導(dǎo)致集群節(jié)點(diǎn)失效解決方法是固定 hostname 或用--hostname參數(shù)。第二是節(jié)點(diǎn)間 cookie 不一致/var/lib/rabbitmq/.erlang.cookie要保證一致。第三是用-p 5672:5672這種端口映射方式做集群時(shí)節(jié)點(diǎn)間通訊走了隨機(jī)端口防火墻攔掉就連不上。建議直接用 docker-compose定義好網(wǎng)絡(luò)和固定容器名避免不必要的麻煩。最后再分享一個(gè)小技巧如果你正在準(zhǔn)備面試或剛接手 RabbitMQ 項(xiàng)目建議你自己動(dòng)手搭一套最小的代碼案例一個(gè)生產(chǎn)者定時(shí)推送 JSON 消息兩個(gè)消費(fèi)者分別監(jiān)聽其中一個(gè)消費(fèi)失敗進(jìn)死信隊(duì)列再寫一個(gè)定時(shí)任務(wù)掃描死信隊(duì)列做補(bǔ)償。這套案例跑通后你對(duì)交換機(jī)、隊(duì)列、綁定、確認(rèn)機(jī)制、死信流轉(zhuǎn)的理解會(huì)徹底貫通。很多概念看十遍文檔不如親手看一眼消息進(jìn)到死信隊(duì)列時(shí)管理界面上的流轉(zhuǎn)路徑。本文還有配套的精品資源點(diǎn)擊獲取