我第一次正兒八經(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、持久化到底意味著什么的人。我會(huì)把規(guī)范、代碼、踩坑都放在一起講盡量還原當(dāng)時(shí)的學(xué)習(xí)路徑也把網(wǎng)絡(luò)上那些講得含糊的地方補(bǔ)完整。1. 為什么我先啃 JMS 而不是直接學(xué)某個(gè) MQ 客戶端1.1 消息中間件解決的核心問題把“同步等待”變成“異步解耦”先說我當(dāng)時(shí)的場(chǎng)景。有個(gè)訂單創(chuàng)建服務(wù)處理完訂單之后要通知庫存服務(wù)扣庫存、通知積分服務(wù)加積分、通知短信服務(wù)發(fā)通知。最開始寫的代碼很“樸素”——A 服務(wù)直接用 HTTP 調(diào)用 B 服務(wù)同步等返回?？雌饋磉壿嫼苤卑椎粔簻y(cè)就出問題任何一個(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é)消息的存儲(chǔ)和轉(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)先理解這對(duì)關(guān)系JMSJava Message Service是 Java 平臺(tái)制定的一套消息服務(wù)接口規(guī)范它規(guī)定了 Java 程序怎么創(chuàng)建連接、怎么發(fā)消息、怎么收消息但 JMS 自己不干活真正干活的是符合這套規(guī)范的 provider比如 Apache ActiveMQ、開源的 OpenMQ 等。這個(gè)關(guān)系有點(diǎn)像 JDBC 和 MySQL 驅(qū)動(dòng)的關(guān)系——JDBC 規(guī)定了你用什么方法連接數(shù)據(jù)庫MySQL 驅(qū)動(dòng)負(fù)責(zé)把這些方法翻譯成 MySQL 能懂的協(xié)議。這就解釋了為什么學(xué)習(xí)的時(shí)候要先看 JMS 再看 ActiveMQ如果你只對(duì)著 ActiveMQ 的 API 抄一遍換個(gè) MQ 你就廢了如果你把 JMS 的核心模型理解了接觸任何一個(gè)兼容 JMS 的中間件都能很快上手。我當(dāng)初踩過一個(gè)認(rèn)知誤區(qū)以為 ActiveMQ 提供了比 JMS 多得多的高級(jí)功能所以直接按 ActiveMQ 的文檔寫就行。實(shí)際上 ActiveMQ 對(duì) JMS 規(guī)范做了很完整的實(shí)現(xiàn)它的核心 API 就是 JMS API你寫代碼時(shí) import 的主要是javax.jms.*這個(gè)包而不是org.apache.activemq.*。后者只在構(gòu)造連接工廠時(shí)才會(huì)用到。1.3 什么場(chǎng)景才需要引入消息中間件先潑盆冷水消息中間件不是銀彈學(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)建之后庫存、積分、日志都要處理。反過來說如果只是簡(jiǎn)單的 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è)對(duì)象通常是重量級(jí)對(duì)象創(chuàng)建成本高所以應(yīng)該復(fù)用不要每次發(fā)消息都 new 一個(gè)。我看到不少剛?cè)腴T的同學(xué)把ActiveMQConnectionFactory寫在方法內(nèi)部短時(shí)間大量請(qǐng)求一來連接資源就爆炸了。正確做法是使用連接池或至少是單例持有。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定了本次會(huì)話的確認(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ù)會(huì)被忽略因?yàn)槭聞?wù)模式下發(fā)送和接收都通過commit和rollback控制。剛開始不用懵理解成“Session 是一段通信上下文凡是這次會(huì)話內(nèi)的消息都可以在這個(gè)上下文里標(biāo)記狀態(tài)”就行。2.3 Destination消息要發(fā)往的地址Destination 是抽象類實(shí)際子類要么是Queue要么是Topic。Queue 對(duì)應(yīng)點(diǎn)對(duì)點(diǎn)模型Topic 對(duì)應(yīng)發(fā)布訂閱模型。在代碼里session.createQueue()或session.createTopic()創(chuàng)建的只是邏輯地址不需要在 Broker 側(cè)預(yù)先建好ActiveMQ 會(huì)在第一次使用時(shí)自動(dòng)創(chuàng)建。Queue queue session.createQueue(DEMO.ORDER); Topic topic session.createTopic(DEMO.NEWS);這里有個(gè)細(xì)節(jié)隊(duì)列名稱建議使用“域名前綴 業(yè)務(wù)名”的格式比如ORDER.CREATED、STOCK.DEDUCTED為了在管理后臺(tái)一眼能看出用途。我見過用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)先級(jí)、存活時(shí)間等特性這部分會(huì)影響消息的存儲(chǔ)行為后面展開。MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); TextMessage message session.createTextMessage(hello jms); producer.send(message);消費(fèi)者既可以主動(dòng)拉取消息也可以注冊(cè)消息監(jiān)聽器異步接收消息。主動(dòng)拉取用consumer.receive()它會(huì)阻塞等待可以傳入超時(shí)時(shí)間異步監(jiān)聽用MessageListener接口Broker 一有消息就回調(diào)onMessage方法。2.5 Message 結(jié)構(gòu)消息頭、屬性、消息體三層分工JMS 的消息對(duì)象由三部分組成。第一個(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適合鍵值對(duì)結(jié)構(gòu)ObjectMessage適合序列化對(duì)象但跨語言不友好StreamMessage用得最少。實(shí)際項(xiàng)目里 80% 的流量都是TextMessage因?yàn)闃I(yè)務(wù)方一般約定用 JSON 傳結(jié)構(gòu)化數(shù)據(jù)不會(huì)強(qiáng)行依賴 Java 對(duì)象序列化。我自己的習(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í)體變更會(huì)導(dǎo)致下游反序列化失敗。3. ActiveMQ 落地實(shí)操從下載啟動(dòng)到第一個(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 上對(duì) JMS 規(guī)范的支持更現(xiàn)代。作為學(xué)習(xí) JMS 的入口我建議用 Classic因?yàn)橘Y料多、踩坑記錄全先把規(guī)范玩明白。3.1 啟動(dòng) Broker十分鐘確認(rèn)環(huán)境下載解壓后進(jìn)入 bin 目錄Linux/macOS 下執(zhí)行./activemq startWindows 下執(zhí)行activemq.bat start。啟動(dòng)完成后默認(rèn)監(jiān)聽以下幾個(gè)重要端口端口用途61616JMS 客戶端連接的 OpenWire 端口8161Web 管理控制臺(tái)端口5672AMQP 協(xié)議端口默認(rèn)也開啟打開瀏覽器訪問http://127.0.0.1:8161/默認(rèn)登錄賬號(hào)是 admin密碼也是 admin進(jìn)去就能看到隊(duì)列列表、Topic 列表、連接數(shù)、消息統(tǒng)計(jì)等。我強(qiáng)烈建議學(xué)習(xí)階段養(yǎng)成盯控制臺(tái)的習(xí)慣很多代碼“看起來沒反應(yīng)”的問題在控制臺(tái)里一眼就能看出來——消息到底有沒有進(jìn)來、被誰消費(fèi)了、有沒有進(jìn)入死信。3.2 第一個(gè) JMS 生產(chǎn)者發(fā)送一條持久化文本消息下面這端代碼是我當(dāng)時(shí)跑通的第一個(gè)生產(chǎn)者。為了方便閱讀我把異常處理簡(jiǎn)化了學(xué)習(xí)階段先跑通生產(chǎn)上請(qǐng)務(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)建連接并啟動(dòng) Connection connection factory.createConnection(); connection.start(); // 3. 創(chuàng)建會(huì)話非事務(wù) 自動(dòng)確認(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)建訂單號(hào)20250101); producer.send(textMessage); System.out.println(已發(fā)送消息 textMessage.getText()); // 7. 關(guān)閉資源 producer.close(); session.close(); connection.close(); } }這個(gè)例子里最容易忽略的是connection.start()。有的教程會(huì)在創(chuàng)建生產(chǎn)者之前就調(diào)用 start有的會(huì)在發(fā)送之后才調(diào)用。start()本身不是發(fā)送動(dòng)作的開關(guān)但它會(huì)讓連接進(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)者有兩種寫法我建議都試一遍。第一種是主動(dòng) receive適合測(cè)試和批處理任務(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 秒沒消息會(huì)超時(shí)返回 null Message message consumer.receive(3000); if (message instanceof TextMessage) { System.out.println(收到消息 ((TextMessage) message).getText()); } consumer.close(); session.close(); connection.close(); } }第二種是注冊(cè)監(jiān)聽器異步接收實(shí)際項(xiàng)目里這種用得更多因?yàn)樗粫?huì)阻塞業(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)聽器注冊(cè)完還沒來得及收消息進(jìn)程就沒了。看起來像是“監(jiān)聽器沒生效”其實(shí)只是主線程提前退出。生產(chǎn)環(huán)境一般由應(yīng)用服務(wù)器或 Spring 容器持有消費(fèi)者生命周期不存在這個(gè)問題。3.4 跑完之后去控制臺(tái)觀察四個(gè)維度第一個(gè)消息鏈路跑通后我不急著寫第二個(gè) Demo而是去管理控制臺(tái)把所有變化對(duì)照了一遍。四個(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)者處理不過來。先建立這套觀察手段后面排坑會(huì)省很多時(shí)間?？湛谡f“消息丟了”“隊(duì)列卡了”很容易被運(yùn)維懟把控制臺(tái)截圖甩過去才是正確的溝通方式。4. 隊(duì)列模式和發(fā)布訂閱模式看起來代碼差不多語義差遠(yuǎn)了JMS 里最繞、也最影響選型的就是兩種消息模型點(diǎn)對(duì)點(diǎn)模式Queue和發(fā)布訂閱模式Topic。很多初學(xué)者能分別跑通但沒搞明白為什么需要兩種。4.1 Queue一條消息只被一個(gè)消費(fèi)者取走點(diǎn)對(duì)點(diǎn)模型下一條消息只會(huì)被一個(gè)消費(fèi)者成功接收。你可以啟動(dòng)多個(gè)消費(fèi)者實(shí)例同時(shí)消費(fèi)同一個(gè)隊(duì)列ActiveMQ 會(huì)自動(dòng)做負(fù)載均衡把不同消息分發(fā)給不同的消費(fèi)者。這像是一堆快遞包裹放在驛站多個(gè)配送員各自拿走一部分每個(gè)包裹最終只會(huì)被一個(gè)配送員取走。這個(gè)模型適合“任務(wù)分發(fā)”場(chǎng)景比如一批待處理文件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條件互斥消息可能會(huì)按條件被分流但默認(rèn)無 selector 時(shí)就是平均分發(fā)不是廣播。4.2 Topic一條消息廣播給所有訂閱者發(fā)布訂閱模型下一條消息會(huì)被所有“當(dāng)前活躍訂閱者”都收到。這像是廣播喇叭誰在聽誰就能聽見。Topic 模式下沒有“競(jìng)爭(zhēng)消費(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)者掛多久都不怕消息會(huì)積壓著但 Topic 不會(huì)。要解決離線訂閱丟失的問題JMS 提供了持久訂閱Durable Subscription。持久訂閱狀態(tài)下Broker 會(huì)為這個(gè)訂閱者保存未消費(fèi)的消息等它上線后再推送。4.3 代碼對(duì)比一條線上隊(duì)列一條線上廣播Queue 和 Topic 的代碼結(jié)構(gòu)幾乎一致只有兩行不同——目的地對(duì)象和創(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)者寫法會(huì)多一點(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 連接兩次第二次會(huì)替代第一次的訂閱關(guān)系。發(fā)送端完全一樣MessageProducer producer session.createProducer(topic); producer.send(session.createTextMessage(財(cái)經(jīng)早報(bào)));4.4 選型判斷別讓“感覺好用”替你做決定學(xué)習(xí)期間我總結(jié)了一個(gè)很簡(jiǎn)單的判斷規(guī)則同一業(yè)務(wù)事件如果多個(gè)系統(tǒng)需要各自獨(dú)立處理用 Topic如果一個(gè)任務(wù)的多個(gè)實(shí)例競(jìng)爭(zhēng)處理同一批消息用 Queue。舉兩個(gè)對(duì)比場(chǎng)景訂單創(chuàng)建后觸發(fā)庫存扣減、積分增加、短信通知這是“一魚多吃”三個(gè)系統(tǒng)各自消費(fèi)同一事件應(yīng)該用 Topic 或給每個(gè)消費(fèi)者建獨(dú)立 Queue。秒殺活動(dòng)里有 50 萬條搶購請(qǐng)求后面有 10 臺(tái)冪等處理服務(wù)實(shí)例消化這是“任務(wù)競(jìng)爭(zhēng)”應(yīng)該用 Queue。更常見的企業(yè)做法是“混合使用”用 Topic 接收領(lǐng)域事件然后每個(gè)訂閱者內(nèi)部再用 Queue 做異步削峰。初學(xué)階段先別追求花哨能把 Queue 和 Topic 的邊界吃透就超過了大多數(shù)人。5. 可靠消息的底層機(jī)制確認(rèn)、事務(wù)、持久化一次講清消息中間件最常被問的一句話是“消息到底會(huì)不會(huì)丟”答案取決于三層配置消費(fèi)者的確認(rèn)方式、會(huì)話的事務(wù)邊界、Broker 的持久化策略。每一層都直接影響消息的可靠性但要組合起來才有意義。5.1 確認(rèn)模式消費(fèi)者收完之后Broker 才敢刪消息主動(dòng)和異步監(jiān)聽之外Session 的確認(rèn)模式?jīng)Q定了消息何時(shí)被“算作處理成功”。在 JMS 1.1 里有四種常見模式模式行為適用場(chǎng)景AUTO_ACKNOWLEDGE消費(fèi)者收到消息后自動(dòng)確認(rèn)receive 返回或 onMessage 返回時(shí)業(yè)務(wù)處理邏輯簡(jiǎn)單、處理失敗可以接受消息丟失的場(chǎng)景CLIENT_ACKNOWLEDGE需要手動(dòng)調(diào)用 message.acknowledge() 確認(rèn)處理流程較復(fù)雜希望確認(rèn)時(shí)機(jī)由業(yè)務(wù)代碼控制的場(chǎng)景DUPS_OK_ACKNOWLEDGE允許重復(fù)消息延遲確認(rèn)性能好可容忍重復(fù)消費(fèi)的統(tǒng)計(jì)類場(chǎng)景SESSION_TRANSACTED配合事務(wù)會(huì)話使用需要保證消息收發(fā)與業(yè)務(wù)操作原子一致的場(chǎng)景最常見的坑是使用 AUTO_ACKNOWLEDGE但是消費(fèi)者在onMessage里處理到一半拋異常。此時(shí)消息實(shí)際上已經(jīng)被確認(rèn)了Broker 會(huì)把它刪除重啟消費(fèi)者后這條消息不會(huì)再出現(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 會(huì)把這批全確認(rèn)了。要精確控制單條消息必須結(jié)合事務(wù)會(huì)話每個(gè)事務(wù)處理一條就 commit 一次。5.2 事務(wù)會(huì)話讓發(fā)送和接收具備原子性connection.createSession(true, Session.SESSION_TRANSACTED)開啟了事務(wù)模式。事務(wù)模式下生產(chǎn)者的發(fā)送動(dòng)作不會(huì)立即生效而是要等session.commit()才真正把消息提交給 Broker消費(fèi)者的接收動(dòng)作也不會(huì)立即被確認(rèn)而是等 commit 時(shí)才確認(rèn)這一批消息。如果業(yè)務(wù)處理失敗調(diào)用session.rollback()消息會(huì)被回滾之后可以再次消費(fèi)。事務(wù)會(huì)話對(duì)“發(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)爭(zhēng)和日志壓力越大。經(jīng)驗(yàn)值是一個(gè)事務(wù)里處理的業(yè)務(wù)消息量控制在幾十條以內(nèi)尤其在高并發(fā)場(chǎng)景下過大的事務(wù)反而會(huì)成為瓶頸。5.3 持久化消息存儲(chǔ)到磁盤進(jìn)程重啟才能恢復(fù)ActiveMQ 默認(rèn)將消息保存在內(nèi)存里同時(shí)提供 KahaDB 作為默認(rèn)持久化方案它會(huì)將消息寫入本地磁盤。對(duì)生產(chǎn)環(huán)境來說強(qiáng)烈建議開啟持久化并且生產(chǎn)者發(fā)送時(shí)設(shè)置DeliveryMode.PERSISTENT?？梢赃@樣理解二者的關(guān)系持久化確保 Broker 重啟后消息還在PERSISTENT 發(fā)送模式確保每條消息都走持久化存儲(chǔ)。即使 Broker 配置了 KahaDB如果生產(chǎn)者發(fā)送時(shí)用的是非持久化模式消息還是可能因?yàn)?Broker 崩潰而丟失。ActiveMQ 的主配置文件中persistenceAdapter的部分默認(rèn)就是 KahaDBpersistenceAdapter kahaDB directory${activemq.data}/kahadb/ /persistenceAdapter學(xué)習(xí)階段可以刻意做一次“重啟測(cè)試”發(fā)送一條持久化消息但消費(fèi)者不啟動(dòng)然后重啟 Broker再啟動(dòng)消費(fèi)者觀察消息是否還在。我第一次測(cè)的時(shí)候發(fā)現(xiàn)消息還在其實(shí)很不解——后來意識(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è)如果對(duì)處理結(jié)果有嚴(yán)格成功要求通常選擇事務(wù)會(huì)話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 會(huì)重新投遞這條消息。重新投遞可能導(dǎo)致重復(fù)消費(fèi)所以消費(fèi)者的業(yè)務(wù)邏輯必須做冪等。消息中間件本身并不能徹底避免重復(fù)投遞它只能保證“至少一次”不能保證“正好一次”。這句話我建議所有初學(xué)者刻在腦子里。6. 學(xué)習(xí)過程中實(shí)際踩過的四個(gè)坑這部分是我最想寫的。Demo 都能跑通但生產(chǎn)要素一進(jìn)來問題就花樣百出。下面四個(gè)坑都是我真實(shí)遇到過的按照排查鏈路寫出來方便你以后對(duì)著復(fù)現(xiàn)。6.1 第一個(gè)坑啟動(dòng) Broker 就報(bào)端口占用現(xiàn)象執(zhí)行./activemq start后控制臺(tái)報(bào)java.net.BindException: Address already in use: JVM_Bind。排查我把堆棧里暴露的端口號(hào) 61616 用lsof -i:61616查了下發(fā)現(xiàn)是本機(jī)之前跑過一個(gè)沒關(guān)干凈的 ActiveMQ 實(shí)例。這提醒我兩個(gè)習(xí)慣一是問題定位要先確認(rèn)端口占用而不是盲目重裝二是啟動(dòng)前用./activemq stop或jps查看已有 Java 進(jìn)程別把多個(gè) broker 實(shí)例混在一起跑。如果你是在同一臺(tái)機(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ā)送成功控制臺(tái)里Number Of Pending Messages是 5但消費(fèi)者啟動(dòng)后一條都沒收到。排查過程我先確認(rèn)消費(fèi)者確實(shí)連接上了 broker——控制臺(tái)的 Connections 頁面能看到消費(fèi)者連接。然后檢查消費(fèi)的隊(duì)列名發(fā)現(xiàn)生產(chǎn)者和消費(fèi)者用的隊(duì)列名不一致。這個(gè)屬于低級(jí)錯(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ā)布訂閱模式下“先啟動(dòng)生產(chǎn)者后啟動(dòng)消費(fèi)者”丟消息現(xiàn)象我用 Topic 建模一個(gè)廣播消息先運(yùn)行生產(chǎn)者發(fā)了兩條消息再啟動(dòng)消費(fèi)者結(jié)果消息一條都沒收到。排查過程這其實(shí)是 Topic 模型的正常行為不是 bug。非持久訂閱者不會(huì)收到訂閱之前的消息。我當(dāng)時(shí)誤以為 Topic 和 Queue 一樣消息會(huì)先積壓在 broker 里消費(fèi)者一來就全推給它。搞清楚這個(gè)機(jī)制后我重新設(shè)計(jì)了一版客戶端先啟動(dòng)消費(fèi)者并保持在線再觸發(fā)生產(chǎn)者發(fā)送消息就正常收到了。如果業(yè)務(wù)上確實(shí)需要“離線訂閱不丟消息”就要在消費(fèi)端使用createDurableSubscriber。注意持久訂閱的訂閱 ID 一旦創(chuàng)建就不能隨意變更否則舊的持久訂閱消息將無法再被消費(fèi)會(huì)一直堆積在 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ù)會(huì)話以為有事務(wù)就萬事大吉。實(shí)際上事務(wù)會(huì)話保證的是“消息確認(rèn)和業(yè)務(wù)處理的原子性”但“業(yè)務(wù)處理”本身如果不在同一個(gè)數(shù)據(jù)庫事務(wù)里狀態(tài)就無法對(duì)賬。后來我用了一個(gè)冪等方案消息體帶業(yè)務(wù)主鍵消費(fèi)者在處理前先查一下數(shù)據(jù)庫是否已有對(duì)應(yīng)記錄有就直接返回沒有才執(zhí)行業(yè)務(wù)并落庫且主鍵唯一索引兜底。這樣即使消息重復(fù)投遞也不會(huì)產(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)用雪崩”場(chǎng)景。用上 ActiveMQ 之后其實(shí)還沒有結(jié)束還配套做了消費(fèi)并發(fā)度控制、數(shù)據(jù)庫連接數(shù)控制、失敗重試隊(duì)列、監(jiān)控告警才敢說這個(gè)鏈路基本可靠了。這個(gè)認(rèn)知我覺得比學(xué)會(huì) API 本身更重要消息中間件解決的是通信模式問題但一個(gè)系統(tǒng)的穩(wěn)定性永遠(yuǎn)要靠整體設(shè)計(jì)不會(huì)因?yàn)橐胍粋€(gè)組件就自動(dòng)變好。如果你正在學(xué) JMS 和 ActiveMQ我給三條很實(shí)在的建議。第一條不要只看著我的代碼抄一定自己把生產(chǎn)者、消費(fèi)者、管理控制臺(tái)三樣對(duì)起來跑通一個(gè)完整的收發(fā)。第二條把 Queue 和 Topic 的區(qū)別多做幾個(gè)實(shí)驗(yàn)包括離線訂閱、多個(gè)消費(fèi)者消費(fèi)同一個(gè)隊(duì)列、持久訂閱恢復(fù)這些體驗(yàn)比背概念管用太多。第三條第一次接觸消息可靠性的時(shí)候強(qiáng)制自己回答“消息什么時(shí)候會(huì)丟、重復(fù)消費(fèi)怎么兜底、中間件重啟后消息還在不在”這三個(gè)問題回答完再繼續(xù)深入事務(wù)和持久化否則知識(shí)很容易停留在 API 層面。最后分享一個(gè)小技巧學(xué)習(xí)期間可以在 ActiveMQ 的配置里臨時(shí)把日志級(jí)別調(diào)成 DEBUG重點(diǎn)看 messages 相關(guān)日志和 session 相關(guān)日志。以前我一直覺得中間件日志太多不看后來排查那兩次詭異問題時(shí)DEBUG 日志里的 broker 內(nèi)部流轉(zhuǎn)信息幫了大忙。日志是排查問題最早出手的武器不要等到控制臺(tái)里消息都詭異了才想起來翻日志。