戰(zhàn)筆記：從消息模型到可靠消息機(jī)制)
我第一次正兒八經(jīng)看 JMS 和 ActiveMQ并不是因?yàn)橐嬖嚤嘲斯啥琼?xiàng)目里兩個(gè)服務(wù)之間的同步調(diào)用把數(shù)據(jù)庫連接池打爆了。上游接口一個(gè)超時(shí)下游跟著雪崩運(yùn)維的告警群半夜響成一片。翻來覆去折騰了幾天之后同事丟給我一句話“用消息中間件把調(diào)用拆開吧先看看 JMS 和 ActiveMQ?！碑?dāng)時(shí)我連 JMS 和 MQ 的關(guān)系都說不清楚硬著頭皮把那套消息模型啃了一遍才有了這篇“JMS、ActiveMQ 學(xué)習(xí)一則”。這篇筆記適合兩類人一是剛開始接觸消息中間件、被一堆 JMS API 術(shù)語搞暈的開發(fā)者二是能把 Demo 跑通但不清楚 Queue、Topic、ACK、持久化到底意味著什么的人。我會把規(guī)范、代碼、踩坑都放在一起講盡量還原當(dāng)時(shí)的學(xué)習(xí)路徑也把網(wǎng)絡(luò)上那些講得含糊的地方補(bǔ)完整。1. 為什么我先啃 JMS 而不是直接學(xué)某個(gè) MQ 客戶端1.1 消息中間件解決的核心問題把“同步等待”變成“異步解耦”先說我當(dāng)時(shí)的場景。有個(gè)訂單創(chuàng)建服務(wù)處理完訂單之后要通知庫存服務(wù)扣庫存、通知積分服務(wù)加積分、通知短信服務(wù)發(fā)通知。最開始寫的代碼很“樸素”——A 服務(wù)直接用 HTTP 調(diào)用 B 服務(wù)同步等返回。看起來邏輯很直白但一壓測就出問題任何一個(gè)下游服務(wù)變慢訂單接口的響應(yīng)時(shí)間立刻被拖垮下游服務(wù)一重啟上游的調(diào)用直接報(bào)超時(shí)訂單雖然建成功了但庫存、積分、短信全部沒執(zhí)行數(shù)據(jù)就錯(cuò)亂了。消息中間件解決的就是這種“你完成你的我完成我的然后把變化告訴別人”的通信問題。訂單服務(wù)把“訂單已創(chuàng)建”這個(gè)事實(shí)寫到消息隊(duì)列里訂單服務(wù)的接口直接返回成功庫存、積分、短信服務(wù)各自從隊(duì)列里取消息按自己的節(jié)奏處理。這樣三個(gè)下游服務(wù)誰慢、誰掛都不影響訂單主流程。這個(gè)模型里有幾個(gè)術(shù)語你得先建立直覺Broker是中間件服務(wù)器負(fù)責(zé)消息的存儲和轉(zhuǎn)發(fā)Producer是消息的生產(chǎn)者Consumer是消息的消費(fèi)者。消息不是直接給某個(gè)消費(fèi)者的而是先放到 Broker 的某個(gè)“容器”里消費(fèi)者自己來取。這個(gè)“容器”就是后來的 Queue 或 Topic。1.2 JMS 是規(guī)范ActiveMQ 是實(shí)現(xiàn)先理解這對關(guān)系JMSJava Message Service是 Java 平臺制定的一套消息服務(wù)接口規(guī)范它規(guī)定了 Java 程序怎么創(chuàng)建連接、怎么發(fā)消息、怎么收消息但 JMS 自己不干活真正干活的是符合這套規(guī)范的 provider比如 Apache ActiveMQ、開源的 OpenMQ 等。這個(gè)關(guān)系有點(diǎn)像 JDBC 和 MySQL 驅(qū)動的關(guān)系——JDBC 規(guī)定了你用什么方法連接數(shù)據(jù)庫MySQL 驅(qū)動負(fù)責(zé)把這些方法翻譯成 MySQL 能懂的協(xié)議。這就解釋了為什么學(xué)習(xí)的時(shí)候要先看 JMS 再看 ActiveMQ如果你只對著 ActiveMQ 的 API 抄一遍換個(gè) MQ 你就廢了如果你把 JMS 的核心模型理解了接觸任何一個(gè)兼容 JMS 的中間件都能很快上手。我當(dāng)初踩過一個(gè)認(rèn)知誤區(qū)以為 ActiveMQ 提供了比 JMS 多得多的高級功能所以直接按 ActiveMQ 的文檔寫就行。實(shí)際上 ActiveMQ 對 JMS 規(guī)范做了很完整的實(shí)現(xiàn)它的核心 API 就是 JMS API你寫代碼時(shí) import 的主要是javax.jms.*這個(gè)包而不是org.apache.activemq.*。后者只在構(gòu)造連接工廠時(shí)才會用到。1.3 什么場景才需要引入消息中間件先潑盆冷水消息中間件不是銀彈學(xué)習(xí)過程中我也遇到過濫用它的團(tuán)隊(duì)。一個(gè)內(nèi)部查詢接口沒有任何瞬時(shí)流量壓力硬是套一層消息隊(duì)列結(jié)果多了一個(gè)運(yùn)維節(jié)點(diǎn)、多了一段延遲、多了一堆消息亂序問題。什么時(shí)候才值得引入我的判斷標(biāo)準(zhǔn)有三條一是上下游節(jié)奏不一致下游來不及同步處理需要用隊(duì)列做緩沖二是下游故障不應(yīng)該影響上游主流程需要隔離故障三是同一份業(yè)務(wù)事件需要被多個(gè)系統(tǒng)各自消費(fèi)比如訂單創(chuàng)建之后庫存、積分、日志都要處理。反過來說如果只是簡單的 A 調(diào) B響應(yīng)時(shí)間敏感且 B 很少故障那直接 HTTP 調(diào)用反而更合適。這部分認(rèn)知雖然當(dāng)時(shí)不成熟但后來回看幫我在選型時(shí)規(guī)避了不少坑。2. JMS 核心模型五個(gè)抽象角色串起全部消息邏輯JMS 1.1 里抽象出的角色其實(shí)就五個(gè)ConnectionFactory、Connection、Session、Destination、MessageProducer/MessageConsumer。剩下的概念比如 Message、AcknowledgeMode 都是在這五者之上附加的。把這五個(gè)角色串成一個(gè)故事你就能看懂 90% 的 JMS 代碼。2.1 ConnectionFactory創(chuàng)建連接的唯一入口ConnectionFactory 是一個(gè)連接工廠接口它封裝了與某個(gè) Broker 建立連接所需的參數(shù)。在 ActiveMQ 中最常見的是ActiveMQConnectionFactory構(gòu)造時(shí)傳入 Broker 的地址典型格式是tcp://127.0.0.1:61616。ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616);這個(gè)對象通常是重量級對象創(chuàng)建成本高所以應(yīng)該復(fù)用不要每次發(fā)消息都 new 一個(gè)。我看到不少剛?cè)腴T的同學(xué)把ActiveMQConnectionFactory寫在方法內(nèi)部短時(shí)間大量請求一來連接資源就爆炸了。正確做法是使用連接池或至少是單例持有。2.2 Connection 與 Session消息通信的骨架從工廠拿到Connection之后需要顯式調(diào)用connection.start()這時(shí)候連接才真正建立。然后從 Connection 創(chuàng)建 Session。Session 是一個(gè)非常重要的中間層消息的生產(chǎn)和消費(fèi)都是基于 Session 的Session 的創(chuàng)建方式?jīng)Q定了本次會話的確認(rèn)模式和事務(wù)邊界。Connection connection factory.createConnection(); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);createSession的第一個(gè)參數(shù)是事務(wù)標(biāo)志第二參數(shù)是確認(rèn)模式。在 JMS 1.1 中如果第一個(gè)參數(shù)傳 true第二個(gè)參數(shù)會被忽略因?yàn)槭聞?wù)模式下發(fā)送和接收都通過commit和rollback控制。剛開始不用懵理解成“Session 是一段通信上下文凡是這次會話內(nèi)的消息都可以在這個(gè)上下文里標(biāo)記狀態(tài)”就行。2.3 Destination消息要發(fā)往的地址Destination 是抽象類實(shí)際子類要么是Queue要么是Topic。Queue 對應(yīng)點(diǎn)對點(diǎn)模型Topic 對應(yīng)發(fā)布訂閱模型。在代碼里session.createQueue()或session.createTopic()創(chuàng)建的只是邏輯地址不需要在 Broker 側(cè)預(yù)先建好ActiveMQ 會在第一次使用時(shí)自動創(chuàng)建。Queue queue session.createQueue(DEMO.ORDER); Topic topic session.createTopic(DEMO.NEWS);這里有個(gè)細(xì)節(jié)隊(duì)列名稱建議使用“域名前綴 業(yè)務(wù)名”的格式比如ORDER.CREATED、STOCK.DEDUCTED為了在管理后臺一眼能看出用途。我見過用queue1、test2這樣的名字上生產(chǎn)的后來排查問題差點(diǎn)被隊(duì)友打死。2.4 MessageProducer 與 MessageConsumer實(shí)際干活的兩個(gè)角色生產(chǎn)者和消費(fèi)者都在 Session 上創(chuàng)建。生產(chǎn)者創(chuàng)建時(shí)可以指定默認(rèn)目的地也可以不指定發(fā)送時(shí)再傳目的地。生產(chǎn)者還支持設(shè)置是否持久化、優(yōu)先級、存活時(shí)間等特性這部分會影響消息的存儲行為后面展開。MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); TextMessage message session.createTextMessage(hello jms); producer.send(message);消費(fèi)者既可以主動拉取消息也可以注冊消息監(jiān)聽器異步接收消息。主動拉取用consumer.receive()它會阻塞等待可以傳入超時(shí)時(shí)間異步監(jiān)聽用MessageListener接口Broker 一有消息就回調(diào)onMessage方法。2.5 Message 結(jié)構(gòu)消息頭、屬性、消息體三層分工JMS 的消息對象由三部分組成。第一個(gè)是消息頭header存放 JMS 規(guī)范定義的固定字段比如JMSDestination、JMSMessageID、JMSPriority、JMSExpiration。第二個(gè)是消息屬性properties可以看成附加的元數(shù)據(jù)支持字符串、數(shù)字、布爾等類型常用在消息篩選和路由上。第三個(gè)是消息體body存放業(yè)務(wù)數(shù)據(jù)本身。消息體有幾種類型TextMessage適合傳輸文本和 JSON 字符串BytesMessage適合二進(jìn)制數(shù)據(jù)MapMessage適合鍵值對結(jié)構(gòu)ObjectMessage適合序列化對象但跨語言不友好StreamMessage用得最少。實(shí)際項(xiàng)目里 80% 的流量都是TextMessage因?yàn)闃I(yè)務(wù)方一般約定用 JSON 傳結(jié)構(gòu)化數(shù)據(jù)不會強(qiáng)行依賴 Java 對象序列化。我自己的習(xí)慣是消息體只放必要的業(yè)務(wù)數(shù)據(jù)別把整個(gè)數(shù)據(jù)庫實(shí)體塞進(jìn)去。因?yàn)橄⒖赡鼙欢鄠€(gè)系統(tǒng)消費(fèi)每個(gè)系統(tǒng)關(guān)心字段不一樣塞整個(gè)實(shí)體不僅體積大而且上游實(shí)體變更會導(dǎo)致下游反序列化失敗。3. ActiveMQ 落地實(shí)操從下載啟動到第一個(gè)消息跑通理論講完了得有真家伙。我當(dāng)時(shí)本地環(huán)境是 JDK 8下載的是 Apache ActiveMQ 的穩(wěn)定發(fā)行版。注意 ActiveMQ 有兩個(gè)分支一個(gè)叫 ActiveMQ Classic一個(gè)叫 ActiveMQ Artemis兩者都是 ActiveMQ但 Artemis 是新一代架構(gòu)API 上對 JMS 規(guī)范的支持更現(xiàn)代。作為學(xué)習(xí) JMS 的入口我建議用 Classic因?yàn)橘Y料多、踩坑記錄全先把規(guī)范玩明白。3.1 啟動 Broker十分鐘確認(rèn)環(huán)境下載解壓后進(jìn)入 bin 目錄Linux/macOS 下執(zhí)行./activemq startWindows 下執(zhí)行activemq.bat start。啟動完成后默認(rèn)監(jiān)聽以下幾個(gè)重要端口端口用途61616JMS 客戶端連接的 OpenWire 端口8161Web 管理控制臺端口5672AMQP 協(xié)議端口默認(rèn)也開啟打開瀏覽器訪問http://127.0.0.1:8161/默認(rèn)登錄賬號是 admin密碼也是 admin進(jìn)去就能看到隊(duì)列列表、Topic 列表、連接數(shù)、消息統(tǒng)計(jì)等。我強(qiáng)烈建議學(xué)習(xí)階段養(yǎng)成盯控制臺的習(xí)慣很多代碼“看起來沒反應(yīng)”的問題在控制臺里一眼就能看出來——消息到底有沒有進(jìn)來、被誰消費(fèi)了、有沒有進(jìn)入死信。3.2 第一個(gè) JMS 生產(chǎn)者發(fā)送一條持久化文本消息下面這端代碼是我當(dāng)時(shí)跑通的第一個(gè)生產(chǎn)者。為了方便閱讀我把異常處理簡化了學(xué)習(xí)階段先跑通生產(chǎn)上請務(wù)必正確封裝資源關(guān)閉邏輯。import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class QueueProducer { public static void main(String[] args) throws Exception { // 1. 連接工廠指向本地 Broker ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); // 2. 創(chuàng)建連接并啟動 Connection connection factory.createConnection(); connection.start(); // 3. 創(chuàng)建會話非事務(wù) 自動確認(rèn) Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 創(chuàng)建目的地 Queue queue session.createQueue(DEMO.ORDER); // 5. 創(chuàng)建生產(chǎn)者 MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 6. 發(fā)送一條文本消息 TextMessage textMessage session.createTextMessage(訂單已創(chuàng)建訂單號20250101); producer.send(textMessage); System.out.println(已發(fā)送消息 textMessage.getText()); // 7. 關(guān)閉資源 producer.close(); session.close(); connection.close(); } }這個(gè)例子里最容易忽略的是connection.start()。有的教程會在創(chuàng)建生產(chǎn)者之前就調(diào)用 start有的會在發(fā)送之后才調(diào)用。start()本身不是發(fā)送動作的開關(guān)但它會讓連接進(jìn)入可發(fā)送/接收狀態(tài)。如果你只創(chuàng)建連接不 start消息照樣能發(fā)出去但消費(fèi)者端不 start 的話就一直收不到消息。我當(dāng)初就踩過這個(gè)坑后面專門講。3.3 第一個(gè) JMS 消費(fèi)者同步接收和異步監(jiān)聽兩種方式消費(fèi)者有兩種寫法我建議都試一遍。第一種是主動 receive適合測試和批處理任務(wù)。import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class QueueConsumer { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); Connection connection factory.createConnection(); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(DEMO.ORDER); MessageConsumer consumer session.createConsumer(queue); // 阻塞 3 秒沒消息會超時(shí)返回 null Message message consumer.receive(3000); if (message instanceof TextMessage) { System.out.println(收到消息 ((TextMessage) message).getText()); } consumer.close(); session.close(); connection.close(); } }第二種是注冊監(jiān)聽器異步接收實(shí)際項(xiàng)目里這種用得更多因?yàn)樗粫枞麡I(yè)務(wù)線程。示例代碼我放在下面consumer.setMessageListener(new MessageListener() { Override public void onMessage(Message message) { if (message instanceof TextMessage) { try { System.out.println(異步收到消息 ((TextMessage) message).getText()); } catch (JMSException e) { e.printStackTrace(); } } } }); Thread.sleep(5000);這里有個(gè)容易翻車的地方Thread.sleep(5000)是必要的。因?yàn)閙ain方法執(zhí)行完后 Java 進(jìn)程就退出了消費(fèi)者監(jiān)聽器注冊完還沒來得及收消息進(jìn)程就沒了。看起來像是“監(jiān)聽器沒生效”其實(shí)只是主線程提前退出。生產(chǎn)環(huán)境一般由應(yīng)用服務(wù)器或 Spring 容器持有消費(fèi)者生命周期不存在這個(gè)問題。3.4 跑完之后去控制臺觀察四個(gè)維度第一個(gè)消息鏈路跑通后我不急著寫第二個(gè) Demo而是去管理控制臺把所有變化對照了一遍。四個(gè)維度最有價(jià)值Number Of Pending Messages隊(duì)列里待消費(fèi)的消息數(shù)量。如果這個(gè)數(shù)字一直漲說明消費(fèi)端有問題。Number Of Consumers當(dāng)前有多少消費(fèi)者掛著。Messages Enqueued / Dequeued入隊(duì)和出隊(duì)的累計(jì)數(shù)能看出積壓到底是多少。Blocked Time連接被阻塞的時(shí)間如果很大說明消費(fèi)者處理不過來。先建立這套觀察手段后面排坑會省很多時(shí)間。空口說“消息丟了”“隊(duì)列卡了”很容易被運(yùn)維懟把控制臺截圖甩過去才是正確的溝通方式。4. 隊(duì)列模式和發(fā)布訂閱模式看起來代碼差不多語義差遠(yuǎn)了JMS 里最繞、也最影響選型的就是兩種消息模型點(diǎn)對點(diǎn)模式Queue和發(fā)布訂閱模式Topic。很多初學(xué)者能分別跑通但沒搞明白為什么需要兩種。4.1 Queue一條消息只被一個(gè)消費(fèi)者取走點(diǎn)對點(diǎn)模型下一條消息只會被一個(gè)消費(fèi)者成功接收。你可以啟動多個(gè)消費(fèi)者實(shí)例同時(shí)消費(fèi)同一個(gè)隊(duì)列ActiveMQ 會自動做負(fù)載均衡把不同消息分發(fā)給不同的消費(fèi)者。這像是一堆快遞包裹放在驛站多個(gè)配送員各自拿走一部分每個(gè)包裹最終只會被一個(gè)配送員取走。這個(gè)模型適合“任務(wù)分發(fā)”場景比如一批待處理文件10 個(gè) worker 搶占處理比如訂單狀態(tài)變更庫存服務(wù)和積分服務(wù)如果都要處理就不能用同一個(gè) Queue因?yàn)橄⒅唤o一個(gè)人。這里我補(bǔ)充一個(gè)容易誤解的點(diǎn)多個(gè)消費(fèi)者消費(fèi)同一個(gè)隊(duì)列和 JMS 1.1 里的 QueueReceiver 概念有細(xì)微差別。如果兩個(gè)消費(fèi)者的selector條件互斥消息可能會按條件被分流但默認(rèn)無 selector 時(shí)就是平均分發(fā)不是廣播。4.2 Topic一條消息廣播給所有訂閱者發(fā)布訂閱模型下一條消息會被所有“當(dāng)前活躍訂閱者”都收到。這像是廣播喇叭誰在聽誰就能聽見。Topic 模式下沒有“競爭消費(fèi)”的概念消費(fèi)者訂閱同一個(gè) Topic各自都能收到全量消息。但 Topic 有一個(gè)關(guān)鍵限制默認(rèn)情況下非持久訂閱者只在連接期間且保持訂閱狀態(tài)時(shí)才能收到消息。如果消費(fèi)者先掛了生產(chǎn)者后發(fā)消息這個(gè)消費(fèi)者重啟之后是收不到“離線期間”的消息的。這讓我當(dāng)初很意外因?yàn)樗?Queue 的行為明顯不一樣——隊(duì)列里消費(fèi)者掛多久都不怕消息會積壓著但 Topic 不會。要解決離線訂閱丟失的問題JMS 提供了持久訂閱Durable Subscription。持久訂閱狀態(tài)下Broker 會為這個(gè)訂閱者保存未消費(fèi)的消息等它上線后再推送。4.3 代碼對比一條線上隊(duì)列一條線上廣播Queue 和 Topic 的代碼結(jié)構(gòu)幾乎一致只有兩行不同——目的地對象和創(chuàng)建方式。// 消費(fèi)者隊(duì)列模式 Queue queue session.createQueue(DEMO.ORDER); MessageConsumer consumer session.createConsumer(queue);// 消費(fèi)者Topic 模式非持久訂閱 Topic topic session.createTopic(DEMO.NEWS); MessageConsumer consumer session.createConsumer(topic);持久訂閱的消費(fèi)者寫法會多一點(diǎn)Topic topic session.createTopic(DEMO.NEWS); // 需要為訂閱者設(shè)置唯一名稱 MessageConsumer consumer session.createDurableSubscriber(topic, news-consumer-1);createDurableSubscriber第二個(gè)參數(shù)是訂閱 IDBroker 用這個(gè) ID 區(qū)分不同的持久訂閱者。如果你用相同的訂閱 ID 連接兩次第二次會替代第一次的訂閱關(guān)系。發(fā)送端完全一樣MessageProducer producer session.createProducer(topic); producer.send(session.createTextMessage(財(cái)經(jīng)早報(bào)));4.4 選型判斷別讓“感覺好用”替你做決定學(xué)習(xí)期間我總結(jié)了一個(gè)很簡單的判斷規(guī)則同一業(yè)務(wù)事件如果多個(gè)系統(tǒng)需要各自獨(dú)立處理用 Topic如果一個(gè)任務(wù)的多個(gè)實(shí)例競爭處理同一批消息用 Queue。舉兩個(gè)對比場景訂單創(chuàng)建后觸發(fā)庫存扣減、積分增加、短信通知這是“一魚多吃”三個(gè)系統(tǒng)各自消費(fèi)同一事件應(yīng)該用 Topic 或給每個(gè)消費(fèi)者建獨(dú)立 Queue。秒殺活動里有 50 萬條搶購請求后面有 10 臺冪等處理服務(wù)實(shí)例消化這是“任務(wù)競爭”應(yīng)該用 Queue。更常見的企業(yè)做法是“混合使用”用 Topic 接收領(lǐng)域事件然后每個(gè)訂閱者內(nèi)部再用 Queue 做異步削峰。初學(xué)階段先別追求花哨能把 Queue 和 Topic 的邊界吃透就超過了大多數(shù)人。5. 可靠消息的底層機(jī)制確認(rèn)、事務(wù)、持久化一次講清消息中間件最常被問的一句話是“消息到底會不會丟”答案取決于三層配置消費(fèi)者的確認(rèn)方式、會話的事務(wù)邊界、Broker 的持久化策略。每一層都直接影響消息的可靠性但要組合起來才有意義。5.1 確認(rèn)模式消費(fèi)者收完之后Broker 才敢刪消息主動和異步監(jiān)聽之外Session 的確認(rèn)模式?jīng)Q定了消息何時(shí)被“算作處理成功”。在 JMS 1.1 里有四種常見模式模式行為適用場景AUTO_ACKNOWLEDGE消費(fèi)者收到消息后自動確認(rèn)receive 返回或 onMessage 返回時(shí)業(yè)務(wù)處理邏輯簡單、處理失敗可以接受消息丟失的場景CLIENT_ACKNOWLEDGE需要手動調(diào)用 message.acknowledge() 確認(rèn)處理流程較復(fù)雜希望確認(rèn)時(shí)機(jī)由業(yè)務(wù)代碼控制的場景DUPS_OK_ACKNOWLEDGE允許重復(fù)消息延遲確認(rèn)性能好可容忍重復(fù)消費(fèi)的統(tǒng)計(jì)類場景SESSION_TRANSACTED配合事務(wù)會話使用需要保證消息收發(fā)與業(yè)務(wù)操作原子一致的場景最常見的坑是使用 AUTO_ACKNOWLEDGE但是消費(fèi)者在onMessage里處理到一半拋異常。此時(shí)消息實(shí)際上已經(jīng)被確認(rèn)了Broker 會把它刪除重啟消費(fèi)者后這條消息不會再出現(xiàn)處理自然就丟失了。正確的做法是——如果處理失敗需要重試要么用 CLIENT_ACKNOWLEDGE 模式收到消息后先不確認(rèn)處理成功再調(diào)用message.acknowledge()要么開啟事務(wù)處理失敗就session.rollback()。CLIENT_ACKNOWLEDGE 模式有一個(gè)細(xì)節(jié)很容易忽略acknowledge()不只是確認(rèn)當(dāng)前這條消息而是確認(rèn)這個(gè) Session 里“已經(jīng)消費(fèi)但尚未確認(rèn)”的所有消息。如果同時(shí)消費(fèi)了多條且逐條處理調(diào)用一次 acknowledge 會把這批全確認(rèn)了。要精確控制單條消息必須結(jié)合事務(wù)會話每個(gè)事務(wù)處理一條就 commit 一次。5.2 事務(wù)會話讓發(fā)送和接收具備原子性connection.createSession(true, Session.SESSION_TRANSACTED)開啟了事務(wù)模式。事務(wù)模式下生產(chǎn)者的發(fā)送動作不會立即生效而是要等session.commit()才真正把消息提交給 Broker消費(fèi)者的接收動作也不會立即被確認(rèn)而是等 commit 時(shí)才確認(rèn)這一批消息。如果業(yè)務(wù)處理失敗調(diào)用session.rollback()消息會被回滾之后可以再次消費(fèi)。事務(wù)會話對“發(fā)送”和“接收”是獨(dú)立的語義。比如在一個(gè)服務(wù)里先接收消息 A處理后發(fā)送消息 B然后 commit。這意味著 A 的消費(fèi)確認(rèn)和 B 的發(fā)送是同時(shí)生效的。這個(gè)組合非常實(shí)用相當(dāng)于把“業(yè)務(wù)成功了才把消息標(biāo)記為已處理”這一個(gè)要求交給中間件來保證。不過事務(wù)也不能濫用。事務(wù)越大、批量越大Broker 的鎖競爭和日志壓力越大。經(jīng)驗(yàn)值是一個(gè)事務(wù)里處理的業(yè)務(wù)消息量控制在幾十條以內(nèi)尤其在高并發(fā)場景下過大的事務(wù)反而會成為瓶頸。5.3 持久化消息存儲到磁盤進(jìn)程重啟才能恢復(fù)ActiveMQ 默認(rèn)將消息保存在內(nèi)存里同時(shí)提供 KahaDB 作為默認(rèn)持久化方案它會將消息寫入本地磁盤。對生產(chǎn)環(huán)境來說強(qiáng)烈建議開啟持久化并且生產(chǎn)者發(fā)送時(shí)設(shè)置DeliveryMode.PERSISTENT?？梢赃@樣理解二者的關(guān)系持久化確保 Broker 重啟后消息還在PERSISTENT 發(fā)送模式確保每條消息都走持久化存儲。即使 Broker 配置了 KahaDB如果生產(chǎn)者發(fā)送時(shí)用的是非持久化模式消息還是可能因?yàn)?Broker 崩潰而丟失。ActiveMQ 的主配置文件中persistenceAdapter的部分默認(rèn)就是 KahaDBpersistenceAdapter kahaDB directory${activemq.data}/kahadb/ /persistenceAdapter學(xué)習(xí)階段可以刻意做一次“重啟測試”發(fā)送一條持久化消息但消費(fèi)者不啟動然后重啟 Broker再啟動消費(fèi)者觀察消息是否還在。我第一次測的時(shí)候發(fā)現(xiàn)消息還在其實(shí)很不解——后來意識到這就是持久化在起作用。5.4 組合起來可靠消息的生產(chǎn)端配置怎么開把三層機(jī)制組合起來一個(gè)穩(wěn)妥、可靠的生產(chǎn)者配置看起來是這樣Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); producer.setTimeToLive(60000); // 消息有效期為 1 分鐘 producer.send(message);消費(fèi)者側(cè)如果對處理結(jié)果有嚴(yán)格成功要求通常選擇事務(wù)會話Session session connection.createSession(true, Session.SESSION_TRANSACTED); MessageConsumer consumer session.createConsumer(queue); TextMessage msg (TextMessage) consumer.receive(); // 處理業(yè)務(wù)邏輯比如更新數(shù)據(jù)庫、調(diào)用外部接口 // 都成功后提交 session.commit(); // 失敗則回滾 session.rollback();這里涉及一個(gè)進(jìn)階考量如果消費(fèi)者處理到一半進(jìn)程崩了事務(wù)尚未 commitBroker 會重新投遞這條消息。重新投遞可能導(dǎo)致重復(fù)消費(fèi)所以消費(fèi)者的業(yè)務(wù)邏輯必須做冪等。消息中間件本身并不能徹底避免重復(fù)投遞它只能保證“至少一次”不能保證“正好一次”。這句話我建議所有初學(xué)者刻在腦子里。6. 學(xué)習(xí)過程中實(shí)際踩過的四個(gè)坑這部分是我最想寫的。Demo 都能跑通但生產(chǎn)要素一進(jìn)來問題就花樣百出。下面四個(gè)坑都是我真實(shí)遇到過的按照排查鏈路寫出來方便你以后對著復(fù)現(xiàn)。6.1 第一個(gè)坑啟動 Broker 就報(bào)端口占用現(xiàn)象執(zhí)行./activemq start后控制臺報(bào)java.net.BindException: Address already in use: JVM_Bind。排查我把堆棧里暴露的端口號 61616 用lsof -i:61616查了下發(fā)現(xiàn)是本機(jī)之前跑過一個(gè)沒關(guān)干凈的 ActiveMQ 實(shí)例。這提醒我兩個(gè)習(xí)慣一是問題定位要先確認(rèn)端口占用而不是盲目重裝二是啟動前用./activemq stop或jps查看已有 Java 進(jìn)程別把多個(gè) broker 實(shí)例混在一起跑。如果你是在同一臺機(jī)器上跑多個(gè) broker 做集群學(xué)習(xí)必須修改conf/activemq.xml里的 transportConnector 端口還要調(diào)整conf/jetty.xml里的管理端口否則必然沖突。6.2 第二個(gè)坑隊(duì)列里明明有消息消費(fèi)者卻收不到現(xiàn)象生產(chǎn)者發(fā)送成功控制臺里Number Of Pending Messages是 5但消費(fèi)者啟動后一條都沒收到。排查過程我先確認(rèn)消費(fèi)者確實(shí)連接上了 broker——控制臺的 Connections 頁面能看到消費(fèi)者連接。然后檢查消費(fèi)的隊(duì)列名發(fā)現(xiàn)生產(chǎn)者和消費(fèi)者用的隊(duì)列名不一致。這個(gè)屬于低級錯(cuò)誤。但還有一個(gè)更隱蔽的原因困擾了我更久消費(fèi)者連接沒調(diào)用connection.start()。我照著網(wǎng)上某篇文章只創(chuàng)建了 connection 和 session沒有顯式調(diào)用 start。這個(gè)操作在“異步監(jiān)聽 等待”的例子中看起來沒問題但實(shí)際消息分發(fā)一直沒有真正開始。補(bǔ)上connection.start()后一切正常。這個(gè)教訓(xùn)讓我從此養(yǎng)成了一個(gè)習(xí)慣JMS 樣板代碼一定逐行核驗(yàn)別漏掉生命周期方法。6.3 第三個(gè)坑發(fā)布訂閱模式下“先啟動生產(chǎn)者后啟動消費(fèi)者”丟消息現(xiàn)象我用 Topic 建模一個(gè)廣播消息先運(yùn)行生產(chǎn)者發(fā)了兩條消息再啟動消費(fèi)者結(jié)果消息一條都沒收到。排查過程這其實(shí)是 Topic 模型的正常行為不是 bug。非持久訂閱者不會收到訂閱之前的消息。我當(dāng)時(shí)誤以為 Topic 和 Queue 一樣消息會先積壓在 broker 里消費(fèi)者一來就全推給它。搞清楚這個(gè)機(jī)制后我重新設(shè)計(jì)了一版客戶端先啟動消費(fèi)者并保持在線再觸發(fā)生產(chǎn)者發(fā)送消息就正常收到了。如果業(yè)務(wù)上確實(shí)需要“離線訂閱不丟消息”就要在消費(fèi)端使用createDurableSubscriber。注意持久訂閱的訂閱 ID 一旦創(chuàng)建就不能隨意變更否則舊的持久訂閱消息將無法再被消費(fèi)會一直堆積在 broker 中。6.4 第四個(gè)坑重復(fù)消費(fèi)引發(fā)的臟數(shù)據(jù)問題現(xiàn)象消費(fèi)者消費(fèi)消息后更新了數(shù)據(jù)庫但進(jìn)程在數(shù)據(jù)庫事務(wù)提交后、Session 提交前崩潰了重啟后同一消息被再次消費(fèi)數(shù)據(jù)庫被更新了兩次。排查過程當(dāng)時(shí)我用了事務(wù)會話以為有事務(wù)就萬事大吉。實(shí)際上事務(wù)會話保證的是“消息確認(rèn)和業(yè)務(wù)處理的原子性”但“業(yè)務(wù)處理”本身如果不在同一個(gè)數(shù)據(jù)庫事務(wù)里狀態(tài)就無法對賬。后來我用了一個(gè)冪等方案消息體帶業(yè)務(wù)主鍵消費(fèi)者在處理前先查一下數(shù)據(jù)庫是否已有對應(yīng)記錄有就直接返回沒有才執(zhí)行業(yè)務(wù)并落庫且主鍵唯一索引兜底。這樣即使消息重復(fù)投遞也不會產(chǎn)生重復(fù)數(shù)據(jù)。這個(gè)坑給我最大的啟發(fā)是消息中間件只保證傳輸層面的至少一次投遞消費(fèi)端必須自己做好冪等。你以為中間件幫你解決了所有問題它只是幫你把問題限定在了一個(gè)可控范圍內(nèi)剩下的還要靠自己的業(yè)務(wù)設(shè)計(jì)。7. 學(xué)習(xí)收尾時(shí)的一些心得過完前面這六個(gè)章節(jié)后我又重新回頭看了最初促使我開始學(xué)習(xí)的那個(gè)“同步調(diào)用雪崩”場景。用上 ActiveMQ 之后其實(shí)還沒有結(jié)束還配套做了消費(fèi)并發(fā)度控制、數(shù)據(jù)庫連接數(shù)控制、失敗重試隊(duì)列、監(jiān)控告警才敢說這個(gè)鏈路基本可靠了。這個(gè)認(rèn)知我覺得比學(xué)會 API 本身更重要消息中間件解決的是通信模式問題但一個(gè)系統(tǒng)的穩(wěn)定性永遠(yuǎn)要靠整體設(shè)計(jì)不會因?yàn)橐胍粋€(gè)組件就自動變好。如果你正在學(xué) JMS 和 ActiveMQ我給三條很實(shí)在的建議。第一條不要只看著我的代碼抄一定自己把生產(chǎn)者、消費(fèi)者、管理控制臺三樣對起來跑通一個(gè)完整的收發(fā)。第二條把 Queue 和 Topic 的區(qū)別多做幾個(gè)實(shí)驗(yàn)包括離線訂閱、多個(gè)消費(fèi)者消費(fèi)同一個(gè)隊(duì)列、持久訂閱恢復(fù)這些體驗(yàn)比背概念管用太多。第三條第一次接觸消息可靠性的時(shí)候強(qiáng)制自己回答“消息什么時(shí)候會丟、重復(fù)消費(fèi)怎么兜底、中間件重啟后消息還在不在”這三個(gè)問題回答完再繼續(xù)深入事務(wù)和持久化否則知識很容易停留在 API 層面。最后分享一個(gè)小技巧學(xué)習(xí)期間可以在 ActiveMQ 的配置里臨時(shí)把日志級別調(diào)成 DEBUG重點(diǎn)看 messages 相關(guān)日志和 session 相關(guān)日志。以前我一直覺得中間件日志太多不看后來排查那兩次詭異問題時(shí)DEBUG 日志里的 broker 內(nèi)部流轉(zhuǎn)信息幫了大忙。日志是排查問題最早出手的武器不要等到控制臺里消息都詭異了才想起來翻日志。 JMS與ActiveMQ實(shí)戰(zhàn)筆記：從消息模型到可靠消息機(jī)制
我第一次正兒八經(jīng)看 JMS 和 ActiveMQ并不是因?yàn)橐嬖嚤嘲斯啥琼?xiàng)目里兩個(gè)服務(wù)之間的同步調(diào)用把數(shù)據(jù)庫連接池打爆了。上游接口一個(gè)超時(shí)下游跟著雪崩運(yùn)維的告警群半夜響成一片。翻來覆去折騰了幾天之后同事丟給我一句話“用消息中間件把調(diào)用拆開吧先看看 JMS 和 ActiveMQ?！碑?dāng)時(shí)我連 JMS 和 MQ 的關(guān)系都說不清楚硬著頭皮把那套消息模型啃了一遍才有了這篇“JMS、ActiveMQ 學(xué)習(xí)一則”。這篇筆記適合兩類人一是剛開始接觸消息中間件、被一堆 JMS API 術(shù)語搞暈的開發(fā)者二是能把 Demo 跑通但不清楚 Queue、Topic、ACK、持久化到底意味著什么的人。我會把規(guī)范、代碼、踩坑都放在一起講盡量還原當(dāng)時(shí)的學(xué)習(xí)路徑也把網(wǎng)絡(luò)上那些講得含糊的地方補(bǔ)完整。1. 為什么我先啃 JMS 而不是直接學(xué)某個(gè) MQ 客戶端1.1 消息中間件解決的核心問題把“同步等待”變成“異步解耦”先說我當(dāng)時(shí)的場景。有個(gè)訂單創(chuàng)建服務(wù)處理完訂單之后要通知庫存服務(wù)扣庫存、通知積分服務(wù)加積分、通知短信服務(wù)發(fā)通知。最開始寫的代碼很“樸素”——A 服務(wù)直接用 HTTP 調(diào)用 B 服務(wù)同步等返回。看起來邏輯很直白但一壓測就出問題任何一個(gè)下游服務(wù)變慢訂單接口的響應(yīng)時(shí)間立刻被拖垮下游服務(wù)一重啟上游的調(diào)用直接報(bào)超時(shí)訂單雖然建成功了但庫存、積分、短信全部沒執(zhí)行數(shù)據(jù)就錯(cuò)亂了。消息中間件解決的就是這種“你完成你的我完成我的然后把變化告訴別人”的通信問題。訂單服務(wù)把“訂單已創(chuàng)建”這個(gè)事實(shí)寫到消息隊(duì)列里訂單服務(wù)的接口直接返回成功庫存、積分、短信服務(wù)各自從隊(duì)列里取消息按自己的節(jié)奏處理。這樣三個(gè)下游服務(wù)誰慢、誰掛都不影響訂單主流程。這個(gè)模型里有幾個(gè)術(shù)語你得先建立直覺Broker是中間件服務(wù)器負(fù)責(zé)消息的存儲和轉(zhuǎn)發(fā)Producer是消息的生產(chǎn)者Consumer是消息的消費(fèi)者。消息不是直接給某個(gè)消費(fèi)者的而是先放到 Broker 的某個(gè)“容器”里消費(fèi)者自己來取。這個(gè)“容器”就是后來的 Queue 或 Topic。1.2 JMS 是規(guī)范ActiveMQ 是實(shí)現(xiàn)先理解這對關(guān)系JMSJava Message Service是 Java 平臺制定的一套消息服務(wù)接口規(guī)范它規(guī)定了 Java 程序怎么創(chuàng)建連接、怎么發(fā)消息、怎么收消息但 JMS 自己不干活真正干活的是符合這套規(guī)范的 provider比如 Apache ActiveMQ、開源的 OpenMQ 等。這個(gè)關(guān)系有點(diǎn)像 JDBC 和 MySQL 驅(qū)動的關(guān)系——JDBC 規(guī)定了你用什么方法連接數(shù)據(jù)庫MySQL 驅(qū)動負(fù)責(zé)把這些方法翻譯成 MySQL 能懂的協(xié)議。這就解釋了為什么學(xué)習(xí)的時(shí)候要先看 JMS 再看 ActiveMQ如果你只對著 ActiveMQ 的 API 抄一遍換個(gè) MQ 你就廢了如果你把 JMS 的核心模型理解了接觸任何一個(gè)兼容 JMS 的中間件都能很快上手。我當(dāng)初踩過一個(gè)認(rèn)知誤區(qū)以為 ActiveMQ 提供了比 JMS 多得多的高級功能所以直接按 ActiveMQ 的文檔寫就行。實(shí)際上 ActiveMQ 對 JMS 規(guī)范做了很完整的實(shí)現(xiàn)它的核心 API 就是 JMS API你寫代碼時(shí) import 的主要是javax.jms.*這個(gè)包而不是org.apache.activemq.*。后者只在構(gòu)造連接工廠時(shí)才會用到。1.3 什么場景才需要引入消息中間件先潑盆冷水消息中間件不是銀彈學(xué)習(xí)過程中我也遇到過濫用它的團(tuán)隊(duì)。一個(gè)內(nèi)部查詢接口沒有任何瞬時(shí)流量壓力硬是套一層消息隊(duì)列結(jié)果多了一個(gè)運(yùn)維節(jié)點(diǎn)、多了一段延遲、多了一堆消息亂序問題。什么時(shí)候才值得引入我的判斷標(biāo)準(zhǔn)有三條一是上下游節(jié)奏不一致下游來不及同步處理需要用隊(duì)列做緩沖二是下游故障不應(yīng)該影響上游主流程需要隔離故障三是同一份業(yè)務(wù)事件需要被多個(gè)系統(tǒng)各自消費(fèi)比如訂單創(chuàng)建之后庫存、積分、日志都要處理。反過來說如果只是簡單的 A 調(diào) B響應(yīng)時(shí)間敏感且 B 很少故障那直接 HTTP 調(diào)用反而更合適。這部分認(rèn)知雖然當(dāng)時(shí)不成熟但后來回看幫我在選型時(shí)規(guī)避了不少坑。2. JMS 核心模型五個(gè)抽象角色串起全部消息邏輯JMS 1.1 里抽象出的角色其實(shí)就五個(gè)ConnectionFactory、Connection、Session、Destination、MessageProducer/MessageConsumer。剩下的概念比如 Message、AcknowledgeMode 都是在這五者之上附加的。把這五個(gè)角色串成一個(gè)故事你就能看懂 90% 的 JMS 代碼。2.1 ConnectionFactory創(chuàng)建連接的唯一入口ConnectionFactory 是一個(gè)連接工廠接口它封裝了與某個(gè) Broker 建立連接所需的參數(shù)。在 ActiveMQ 中最常見的是ActiveMQConnectionFactory構(gòu)造時(shí)傳入 Broker 的地址典型格式是tcp://127.0.0.1:61616。ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616);這個(gè)對象通常是重量級對象創(chuàng)建成本高所以應(yīng)該復(fù)用不要每次發(fā)消息都 new 一個(gè)。我看到不少剛?cè)腴T的同學(xué)把ActiveMQConnectionFactory寫在方法內(nèi)部短時(shí)間大量請求一來連接資源就爆炸了。正確做法是使用連接池或至少是單例持有。2.2 Connection 與 Session消息通信的骨架從工廠拿到Connection之后需要顯式調(diào)用connection.start()這時(shí)候連接才真正建立。然后從 Connection 創(chuàng)建 Session。Session 是一個(gè)非常重要的中間層消息的生產(chǎn)和消費(fèi)都是基于 Session 的Session 的創(chuàng)建方式?jīng)Q定了本次會話的確認(rèn)模式和事務(wù)邊界。Connection connection factory.createConnection(); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);createSession的第一個(gè)參數(shù)是事務(wù)標(biāo)志第二參數(shù)是確認(rèn)模式。在 JMS 1.1 中如果第一個(gè)參數(shù)傳 true第二個(gè)參數(shù)會被忽略因?yàn)槭聞?wù)模式下發(fā)送和接收都通過commit和rollback控制。剛開始不用懵理解成“Session 是一段通信上下文凡是這次會話內(nèi)的消息都可以在這個(gè)上下文里標(biāo)記狀態(tài)”就行。2.3 Destination消息要發(fā)往的地址Destination 是抽象類實(shí)際子類要么是Queue要么是Topic。Queue 對應(yīng)點(diǎn)對點(diǎn)模型Topic 對應(yīng)發(fā)布訂閱模型。在代碼里session.createQueue()或session.createTopic()創(chuàng)建的只是邏輯地址不需要在 Broker 側(cè)預(yù)先建好ActiveMQ 會在第一次使用時(shí)自動創(chuàng)建。Queue queue session.createQueue(DEMO.ORDER); Topic topic session.createTopic(DEMO.NEWS);這里有個(gè)細(xì)節(jié)隊(duì)列名稱建議使用“域名前綴 業(yè)務(wù)名”的格式比如ORDER.CREATED、STOCK.DEDUCTED為了在管理后臺一眼能看出用途。我見過用queue1、test2這樣的名字上生產(chǎn)的后來排查問題差點(diǎn)被隊(duì)友打死。2.4 MessageProducer 與 MessageConsumer實(shí)際干活的兩個(gè)角色生產(chǎn)者和消費(fèi)者都在 Session 上創(chuàng)建。生產(chǎn)者創(chuàng)建時(shí)可以指定默認(rèn)目的地也可以不指定發(fā)送時(shí)再傳目的地。生產(chǎn)者還支持設(shè)置是否持久化、優(yōu)先級、存活時(shí)間等特性這部分會影響消息的存儲行為后面展開。MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); TextMessage message session.createTextMessage(hello jms); producer.send(message);消費(fèi)者既可以主動拉取消息也可以注冊消息監(jiān)聽器異步接收消息。主動拉取用consumer.receive()它會阻塞等待可以傳入超時(shí)時(shí)間異步監(jiān)聽用MessageListener接口Broker 一有消息就回調(diào)onMessage方法。2.5 Message 結(jié)構(gòu)消息頭、屬性、消息體三層分工JMS 的消息對象由三部分組成。第一個(gè)是消息頭header存放 JMS 規(guī)范定義的固定字段比如JMSDestination、JMSMessageID、JMSPriority、JMSExpiration。第二個(gè)是消息屬性properties可以看成附加的元數(shù)據(jù)支持字符串、數(shù)字、布爾等類型常用在消息篩選和路由上。第三個(gè)是消息體body存放業(yè)務(wù)數(shù)據(jù)本身。消息體有幾種類型TextMessage適合傳輸文本和 JSON 字符串BytesMessage適合二進(jìn)制數(shù)據(jù)MapMessage適合鍵值對結(jié)構(gòu)ObjectMessage適合序列化對象但跨語言不友好StreamMessage用得最少。實(shí)際項(xiàng)目里 80% 的流量都是TextMessage因?yàn)闃I(yè)務(wù)方一般約定用 JSON 傳結(jié)構(gòu)化數(shù)據(jù)不會強(qiáng)行依賴 Java 對象序列化。我自己的習(xí)慣是消息體只放必要的業(yè)務(wù)數(shù)據(jù)別把整個(gè)數(shù)據(jù)庫實(shí)體塞進(jìn)去。因?yàn)橄⒖赡鼙欢鄠€(gè)系統(tǒng)消費(fèi)每個(gè)系統(tǒng)關(guān)心字段不一樣塞整個(gè)實(shí)體不僅體積大而且上游實(shí)體變更會導(dǎo)致下游反序列化失敗。3. ActiveMQ 落地實(shí)操從下載啟動到第一個(gè)消息跑通理論講完了得有真家伙。我當(dāng)時(shí)本地環(huán)境是 JDK 8下載的是 Apache ActiveMQ 的穩(wěn)定發(fā)行版。注意 ActiveMQ 有兩個(gè)分支一個(gè)叫 ActiveMQ Classic一個(gè)叫 ActiveMQ Artemis兩者都是 ActiveMQ但 Artemis 是新一代架構(gòu)API 上對 JMS 規(guī)范的支持更現(xiàn)代。作為學(xué)習(xí) JMS 的入口我建議用 Classic因?yàn)橘Y料多、踩坑記錄全先把規(guī)范玩明白。3.1 啟動 Broker十分鐘確認(rèn)環(huán)境下載解壓后進(jìn)入 bin 目錄Linux/macOS 下執(zhí)行./activemq startWindows 下執(zhí)行activemq.bat start。啟動完成后默認(rèn)監(jiān)聽以下幾個(gè)重要端口端口用途61616JMS 客戶端連接的 OpenWire 端口8161Web 管理控制臺端口5672AMQP 協(xié)議端口默認(rèn)也開啟打開瀏覽器訪問http://127.0.0.1:8161/默認(rèn)登錄賬號是 admin密碼也是 admin進(jìn)去就能看到隊(duì)列列表、Topic 列表、連接數(shù)、消息統(tǒng)計(jì)等。我強(qiáng)烈建議學(xué)習(xí)階段養(yǎng)成盯控制臺的習(xí)慣很多代碼“看起來沒反應(yīng)”的問題在控制臺里一眼就能看出來——消息到底有沒有進(jìn)來、被誰消費(fèi)了、有沒有進(jìn)入死信。3.2 第一個(gè) JMS 生產(chǎn)者發(fā)送一條持久化文本消息下面這端代碼是我當(dāng)時(shí)跑通的第一個(gè)生產(chǎn)者。為了方便閱讀我把異常處理簡化了學(xué)習(xí)階段先跑通生產(chǎn)上請務(wù)必正確封裝資源關(guān)閉邏輯。import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class QueueProducer { public static void main(String[] args) throws Exception { // 1. 連接工廠指向本地 Broker ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); // 2. 創(chuàng)建連接并啟動 Connection connection factory.createConnection(); connection.start(); // 3. 創(chuàng)建會話非事務(wù) 自動確認(rèn) Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 創(chuàng)建目的地 Queue queue session.createQueue(DEMO.ORDER); // 5. 創(chuàng)建生產(chǎn)者 MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 6. 發(fā)送一條文本消息 TextMessage textMessage session.createTextMessage(訂單已創(chuàng)建訂單號20250101); producer.send(textMessage); System.out.println(已發(fā)送消息 textMessage.getText()); // 7. 關(guān)閉資源 producer.close(); session.close(); connection.close(); } }這個(gè)例子里最容易忽略的是connection.start()。有的教程會在創(chuàng)建生產(chǎn)者之前就調(diào)用 start有的會在發(fā)送之后才調(diào)用。start()本身不是發(fā)送動作的開關(guān)但它會讓連接進(jìn)入可發(fā)送/接收狀態(tài)。如果你只創(chuàng)建連接不 start消息照樣能發(fā)出去但消費(fèi)者端不 start 的話就一直收不到消息。我當(dāng)初就踩過這個(gè)坑后面專門講。3.3 第一個(gè) JMS 消費(fèi)者同步接收和異步監(jiān)聽兩種方式消費(fèi)者有兩種寫法我建議都試一遍。第一種是主動 receive適合測試和批處理任務(wù)。import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class QueueConsumer { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); Connection connection factory.createConnection(); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(DEMO.ORDER); MessageConsumer consumer session.createConsumer(queue); // 阻塞 3 秒沒消息會超時(shí)返回 null Message message consumer.receive(3000); if (message instanceof TextMessage) { System.out.println(收到消息 ((TextMessage) message).getText()); } consumer.close(); session.close(); connection.close(); } }第二種是注冊監(jiān)聽器異步接收實(shí)際項(xiàng)目里這種用得更多因?yàn)樗粫枞麡I(yè)務(wù)線程。示例代碼我放在下面consumer.setMessageListener(new MessageListener() { Override public void onMessage(Message message) { if (message instanceof TextMessage) { try { System.out.println(異步收到消息 ((TextMessage) message).getText()); } catch (JMSException e) { e.printStackTrace(); } } } }); Thread.sleep(5000);這里有個(gè)容易翻車的地方Thread.sleep(5000)是必要的。因?yàn)閙ain方法執(zhí)行完后 Java 進(jìn)程就退出了消費(fèi)者監(jiān)聽器注冊完還沒來得及收消息進(jìn)程就沒了。看起來像是“監(jiān)聽器沒生效”其實(shí)只是主線程提前退出。生產(chǎn)環(huán)境一般由應(yīng)用服務(wù)器或 Spring 容器持有消費(fèi)者生命周期不存在這個(gè)問題。3.4 跑完之后去控制臺觀察四個(gè)維度第一個(gè)消息鏈路跑通后我不急著寫第二個(gè) Demo而是去管理控制臺把所有變化對照了一遍。四個(gè)維度最有價(jià)值Number Of Pending Messages隊(duì)列里待消費(fèi)的消息數(shù)量。如果這個(gè)數(shù)字一直漲說明消費(fèi)端有問題。Number Of Consumers當(dāng)前有多少消費(fèi)者掛著。Messages Enqueued / Dequeued入隊(duì)和出隊(duì)的累計(jì)數(shù)能看出積壓到底是多少。Blocked Time連接被阻塞的時(shí)間如果很大說明消費(fèi)者處理不過來。先建立這套觀察手段后面排坑會省很多時(shí)間。空口說“消息丟了”“隊(duì)列卡了”很容易被運(yùn)維懟把控制臺截圖甩過去才是正確的溝通方式。4. 隊(duì)列模式和發(fā)布訂閱模式看起來代碼差不多語義差遠(yuǎn)了JMS 里最繞、也最影響選型的就是兩種消息模型點(diǎn)對點(diǎn)模式Queue和發(fā)布訂閱模式Topic。很多初學(xué)者能分別跑通但沒搞明白為什么需要兩種。4.1 Queue一條消息只被一個(gè)消費(fèi)者取走點(diǎn)對點(diǎn)模型下一條消息只會被一個(gè)消費(fèi)者成功接收。你可以啟動多個(gè)消費(fèi)者實(shí)例同時(shí)消費(fèi)同一個(gè)隊(duì)列ActiveMQ 會自動做負(fù)載均衡把不同消息分發(fā)給不同的消費(fèi)者。這像是一堆快遞包裹放在驛站多個(gè)配送員各自拿走一部分每個(gè)包裹最終只會被一個(gè)配送員取走。這個(gè)模型適合“任務(wù)分發(fā)”場景比如一批待處理文件10 個(gè) worker 搶占處理比如訂單狀態(tài)變更庫存服務(wù)和積分服務(wù)如果都要處理就不能用同一個(gè) Queue因?yàn)橄⒅唤o一個(gè)人。這里我補(bǔ)充一個(gè)容易誤解的點(diǎn)多個(gè)消費(fèi)者消費(fèi)同一個(gè)隊(duì)列和 JMS 1.1 里的 QueueReceiver 概念有細(xì)微差別。如果兩個(gè)消費(fèi)者的selector條件互斥消息可能會按條件被分流但默認(rèn)無 selector 時(shí)就是平均分發(fā)不是廣播。4.2 Topic一條消息廣播給所有訂閱者發(fā)布訂閱模型下一條消息會被所有“當(dāng)前活躍訂閱者”都收到。這像是廣播喇叭誰在聽誰就能聽見。Topic 模式下沒有“競爭消費(fèi)”的概念消費(fèi)者訂閱同一個(gè) Topic各自都能收到全量消息。但 Topic 有一個(gè)關(guān)鍵限制默認(rèn)情況下非持久訂閱者只在連接期間且保持訂閱狀態(tài)時(shí)才能收到消息。如果消費(fèi)者先掛了生產(chǎn)者后發(fā)消息這個(gè)消費(fèi)者重啟之后是收不到“離線期間”的消息的。這讓我當(dāng)初很意外因?yàn)樗?Queue 的行為明顯不一樣——隊(duì)列里消費(fèi)者掛多久都不怕消息會積壓著但 Topic 不會。要解決離線訂閱丟失的問題JMS 提供了持久訂閱Durable Subscription。持久訂閱狀態(tài)下Broker 會為這個(gè)訂閱者保存未消費(fèi)的消息等它上線后再推送。4.3 代碼對比一條線上隊(duì)列一條線上廣播Queue 和 Topic 的代碼結(jié)構(gòu)幾乎一致只有兩行不同——目的地對象和創(chuàng)建方式。// 消費(fèi)者隊(duì)列模式 Queue queue session.createQueue(DEMO.ORDER); MessageConsumer consumer session.createConsumer(queue);// 消費(fèi)者Topic 模式非持久訂閱 Topic topic session.createTopic(DEMO.NEWS); MessageConsumer consumer session.createConsumer(topic);持久訂閱的消費(fèi)者寫法會多一點(diǎn)Topic topic session.createTopic(DEMO.NEWS); // 需要為訂閱者設(shè)置唯一名稱 MessageConsumer consumer session.createDurableSubscriber(topic, news-consumer-1);createDurableSubscriber第二個(gè)參數(shù)是訂閱 IDBroker 用這個(gè) ID 區(qū)分不同的持久訂閱者。如果你用相同的訂閱 ID 連接兩次第二次會替代第一次的訂閱關(guān)系。發(fā)送端完全一樣MessageProducer producer session.createProducer(topic); producer.send(session.createTextMessage(財(cái)經(jīng)早報(bào)));4.4 選型判斷別讓“感覺好用”替你做決定學(xué)習(xí)期間我總結(jié)了一個(gè)很簡單的判斷規(guī)則同一業(yè)務(wù)事件如果多個(gè)系統(tǒng)需要各自獨(dú)立處理用 Topic如果一個(gè)任務(wù)的多個(gè)實(shí)例競爭處理同一批消息用 Queue。舉兩個(gè)對比場景訂單創(chuàng)建后觸發(fā)庫存扣減、積分增加、短信通知這是“一魚多吃”三個(gè)系統(tǒng)各自消費(fèi)同一事件應(yīng)該用 Topic 或給每個(gè)消費(fèi)者建獨(dú)立 Queue。秒殺活動里有 50 萬條搶購請求后面有 10 臺冪等處理服務(wù)實(shí)例消化這是“任務(wù)競爭”應(yīng)該用 Queue。更常見的企業(yè)做法是“混合使用”用 Topic 接收領(lǐng)域事件然后每個(gè)訂閱者內(nèi)部再用 Queue 做異步削峰。初學(xué)階段先別追求花哨能把 Queue 和 Topic 的邊界吃透就超過了大多數(shù)人。5. 可靠消息的底層機(jī)制確認(rèn)、事務(wù)、持久化一次講清消息中間件最常被問的一句話是“消息到底會不會丟”答案取決于三層配置消費(fèi)者的確認(rèn)方式、會話的事務(wù)邊界、Broker 的持久化策略。每一層都直接影響消息的可靠性但要組合起來才有意義。5.1 確認(rèn)模式消費(fèi)者收完之后Broker 才敢刪消息主動和異步監(jiān)聽之外Session 的確認(rèn)模式?jīng)Q定了消息何時(shí)被“算作處理成功”。在 JMS 1.1 里有四種常見模式模式行為適用場景AUTO_ACKNOWLEDGE消費(fèi)者收到消息后自動確認(rèn)receive 返回或 onMessage 返回時(shí)業(yè)務(wù)處理邏輯簡單、處理失敗可以接受消息丟失的場景CLIENT_ACKNOWLEDGE需要手動調(diào)用 message.acknowledge() 確認(rèn)處理流程較復(fù)雜希望確認(rèn)時(shí)機(jī)由業(yè)務(wù)代碼控制的場景DUPS_OK_ACKNOWLEDGE允許重復(fù)消息延遲確認(rèn)性能好可容忍重復(fù)消費(fèi)的統(tǒng)計(jì)類場景SESSION_TRANSACTED配合事務(wù)會話使用需要保證消息收發(fā)與業(yè)務(wù)操作原子一致的場景最常見的坑是使用 AUTO_ACKNOWLEDGE但是消費(fèi)者在onMessage里處理到一半拋異常。此時(shí)消息實(shí)際上已經(jīng)被確認(rèn)了Broker 會把它刪除重啟消費(fèi)者后這條消息不會再出現(xiàn)處理自然就丟失了。正確的做法是——如果處理失敗需要重試要么用 CLIENT_ACKNOWLEDGE 模式收到消息后先不確認(rèn)處理成功再調(diào)用message.acknowledge()要么開啟事務(wù)處理失敗就session.rollback()。CLIENT_ACKNOWLEDGE 模式有一個(gè)細(xì)節(jié)很容易忽略acknowledge()不只是確認(rèn)當(dāng)前這條消息而是確認(rèn)這個(gè) Session 里“已經(jīng)消費(fèi)但尚未確認(rèn)”的所有消息。如果同時(shí)消費(fèi)了多條且逐條處理調(diào)用一次 acknowledge 會把這批全確認(rèn)了。要精確控制單條消息必須結(jié)合事務(wù)會話每個(gè)事務(wù)處理一條就 commit 一次。5.2 事務(wù)會話讓發(fā)送和接收具備原子性connection.createSession(true, Session.SESSION_TRANSACTED)開啟了事務(wù)模式。事務(wù)模式下生產(chǎn)者的發(fā)送動作不會立即生效而是要等session.commit()才真正把消息提交給 Broker消費(fèi)者的接收動作也不會立即被確認(rèn)而是等 commit 時(shí)才確認(rèn)這一批消息。如果業(yè)務(wù)處理失敗調(diào)用session.rollback()消息會被回滾之后可以再次消費(fèi)。事務(wù)會話對“發(fā)送”和“接收”是獨(dú)立的語義。比如在一個(gè)服務(wù)里先接收消息 A處理后發(fā)送消息 B然后 commit。這意味著 A 的消費(fèi)確認(rèn)和 B 的發(fā)送是同時(shí)生效的。這個(gè)組合非常實(shí)用相當(dāng)于把“業(yè)務(wù)成功了才把消息標(biāo)記為已處理”這一個(gè)要求交給中間件來保證。不過事務(wù)也不能濫用。事務(wù)越大、批量越大Broker 的鎖競爭和日志壓力越大。經(jīng)驗(yàn)值是一個(gè)事務(wù)里處理的業(yè)務(wù)消息量控制在幾十條以內(nèi)尤其在高并發(fā)場景下過大的事務(wù)反而會成為瓶頸。5.3 持久化消息存儲到磁盤進(jìn)程重啟才能恢復(fù)ActiveMQ 默認(rèn)將消息保存在內(nèi)存里同時(shí)提供 KahaDB 作為默認(rèn)持久化方案它會將消息寫入本地磁盤。對生產(chǎn)環(huán)境來說強(qiáng)烈建議開啟持久化并且生產(chǎn)者發(fā)送時(shí)設(shè)置DeliveryMode.PERSISTENT?？梢赃@樣理解二者的關(guān)系持久化確保 Broker 重啟后消息還在PERSISTENT 發(fā)送模式確保每條消息都走持久化存儲。即使 Broker 配置了 KahaDB如果生產(chǎn)者發(fā)送時(shí)用的是非持久化模式消息還是可能因?yàn)?Broker 崩潰而丟失。ActiveMQ 的主配置文件中persistenceAdapter的部分默認(rèn)就是 KahaDBpersistenceAdapter kahaDB directory${activemq.data}/kahadb/ /persistenceAdapter學(xué)習(xí)階段可以刻意做一次“重啟測試”發(fā)送一條持久化消息但消費(fèi)者不啟動然后重啟 Broker再啟動消費(fèi)者觀察消息是否還在。我第一次測的時(shí)候發(fā)現(xiàn)消息還在其實(shí)很不解——后來意識到這就是持久化在起作用。5.4 組合起來可靠消息的生產(chǎn)端配置怎么開把三層機(jī)制組合起來一個(gè)穩(wěn)妥、可靠的生產(chǎn)者配置看起來是這樣Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); producer.setTimeToLive(60000); // 消息有效期為 1 分鐘 producer.send(message);消費(fèi)者側(cè)如果對處理結(jié)果有嚴(yán)格成功要求通常選擇事務(wù)會話Session session connection.createSession(true, Session.SESSION_TRANSACTED); MessageConsumer consumer session.createConsumer(queue); TextMessage msg (TextMessage) consumer.receive(); // 處理業(yè)務(wù)邏輯比如更新數(shù)據(jù)庫、調(diào)用外部接口 // 都成功后提交 session.commit(); // 失敗則回滾 session.rollback();這里涉及一個(gè)進(jìn)階考量如果消費(fèi)者處理到一半進(jìn)程崩了事務(wù)尚未 commitBroker 會重新投遞這條消息。重新投遞可能導(dǎo)致重復(fù)消費(fèi)所以消費(fèi)者的業(yè)務(wù)邏輯必須做冪等。消息中間件本身并不能徹底避免重復(fù)投遞它只能保證“至少一次”不能保證“正好一次”。這句話我建議所有初學(xué)者刻在腦子里。6. 學(xué)習(xí)過程中實(shí)際踩過的四個(gè)坑這部分是我最想寫的。Demo 都能跑通但生產(chǎn)要素一進(jìn)來問題就花樣百出。下面四個(gè)坑都是我真實(shí)遇到過的按照排查鏈路寫出來方便你以后對著復(fù)現(xiàn)。6.1 第一個(gè)坑啟動 Broker 就報(bào)端口占用現(xiàn)象執(zhí)行./activemq start后控制臺報(bào)java.net.BindException: Address already in use: JVM_Bind。排查我把堆棧里暴露的端口號 61616 用lsof -i:61616查了下發(fā)現(xiàn)是本機(jī)之前跑過一個(gè)沒關(guān)干凈的 ActiveMQ 實(shí)例。這提醒我兩個(gè)習(xí)慣一是問題定位要先確認(rèn)端口占用而不是盲目重裝二是啟動前用./activemq stop或jps查看已有 Java 進(jìn)程別把多個(gè) broker 實(shí)例混在一起跑。如果你是在同一臺機(jī)器上跑多個(gè) broker 做集群學(xué)習(xí)必須修改conf/activemq.xml里的 transportConnector 端口還要調(diào)整conf/jetty.xml里的管理端口否則必然沖突。6.2 第二個(gè)坑隊(duì)列里明明有消息消費(fèi)者卻收不到現(xiàn)象生產(chǎn)者發(fā)送成功控制臺里Number Of Pending Messages是 5但消費(fèi)者啟動后一條都沒收到。排查過程我先確認(rèn)消費(fèi)者確實(shí)連接上了 broker——控制臺的 Connections 頁面能看到消費(fèi)者連接。然后檢查消費(fèi)的隊(duì)列名發(fā)現(xiàn)生產(chǎn)者和消費(fèi)者用的隊(duì)列名不一致。這個(gè)屬于低級錯(cuò)誤。但還有一個(gè)更隱蔽的原因困擾了我更久消費(fèi)者連接沒調(diào)用connection.start()。我照著網(wǎng)上某篇文章只創(chuàng)建了 connection 和 session沒有顯式調(diào)用 start。這個(gè)操作在“異步監(jiān)聽 等待”的例子中看起來沒問題但實(shí)際消息分發(fā)一直沒有真正開始。補(bǔ)上connection.start()后一切正常。這個(gè)教訓(xùn)讓我從此養(yǎng)成了一個(gè)習(xí)慣JMS 樣板代碼一定逐行核驗(yàn)別漏掉生命周期方法。6.3 第三個(gè)坑發(fā)布訂閱模式下“先啟動生產(chǎn)者后啟動消費(fèi)者”丟消息現(xiàn)象我用 Topic 建模一個(gè)廣播消息先運(yùn)行生產(chǎn)者發(fā)了兩條消息再啟動消費(fèi)者結(jié)果消息一條都沒收到。排查過程這其實(shí)是 Topic 模型的正常行為不是 bug。非持久訂閱者不會收到訂閱之前的消息。我當(dāng)時(shí)誤以為 Topic 和 Queue 一樣消息會先積壓在 broker 里消費(fèi)者一來就全推給它。搞清楚這個(gè)機(jī)制后我重新設(shè)計(jì)了一版客戶端先啟動消費(fèi)者并保持在線再觸發(fā)生產(chǎn)者發(fā)送消息就正常收到了。如果業(yè)務(wù)上確實(shí)需要“離線訂閱不丟消息”就要在消費(fèi)端使用createDurableSubscriber。注意持久訂閱的訂閱 ID 一旦創(chuàng)建就不能隨意變更否則舊的持久訂閱消息將無法再被消費(fèi)會一直堆積在 broker 中。6.4 第四個(gè)坑重復(fù)消費(fèi)引發(fā)的臟數(shù)據(jù)問題現(xiàn)象消費(fèi)者消費(fèi)消息后更新了數(shù)據(jù)庫但進(jìn)程在數(shù)據(jù)庫事務(wù)提交后、Session 提交前崩潰了重啟后同一消息被再次消費(fèi)數(shù)據(jù)庫被更新了兩次。排查過程當(dāng)時(shí)我用了事務(wù)會話以為有事務(wù)就萬事大吉。實(shí)際上事務(wù)會話保證的是“消息確認(rèn)和業(yè)務(wù)處理的原子性”但“業(yè)務(wù)處理”本身如果不在同一個(gè)數(shù)據(jù)庫事務(wù)里狀態(tài)就無法對賬。后來我用了一個(gè)冪等方案消息體帶業(yè)務(wù)主鍵消費(fèi)者在處理前先查一下數(shù)據(jù)庫是否已有對應(yīng)記錄有就直接返回沒有才執(zhí)行業(yè)務(wù)并落庫且主鍵唯一索引兜底。這樣即使消息重復(fù)投遞也不會產(chǎn)生重復(fù)數(shù)據(jù)。這個(gè)坑給我最大的啟發(fā)是消息中間件只保證傳輸層面的至少一次投遞消費(fèi)端必須自己做好冪等。你以為中間件幫你解決了所有問題它只是幫你把問題限定在了一個(gè)可控范圍內(nèi)剩下的還要靠自己的業(yè)務(wù)設(shè)計(jì)。7. 學(xué)習(xí)收尾時(shí)的一些心得過完前面這六個(gè)章節(jié)后我又重新回頭看了最初促使我開始學(xué)習(xí)的那個(gè)“同步調(diào)用雪崩”場景。用上 ActiveMQ 之后其實(shí)還沒有結(jié)束還配套做了消費(fèi)并發(fā)度控制、數(shù)據(jù)庫連接數(shù)控制、失敗重試隊(duì)列、監(jiān)控告警才敢說這個(gè)鏈路基本可靠了。這個(gè)認(rèn)知我覺得比學(xué)會 API 本身更重要消息中間件解決的是通信模式問題但一個(gè)系統(tǒng)的穩(wěn)定性永遠(yuǎn)要靠整體設(shè)計(jì)不會因?yàn)橐胍粋€(gè)組件就自動變好。如果你正在學(xué) JMS 和 ActiveMQ我給三條很實(shí)在的建議。第一條不要只看著我的代碼抄一定自己把生產(chǎn)者、消費(fèi)者、管理控制臺三樣對起來跑通一個(gè)完整的收發(fā)。第二條把 Queue 和 Topic 的區(qū)別多做幾個(gè)實(shí)驗(yàn)包括離線訂閱、多個(gè)消費(fèi)者消費(fèi)同一個(gè)隊(duì)列、持久訂閱恢復(fù)這些體驗(yàn)比背概念管用太多。第三條第一次接觸消息可靠性的時(shí)候強(qiáng)制自己回答“消息什么時(shí)候會丟、重復(fù)消費(fèi)怎么兜底、中間件重啟后消息還在不在”這三個(gè)問題回答完再繼續(xù)深入事務(wù)和持久化否則知識很容易停留在 API 層面。最后分享一個(gè)小技巧學(xué)習(xí)期間可以在 ActiveMQ 的配置里臨時(shí)把日志級別調(diào)成 DEBUG重點(diǎn)看 messages 相關(guān)日志和 session 相關(guān)日志。以前我一直覺得中間件日志太多不看后來排查那兩次詭異問題時(shí)DEBUG 日志里的 broker 內(nèi)部流轉(zhuǎn)信息幫了大忙。日志是排查問題最早出手的武器不要等到控制臺里消息都詭異了才想起來翻日志。