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