戰(zhàn)：消息隊(duì)列核心原理、代碼示例與可靠性機(jī)制解析)
從我自己的經(jīng)歷說起吧。在接手一個(gè)內(nèi)部系統(tǒng)重構(gòu)時(shí)需要把耗時(shí)較長的報(bào)表生成任務(wù)從請求線程里摘出去后端不用一直傻等前端也不用一直轉(zhuǎn)圈。當(dāng)時(shí)組里討論了好幾個(gè)方案最后敲定用 JMS ActiveMQ 做異步消息隊(duì)列。我啃了不少文檔、踩了不少坑也把很多容易繞暈的概念理順了。這篇就把我學(xué)習(xí)和落地過程中的理解、代碼和踩坑記錄梳理成一則學(xué)習(xí)筆記希望能幫到正在學(xué) JMS 和 ActiveMQ 的朋友。剛開始看 JMS 的時(shí)候最大的感覺是術(shù)語多、抽象什么 Session、Connection、Destination不看實(shí)際代碼很難串起來。后來我用快遞系統(tǒng)做類比很多概念一下就通了。消息中間件本質(zhì)就是一個(gè)中間轉(zhuǎn)運(yùn)站生產(chǎn)者是寄件人消費(fèi)者是收件人Broker 就是分揀中心。這個(gè)類比后面還會貫穿全文。1. JMS 是什么它到底在解決什么問題先講清楚一個(gè)東西JMS 全稱 Java Message Service它是一套規(guī)范不是具體的實(shí)現(xiàn)。這套規(guī)范把消息生產(chǎn)、消息投遞、消息消費(fèi)這些操作全部抽象成統(tǒng)一的 Java API這樣你用 ActiveMQ、RabbitMQ 或者其他支持 JMS 的中間件核心代碼結(jié)構(gòu)是差不多的。換個(gè)中間件改改配置業(yè)務(wù)代碼不需要大改。那它到底解決什么問題我覺得可以分成三類場景來理解。第一類是異步解耦。比如用戶下單后系統(tǒng)要發(fā)短信、發(fā)郵件、更新積分這些操作如果全串在請求線程里接口就會很慢。把超時(shí)的任務(wù)丟進(jìn)消息隊(duì)列訂單接口馬上返回后面的消費(fèi)端慢慢處理用戶感知到的速度就快很多。第二類是流量削峰。秒殺、搶券場景瞬時(shí)流量特別大數(shù)據(jù)庫很容易被打垮。讓所有請求先走進(jìn)消息隊(duì)列后端消費(fèi)端按自己能力慢慢處理系統(tǒng)就不會被沖垮。第三類是應(yīng)用間通信。不同的服務(wù)、不同的語言寫出來的模塊常常需要互相傳數(shù)據(jù)。通過 JMS 中間件A 服務(wù)把消息丟到隊(duì)列里B 服務(wù)拉走去處理兩者無需直接建立網(wǎng)絡(luò)連接降低了耦合。我當(dāng)時(shí)做的報(bào)表生成任務(wù)就是典型的異步化和流量削峰一次生成可能要跑幾分鐘而且經(jīng)常大量并發(fā)觸發(fā)直接同步做會拖垮應(yīng)用服務(wù)器所以丟進(jìn)排隊(duì)再逐個(gè)消費(fèi)是最合適的落地方案。注意JMS 是面向 Java 的一套規(guī)范。如果你的系統(tǒng)完全是異構(gòu)的比如 PHP、Go、Python 混用可能更適合直接用 AMQP比如 RabbitMQ或者別的跨語言協(xié)議。JMS 的強(qiáng)項(xiàng)在于 Java 生態(tài)內(nèi)語言統(tǒng)一時(shí)用起來非常順暢。2. JMS 的核心術(shù)語和生活化理解我第一次看 JMS 文檔ConnectionFactory、Connection、Session、MessageProducer、MessageConsumer 這串名詞排在一起直接看懵了。用快遞類比以后這幾個(gè)概念就清晰多了。ConnectionFactory好比快遞公司的營業(yè)點(diǎn)你要寄件先找個(gè)營業(yè)點(diǎn)拿到聯(lián)系渠道。在 JMS 里它負(fù)責(zé)創(chuàng)建 Connection一般啟動(dòng)時(shí)初始化一個(gè)全局實(shí)例就夠了。Connection好比你和快遞公司之間建立的一條專屬運(yùn)輸通道。它代表客戶端與 Broker 之間的物理連接通常一個(gè)應(yīng)用只創(chuàng)建一兩個(gè) Connection。Session好比一次寄件業(yè)務(wù)的會話窗口。它是創(chuàng)建消息、生產(chǎn)者和消費(fèi)者的工廠也是事務(wù)的邊界。一個(gè) Connection 可以創(chuàng)建多個(gè) Session。Destination就是消息要去的地址。Queue隊(duì)列是點(diǎn)對點(diǎn)模式的地址Topic主題是發(fā)布訂閱模式的地址。類比快遞就是寄到哪里比如“某市某街道某小區(qū)”。MessageProducer消息生產(chǎn)者把消息塞到 Destination 去。類比寄件人把包裹放進(jìn)快遞柜。MessageConsumer消息消費(fèi)者從 Destination 里取消息。類比收件人從驛站取包裹。Message消息本身JMS 規(guī)定了若干種消息類型TextMessage、BytesMessage、MapMessage、ObjectMessage、StreamMessage。日常用的最多的是 TextMessage因?yàn)樗缯Z言、跨版本的兼容性最好。還有兩個(gè)很容易忽略但很重要的概念消息確認(rèn)Acknowledgement和事務(wù)Transaction。消息確認(rèn)解決的是“消費(fèi)者收到消息后Broker 怎么知道可以扔掉了”這個(gè)問題。Broker 把消息投遞給消費(fèi)者后不會立刻刪除需要消費(fèi)者確認(rèn)“我處理好了”才會真正移除。如果消費(fèi)者收到消息后沒來得及確認(rèn)就掛了消息還會被重新投遞。事務(wù)解決的是“一組消息要么全部成功要么全部失敗”的問題跟數(shù)據(jù)庫事務(wù)的思想一模一樣。在 Session 里開啟事務(wù)發(fā)送多條消息后統(tǒng)一 commit任何一條失敗都可以 rollback那這一批消息就全部撤銷。理解了這些概念再回來看 API 代碼就很順了。所有 JMS 客戶端代碼套路都是一樣的創(chuàng)建 ConnectionFactory - 創(chuàng)建 Connection - 創(chuàng)建 Session - 創(chuàng)建 Destination - 創(chuàng)建 Producer/Consumer - 發(fā)/收消息。3. ActiveMQ 為什么適合做 JMS 的入門與實(shí)戰(zhàn)落地點(diǎn)再說說為什么選 ActiveMQ而不是 Kafka 或者 RabbitMQ。Kafka 的強(qiáng)項(xiàng)是海量日志吞吐、流式處理但它原本是分布式日志系統(tǒng)思路消息確認(rèn)、順序性、延遲這些跟業(yè)務(wù)消息中間件的訴求不太一樣。RabbitMQ 是另一套 AMQP 協(xié)議生態(tài)不能直接原生使用 JMS API。ActiveMQ 是 JMS 規(guī)范的原生實(shí)現(xiàn)開箱即用學(xué)習(xí)曲線平滑對中小規(guī)模內(nèi)部系統(tǒng)完全夠用。ActiveMQ 的兩個(gè)主要版本方向也提一下。ActiveMQ Classic5.x是傳統(tǒng)版本功能完整成熟穩(wěn)定我項(xiàng)目里用的就是這套。ActiveMQ Artemis2.x 以后是升級版引擎性能更強(qiáng)、高可用架構(gòu)也更現(xiàn)代。如果你是全新項(xiàng)目可以直接從 Artemis 入手如果你維護(hù)老系統(tǒng)大概率還是碰到 Classic。我的筆記以 Classic 為主但核心 JMS 代碼兩者通用。我的學(xué)習(xí)路徑建議是先在本地把 ActiveMQ 跑起來用眼睛看一下管理控制臺里隊(duì)列和消息長什么樣再去寫第一行代碼。盲寫代碼但腦子里沒有畫面很容易卡殼。3.1 本地安裝與快速啟動(dòng)安裝 ActiveMQ 5.x其實(shí)很簡單# 下載解壓后進(jìn)入目錄 cd apache-activemq-5.18.x # 啟動(dòng)默認(rèn)模式 bin/activemq start # 查看控制臺日志 tail -f data/activemq.log # 管理控制臺地址 # 默認(rèn) 8161 端口打開 http://localhost:8161管理控制臺默認(rèn)賬號密碼是 admin/admin登錄后可以看到隊(duì)列、主題、消費(fèi)者數(shù)量、未消費(fèi)消息數(shù)等信息。我第一次練手時(shí)故意不寫消費(fèi)者往隊(duì)列里發(fā)了幾條消息然后跑到控制臺去看隊(duì)列里確實(shí)能看到消息堆積那個(gè)畫面讓我對“Broker 中轉(zhuǎn)”立刻有了直觀認(rèn)知。如果啟動(dòng)報(bào)端口占用通常是因?yàn)?61616OpenWire 端口或 8161Web 控制臺被占用修改 conf/jetty.xml 里的端口配置或者直接換個(gè)空閑端口。注意ActiveMQ 5.x 默認(rèn)監(jiān)聽 0.0.0.0生產(chǎn)環(huán)境必須做訪問控制和防火墻限制不能把管理控制臺和 OpenWire 端口裸奔到公網(wǎng)這是個(gè)非?，F(xiàn)實(shí)的安全點(diǎn)。3.2 生產(chǎn)落地的兩個(gè)關(guān)鍵配置文件部署到服務(wù)器時(shí)最常用的兩個(gè)配置是 conf/activemq.xml 和 conf/jetty.xml。activemq.xml 負(fù)責(zé)核心配置包括持久化適配器、傳輸連接器、認(rèn)證插件等。jetty.xml 則是 Web 管理控制臺的配置。最常見的調(diào)整就是把默認(rèn)的 KahaDB 存儲路徑改到有足夠磁盤空間的目錄防止磁盤寫滿導(dǎo)致 Broker 崩潰persistenceAdapter kahaDB directory/data/activemq/kahadb / /persistenceAdapter如果你要開啟連接密碼認(rèn)證則在 broker 標(biāo)簽下加一個(gè)簡單認(rèn)證插件配置用戶名和密碼plugins simpleAuthenticationPlugin users authenticationUser usernameapp-user passwordapp-pass groupsusers/ /users /simpleAuthenticationPlugin /plugins這些配置不復(fù)雜但每一條都可能直接影響線上穩(wěn)定性。改完配置文件都需要重啟 Broker 才生效測試環(huán)境改完務(wù)必驗(yàn)證一遍。4. 動(dòng)手實(shí)戰(zhàn)Producer 和 Consumer 的關(guān)鍵代碼與運(yùn)行細(xì)節(jié)本地環(huán)境準(zhǔn)備好了就可以開始寫代碼了。我先用 Maven 建一個(gè)普通 Java 工程引入 ActiveMQ 客戶端依賴。版本號要和 Broker 版本匹配我用的版本組合是 ActiveMQ 5.18.x 對應(yīng)客戶端依賴 5.18.x不要混用大版本否則可能出現(xiàn)協(xié)議兼容問題。dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId version5.18.2/version /dependency4.1 生產(chǎn)端代碼發(fā)送一條消息到隊(duì)列先寫一個(gè)最簡生產(chǎn)者public class SimpleProducer { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME demo.queue; public static void main(String[] args) throws Exception { // 1. 創(chuàng)建連接工廠 ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); // 2. 創(chuàng)建連接并啟動(dòng) try (ActiveMQConnection connection (ActiveMQConnection) factory.createConnection()) { connection.start(); // 3. 第一個(gè)參數(shù) true 表示開啟事務(wù) Session session connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 4. 創(chuàng)建隊(duì)列目標(biāo) Destination destination session.createQueue(QUEUE_NAME); // 5. 創(chuàng)建生產(chǎn)者 MessageProducer producer session.createProducer(destination); // 6. 創(chuàng)建文本消息 TextMessage textMessage session.createTextMessage(hello jms activemq); // 7. 發(fā)送 producer.send(textMessage); // 8. 提交事務(wù)消息才會真正進(jìn)入隊(duì)列 session.commit(); System.out.println(消息發(fā)送成功: textMessage.getText()); } } }這里有幾個(gè)細(xì)節(jié)要注意都是新手容易踩的坑。第一個(gè)是事務(wù)模式下一定要記得 commit。很多人代碼寫完了消息似乎也 send 成功了去控制臺看卻什么都沒有原因往往是忘了 commit。沒有提交的事務(wù)消息不會真正被 Broker 接收。第二個(gè)是連接工廠的 URL。默認(rèn)tcp://localhost:61616是 ActiveMQ Classic 的 OpenWire 協(xié)議端口不要用 61616 去連 8161這是很常見的混淆。8161 是 Web 控制臺61616 才是消息端口。第三個(gè)是 try-with-resources 的使用。Connection 和 Session 都實(shí)現(xiàn)了 Closeable用 try-with-resources 能保證程序結(jié)束前正常釋放資源避免連接一直占著不釋放。提示生產(chǎn)者的 Connection 啟動(dòng)不是必須的但建議養(yǎng)成習(xí)慣統(tǒng)一connection.start()避免后續(xù)某些消費(fèi)者場景下忘記啟動(dòng)而收不到消息。4.2 消費(fèi)端代碼同步拉取與異步監(jiān)聽消費(fèi)者有兩個(gè)模型一個(gè)是同步阻塞式consumer.receive()一個(gè)是異步回調(diào)式setMessageListener。同步接收代碼結(jié)構(gòu)如下public class SimpleConsumer { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME demo.queue; public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); try (ActiveMQConnection connection (ActiveMQConnection) factory.createConnection()) { connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination session.createQueue(QUEUE_NAME); MessageConsumer consumer session.createConsumer(destination); // 同步阻塞1秒超時(shí) Message message consumer.receive(1000); if (message instanceof TextMessage) { TextMessage textMessage (TextMessage) message; System.out.println(收到消息: textMessage.getText()); } } } }這里我還是建議用receive(long timeout)而不是裸receive()因?yàn)槁?receive 會一直阻塞如果隊(duì)列里沒有消息線程就永遠(yuǎn)掛在那兒了。加個(gè)超時(shí)設(shè)置邏輯更可控代碼也更適合放在后臺任務(wù)里。異步監(jiān)聽是生產(chǎn)上更常用的方式。它的好處是消費(fèi)者線程不用自己寫 while 循環(huán)Broker 有消息推過來回調(diào)方法自動(dòng)執(zhí)行public class AsyncConsumer { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost:61616); try (ActiveMQConnection connection (ActiveMQConnection) factory.createConnection()) { connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination session.createQueue(demo.queue); MessageConsumer consumer session.createConsumer(destination); consumer.setMessageListener(message - { try { if (message instanceof TextMessage) { System.out.println(異步收到消息: ((TextMessage) message).getText()); } } catch (Exception e) { // 這里要特別小心異常必須被捕獲否則監(jiān)聽線程容易出問題 } }); // keep alive Thread.sleep(10000); } } }異步監(jiān)聽里最容易犯的一個(gè)錯(cuò)誤就是回調(diào)方法內(nèi)不處理異常。因?yàn)?JMS 的消息監(jiān)聽回調(diào)是在 JMS 服務(wù)線程里跑的如果回調(diào)拋出 RuntimeException線程可能被中斷消息也可能被重新投遞造成“看起來消費(fèi)者崩了”的問題。所以回調(diào)方法第一件事就是做好異常兜底。注意異步監(jiān)聽模式下的消費(fèi)者比較特殊代碼里不要用Thread.sleep來模擬長任務(wù)。如果回調(diào)業(yè)務(wù)處理時(shí)間長可以另外開工作線程池處理把消息轉(zhuǎn)發(fā)給線程池之后立即返回。否則 JMS 回調(diào)線程被占滿后續(xù)消息推送會被堵住。4.3 Queue 與 Topic 在代碼上的差異和實(shí)際表現(xiàn)很多人以為隊(duì)列和主題就是換一個(gè) API 的問題實(shí)際上行為差異很大。Queue 是點(diǎn)對點(diǎn)模型一條消息只會被一個(gè)消費(fèi)者消費(fèi)。你啟動(dòng)多個(gè)消費(fèi)者監(jiān)聽同一個(gè)隊(duì)列消息會分?jǐn)偟讲煌M(fèi)者手里。這是天然的負(fù)載均衡場景適合任務(wù)分發(fā)、訂單處理這類業(yè)務(wù)。Topic 是發(fā)布訂閱模型一條消息會被所有在線的訂閱者各收到一份。每個(gè)訂閱者相當(dāng)于一個(gè)獨(dú)立的消費(fèi)者組。適合廣播通知、實(shí)時(shí)行情推送這類場景。在代碼上兩者的差異其實(shí)就是一個(gè) API// 隊(duì)列 Destination destination session.createQueue(demo.queue); // 主題 Destination destination session.createTopic(demo.topic);但 Topic 有個(gè)很關(guān)鍵的注意事項(xiàng)普通消費(fèi)者如果先啟動(dòng)并訂閱了主題生產(chǎn)者再發(fā)消息能收到。如果消費(fèi)者晚于生產(chǎn)者啟動(dòng)或者中間斷開重連那就收不到那段時(shí)間內(nèi)的消息了。因?yàn)槠胀ㄓ嗛喪恰霸诰€訂閱”離線消息不保留。如果你想讓后啟動(dòng)的消費(fèi)者也能收到之前漏掉的消息JMS 提供了持久訂閱Durable Subscription機(jī)制。它要求消費(fèi)者必須設(shè)置一個(gè)持久化客戶端 IDBroker 會為它保留離線期間的消息等它重新上線后再推送。connection.setClientID(client-1); // 創(chuàng)建持久訂閱 TopicSubscriber subscriber session.createDurableSubscriber(topic, subscriber-1);這里有個(gè)細(xì)節(jié)持久訂閱的名字和連接客戶端的 ID 一旦固定就不要輕易改因?yàn)?Broker 靠這兩個(gè)標(biāo)識區(qū)分不同的訂閱者。換了名稱舊的消息就找不回來了就相當(dāng)于換了一個(gè)入坑身份。我自己的項(xiàng)目里Topic 用得不多大部分實(shí)際需求其實(shí)用 Queue 就夠。做多模塊廣播時(shí)會更傾向于讓每個(gè)接收方自行創(chuàng)建隊(duì)列并訂閱進(jìn)一步降低耦合。5. 消息可靠性背后的機(jī)制與實(shí)操抉擇對于業(yè)務(wù)系統(tǒng)來說消息不能丟、不能重復(fù)是剛需。JMS 這套體系設(shè)計(jì)了三個(gè)層面來保證可靠性消息確認(rèn)ACK、事務(wù)、持久化。我把學(xué)習(xí)時(shí)的理解都說清楚。5.1 ACK 確認(rèn)機(jī)制Broker 與消費(fèi)者之間的握手簡單說一下 ACK 的四種常見模式AUTO_ACKNOWLEDGE自動(dòng)確認(rèn)。消費(fèi)者順利收到消息沒有拋異常就自動(dòng)回執(zhí)告訴 Broker 可以刪了。這是在代碼里最常見的模式。CLIENT_ACKNOWLEDGE客戶端手動(dòng)確認(rèn)。消費(fèi)者需要顯式調(diào)用message.acknowledge()Broker 收到確認(rèn)后才會刪除消息。適合處理流程長、需要業(yè)務(wù)代碼控制確認(rèn)時(shí)機(jī)的場景。DUPS_OK_ACKNOWLEDGE允許重復(fù)確認(rèn)性能最好但可能丟消息。嚴(yán)格的業(yè)務(wù)系統(tǒng)一般不用。SESSION_TRANSACTED配合事務(wù)使用消息是否確認(rèn)取決于事務(wù)是否提交。我第一次看到 CLIENT_ACKNOWLEDGE 時(shí)有個(gè)疑問手動(dòng)確認(rèn)的意義到底在哪里后來在對接外部短信通道時(shí)明白了。消費(fèi)端拉取消息到本地調(diào)用外部接口發(fā)送短信外部接口超時(shí)或返回錯(cuò)誤但如果代碼里用的是 AUTO_ACKNOWLEDGE消息在消費(fèi)者“拿到”那一刻就已經(jīng)被確認(rèn)了超時(shí)錯(cuò)誤并不會觸發(fā)重發(fā)這條短信就悄悄丟了。改成 CLIENT_ACKNOWLEDGE 后我們可以先拿到消息執(zhí)行完整業(yè)務(wù)邏輯確認(rèn)外部通道成功后再手動(dòng) acknowledge。如果失敗就不確認(rèn)Broker 會重新投遞這條消息短信就不會丟。這是“至少一次投遞”模式下的常見做法。Session session connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); // 收到消息并成功處理業(yè)務(wù)后 message.acknowledge();注意acknowledge()是作用在當(dāng)前 Session 內(nèi)所有未確認(rèn)消息上的它會一次性確認(rèn)這個(gè) Session 里收到的所有消息而不是只確認(rèn)調(diào)用它的那條。如果你在同一 Session 里同時(shí)處理多條消息不想一次全確認(rèn)那就得考慮按消息拆分 Session或者用事務(wù)模式。5.2 事務(wù)模式消息的批量提交與回滾事務(wù)在 JMS 中的作用是保證一批消息的原子性。用createSession(true, ...)開啟事務(wù)模式下第一個(gè)參數(shù)是是否使用事務(wù)第二個(gè) ACK 參數(shù)會被忽略。事務(wù)模式最常見的場景是消費(fèi)端從隊(duì)列批量拉取一批消息比如 100 條全部處理成功后commit()只要有一條失敗就rollback()這 100 條消息全部會被重新投遞。它的好處是保證批量一致性壞處也很明顯回滾會造成大量消息重復(fù)投遞浪費(fèi)。所以事務(wù)粒度要把握好一次事務(wù)最好只處理一個(gè)小批次不要把幾萬條消息放進(jìn)一個(gè)事務(wù)里。生產(chǎn)端使用事務(wù)還有一層意義批量發(fā)送消息時(shí)不開啟事務(wù)每條消息都要和 Broker 做一次網(wǎng)絡(luò)請求性能較差。開啟事務(wù)后多條 send 可以攢在一起commit 時(shí)一次性發(fā)送吞吐量會有明顯提升。尤其是大量批量消息入庫的時(shí)候這個(gè)優(yōu)化非常有效。事務(wù)模式適合批量審批通知、大批量數(shù)據(jù)同步 非事務(wù)自動(dòng)確認(rèn)模式適合實(shí)時(shí)性要求高、單條消息處理快的場景5.3 持久化Broker 掛了消息還能找回ActiveMQ 的持久化解決的是“Broker 宕機(jī)后消息是否還能恢復(fù)”的問題。發(fā)送消息時(shí)可以在 producer 上設(shè)置發(fā)送模式來決定這條消息是否持久化producer.setDeliveryMode(DeliveryMode.PERSISTENT);如果設(shè)置成 PERSISTENT消息會被寫入磁盤存儲。Broker 意外重啟后未消費(fèi)的消息依然在。如果設(shè)置成 NON_PERSISTENT消息只存在內(nèi)存里Broker 重啟后消息就沒了但性能更好。ActiveMQ 5.x 默認(rèn)使用 KahaDB 作為持久化存儲它是一個(gè)基于日志文件的本地?cái)?shù)據(jù)庫專門針對消息存儲優(yōu)化過。有個(gè)生產(chǎn)經(jīng)驗(yàn)值得記住持久化和發(fā)送確認(rèn)是兩回事。即使消息已經(jīng)持久化發(fā)送端的send()方法也不過是告訴你在語義上消息已被收下而不是說“已經(jīng)入盤了”。如果你要的是權(quán)衡可靠性可以在doPublish這類方法里再結(jié)合事務(wù)使用。建議生產(chǎn)環(huán)境一律開啟 PERSISTENT不要為了省那點(diǎn)性能去關(guān)持久化。等真的遇到 Broker 重啟導(dǎo)致消息丟失時(shí)再提性能就晚了。6. 常見問題與排查技巧實(shí)錄學(xué)習(xí)和上線過程中的踩坑記錄我總結(jié)成一張速查表再挑幾個(gè)典型的詳細(xì)說。問題現(xiàn)象可能原因排查思路生產(chǎn)者連接超時(shí)端口寫錯(cuò)、Broker 沒啟動(dòng)、防火墻攔截telnet 測試 61616 端口查看 activemq.log消息發(fā)送了但消費(fèi)者收不到生產(chǎn)端忘了 commit、訂閱 Topic 錯(cuò)過消息、消費(fèi)者沒 start看控制臺隊(duì)列堆積檢查代碼 Session 是否開啟事務(wù)消費(fèi)者一直收消息但業(yè)務(wù)重復(fù)執(zhí)行消費(fèi)回調(diào)異常、ACK 超時(shí)導(dǎo)致重投檢查回調(diào)異常捕獲、檢查 ACK 模式Broker 突然內(nèi)存暴漲有大量非持久化消息堆積、消費(fèi)者卡住控制臺看待消費(fèi)消息數(shù)量檢查消費(fèi)者是否死掉監(jiān)聽回調(diào)不觸發(fā)了回調(diào)線程占滿、連接失敗未重連查看日志、檢查連接工廠 failover 配置下面展開幾個(gè)比較值得講的。6.1 消費(fèi)者回調(diào)里出現(xiàn)異常導(dǎo)致消息無限重投這是我調(diào)試時(shí)碰到的一個(gè)典型問題。消費(fèi)端代碼在回調(diào)里調(diào)用了遠(yuǎn)程接口遠(yuǎn)程接口拋了 RuntimeException如果沒有 try-catch 包住JMS 客戶端收到回調(diào)異常后消息會進(jìn)入重投邏輯導(dǎo)致同一條消息不斷被消費(fèi)日志刷屏業(yè)務(wù)數(shù)據(jù)也可能被重復(fù)寫入。解決思路分兩層第一層是回調(diào)方法內(nèi)必須做異常兜底把單個(gè)消息的失敗隔離出來。第二層是把消息的失敗語義明確了。如果這是一條業(yè)務(wù)上永遠(yuǎn)無法處理成功的“死信消息”一直重投沒有任何意義應(yīng)當(dāng)捕獲后記錄到錯(cuò)誤表或死信隊(duì)列讓后續(xù)人工介入。6.2 消費(fèi)者連接斷了以后沒有自動(dòng)重連默認(rèn)的tcp://localhost:61616連接方式網(wǎng)絡(luò)故障或 Broker 重啟后不會自動(dòng)恢復(fù)連接。生產(chǎn)環(huán)境要使用 failover 協(xié)議讓客戶端能在連接斷開后自動(dòng)重連到 BrokerString brokerURL failover:(tcp://localhost:61616)?initialReconnectDelay1000maxReconnectAttempts20; ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(brokerURL);failover 協(xié)議是 ActiveMQ 留給生產(chǎn)環(huán)境的“保命”配置它能在底層連接斷開時(shí)按照參數(shù)配置重連。配合異步監(jiān)聽模式重連后消費(fèi)者會自動(dòng)恢復(fù)訂閱業(yè)務(wù)代碼幾乎無感。6.3 手動(dòng)確認(rèn)模式下忘記 acknowledge 導(dǎo)致消息永遠(yuǎn)收不到新消息用 CLIENT_ACKNOWLEDGE 模式時(shí)如果消費(fèi)者拉取消息后代碼拋異常沒有調(diào)用 acknowledgeBroker 就不會移除消息。但這里有個(gè)隱蔽問題如果你每次連的是同一條消息一直拉取失敗那后續(xù)新消息就進(jìn)不來因?yàn)榍懊娴南⒍伦×岁?duì)列頭。換一批新消息之前要確保每一條都成功確認(rèn)或者用 try-finally 結(jié)構(gòu)確保在處理異常時(shí)也做出明確決策是確認(rèn)丟棄還是回滾重投。不能模棱兩可。7. 學(xué)習(xí)路徑建議與后續(xù)擴(kuò)展方向如果你剛開始學(xué) JMS 和 ActiveMQ我的建議路徑是這樣先把官方 Quick Start 跑一遍然后自己寫一個(gè)生產(chǎn)者、兩個(gè)消費(fèi)者同時(shí)監(jiān)聽隊(duì)列觀察消息如何分?jǐn)?。再寫一個(gè) Topic 的發(fā)布訂閱 demo對比兩種模型的行為差異。接著把 ACK 模式、事務(wù)模式、持久化三個(gè)可靠性維度分別驗(yàn)證一遍基本概念就牢固了。后續(xù)擴(kuò)展方面有幾個(gè)方向值得繼續(xù)深入第一個(gè)是 Spring 集成。Spring 提供了 JmsTemplate 和JmsListener注解封裝了大部分模板代碼。用 Spring Boot 做集成時(shí)需要配置 ConnectionFactory、JmsListenerContainerFactory然后一個(gè)注解就能監(jiān)聽隊(duì)列消息代碼非常簡潔。但從學(xué)習(xí)角度我強(qiáng)烈建議先寫好原生 JMS API 再進(jìn) Spring因?yàn)槟憷斫饬说讓佑每蚣軙r(shí)才能明白它替你做了哪些事遇到問題才知道去哪里排查。第二個(gè)是失效轉(zhuǎn)移和集群。ActiveMQ 可以配置共享存儲下的主從模式或者網(wǎng)絡(luò)連接器Network Bridge做多個(gè) Broker 互聯(lián)形成集群。集群引入后要考慮的不僅是消息怎么發(fā)送還有消費(fèi)者負(fù)載的均勻分配、訂閱恢復(fù)的機(jī)制。第三個(gè)是吞吐量壓測。在把系統(tǒng)放上線之前最好對隊(duì)列做一輪簡單的壓測把生產(chǎn)端的發(fā)送速度、消費(fèi)端的處理速度、消息大小都記錄下來然后再決定線程池大小、事務(wù)批次大小這些參數(shù)。我做完這個(gè)項(xiàng)目后最大的感觸是JMS 和 ActiveMQ 本身并不難難的是想清楚每一層機(jī)制在解決什么問題。ACK 是為了“消費(fèi)結(jié)果可感知”事務(wù)是為了“多消息操作可原子”持久化是為了“Broker 掛了數(shù)據(jù)不丟”。把這些機(jī)制放到真實(shí)業(yè)務(wù)里去理解代碼怎么寫都順手。 JMS與ActiveMQ實(shí)戰(zhàn)：消息隊(duì)列核心原理、代碼示例與可靠性機(jī)制解析
從我自己的經(jīng)歷說起吧。在接手一個(gè)內(nèi)部系統(tǒng)重構(gòu)時(shí)需要把耗時(shí)較長的報(bào)表生成任務(wù)從請求線程里摘出去后端不用一直傻等前端也不用一直轉(zhuǎn)圈。當(dāng)時(shí)組里討論了好幾個(gè)方案最后敲定用 JMS ActiveMQ 做異步消息隊(duì)列。我啃了不少文檔、踩了不少坑也把很多容易繞暈的概念理順了。這篇就把我學(xué)習(xí)和落地過程中的理解、代碼和踩坑記錄梳理成一則學(xué)習(xí)筆記希望能幫到正在學(xué) JMS 和 ActiveMQ 的朋友。剛開始看 JMS 的時(shí)候最大的感覺是術(shù)語多、抽象什么 Session、Connection、Destination不看實(shí)際代碼很難串起來。后來我用快遞系統(tǒng)做類比很多概念一下就通了。消息中間件本質(zhì)就是一個(gè)中間轉(zhuǎn)運(yùn)站生產(chǎn)者是寄件人消費(fèi)者是收件人Broker 就是分揀中心。這個(gè)類比后面還會貫穿全文。1. JMS 是什么它到底在解決什么問題先講清楚一個(gè)東西JMS 全稱 Java Message Service它是一套規(guī)范不是具體的實(shí)現(xiàn)。這套規(guī)范把消息生產(chǎn)、消息投遞、消息消費(fèi)這些操作全部抽象成統(tǒng)一的 Java API這樣你用 ActiveMQ、RabbitMQ 或者其他支持 JMS 的中間件核心代碼結(jié)構(gòu)是差不多的。換個(gè)中間件改改配置業(yè)務(wù)代碼不需要大改。那它到底解決什么問題我覺得可以分成三類場景來理解。第一類是異步解耦。比如用戶下單后系統(tǒng)要發(fā)短信、發(fā)郵件、更新積分這些操作如果全串在請求線程里接口就會很慢。把超時(shí)的任務(wù)丟進(jìn)消息隊(duì)列訂單接口馬上返回后面的消費(fèi)端慢慢處理用戶感知到的速度就快很多。第二類是流量削峰。秒殺、搶券場景瞬時(shí)流量特別大數(shù)據(jù)庫很容易被打垮。讓所有請求先走進(jìn)消息隊(duì)列后端消費(fèi)端按自己能力慢慢處理系統(tǒng)就不會被沖垮。第三類是應(yīng)用間通信。不同的服務(wù)、不同的語言寫出來的模塊常常需要互相傳數(shù)據(jù)。通過 JMS 中間件A 服務(wù)把消息丟到隊(duì)列里B 服務(wù)拉走去處理兩者無需直接建立網(wǎng)絡(luò)連接降低了耦合。我當(dāng)時(shí)做的報(bào)表生成任務(wù)就是典型的異步化和流量削峰一次生成可能要跑幾分鐘而且經(jīng)常大量并發(fā)觸發(fā)直接同步做會拖垮應(yīng)用服務(wù)器所以丟進(jìn)排隊(duì)再逐個(gè)消費(fèi)是最合適的落地方案。注意JMS 是面向 Java 的一套規(guī)范。如果你的系統(tǒng)完全是異構(gòu)的比如 PHP、Go、Python 混用可能更適合直接用 AMQP比如 RabbitMQ或者別的跨語言協(xié)議。JMS 的強(qiáng)項(xiàng)在于 Java 生態(tài)內(nèi)語言統(tǒng)一時(shí)用起來非常順暢。2. JMS 的核心術(shù)語和生活化理解我第一次看 JMS 文檔ConnectionFactory、Connection、Session、MessageProducer、MessageConsumer 這串名詞排在一起直接看懵了。用快遞類比以后這幾個(gè)概念就清晰多了。ConnectionFactory好比快遞公司的營業(yè)點(diǎn)你要寄件先找個(gè)營業(yè)點(diǎn)拿到聯(lián)系渠道。在 JMS 里它負(fù)責(zé)創(chuàng)建 Connection一般啟動(dòng)時(shí)初始化一個(gè)全局實(shí)例就夠了。Connection好比你和快遞公司之間建立的一條專屬運(yùn)輸通道。它代表客戶端與 Broker 之間的物理連接通常一個(gè)應(yīng)用只創(chuàng)建一兩個(gè) Connection。Session好比一次寄件業(yè)務(wù)的會話窗口。它是創(chuàng)建消息、生產(chǎn)者和消費(fèi)者的工廠也是事務(wù)的邊界。一個(gè) Connection 可以創(chuàng)建多個(gè) Session。Destination就是消息要去的地址。Queue隊(duì)列是點(diǎn)對點(diǎn)模式的地址Topic主題是發(fā)布訂閱模式的地址。類比快遞就是寄到哪里比如“某市某街道某小區(qū)”。MessageProducer消息生產(chǎn)者把消息塞到 Destination 去。類比寄件人把包裹放進(jìn)快遞柜。MessageConsumer消息消費(fèi)者從 Destination 里取消息。類比收件人從驛站取包裹。Message消息本身JMS 規(guī)定了若干種消息類型TextMessage、BytesMessage、MapMessage、ObjectMessage、StreamMessage。日常用的最多的是 TextMessage因?yàn)樗缯Z言、跨版本的兼容性最好。還有兩個(gè)很容易忽略但很重要的概念消息確認(rèn)Acknowledgement和事務(wù)Transaction。消息確認(rèn)解決的是“消費(fèi)者收到消息后Broker 怎么知道可以扔掉了”這個(gè)問題。Broker 把消息投遞給消費(fèi)者后不會立刻刪除需要消費(fèi)者確認(rèn)“我處理好了”才會真正移除。如果消費(fèi)者收到消息后沒來得及確認(rèn)就掛了消息還會被重新投遞。事務(wù)解決的是“一組消息要么全部成功要么全部失敗”的問題跟數(shù)據(jù)庫事務(wù)的思想一模一樣。在 Session 里開啟事務(wù)發(fā)送多條消息后統(tǒng)一 commit任何一條失敗都可以 rollback那這一批消息就全部撤銷。理解了這些概念再回來看 API 代碼就很順了。所有 JMS 客戶端代碼套路都是一樣的創(chuàng)建 ConnectionFactory - 創(chuàng)建 Connection - 創(chuàng)建 Session - 創(chuàng)建 Destination - 創(chuàng)建 Producer/Consumer - 發(fā)/收消息。3. ActiveMQ 為什么適合做 JMS 的入門與實(shí)戰(zhàn)落地點(diǎn)再說說為什么選 ActiveMQ而不是 Kafka 或者 RabbitMQ。Kafka 的強(qiáng)項(xiàng)是海量日志吞吐、流式處理但它原本是分布式日志系統(tǒng)思路消息確認(rèn)、順序性、延遲這些跟業(yè)務(wù)消息中間件的訴求不太一樣。RabbitMQ 是另一套 AMQP 協(xié)議生態(tài)不能直接原生使用 JMS API。ActiveMQ 是 JMS 規(guī)范的原生實(shí)現(xiàn)開箱即用學(xué)習(xí)曲線平滑對中小規(guī)模內(nèi)部系統(tǒng)完全夠用。ActiveMQ 的兩個(gè)主要版本方向也提一下。ActiveMQ Classic5.x是傳統(tǒng)版本功能完整成熟穩(wěn)定我項(xiàng)目里用的就是這套。ActiveMQ Artemis2.x 以后是升級版引擎性能更強(qiáng)、高可用架構(gòu)也更現(xiàn)代。如果你是全新項(xiàng)目可以直接從 Artemis 入手如果你維護(hù)老系統(tǒng)大概率還是碰到 Classic。我的筆記以 Classic 為主但核心 JMS 代碼兩者通用。我的學(xué)習(xí)路徑建議是先在本地把 ActiveMQ 跑起來用眼睛看一下管理控制臺里隊(duì)列和消息長什么樣再去寫第一行代碼。盲寫代碼但腦子里沒有畫面很容易卡殼。3.1 本地安裝與快速啟動(dòng)安裝 ActiveMQ 5.x其實(shí)很簡單# 下載解壓后進(jìn)入目錄 cd apache-activemq-5.18.x # 啟動(dòng)默認(rèn)模式 bin/activemq start # 查看控制臺日志 tail -f data/activemq.log # 管理控制臺地址 # 默認(rèn) 8161 端口打開 http://localhost:8161管理控制臺默認(rèn)賬號密碼是 admin/admin登錄后可以看到隊(duì)列、主題、消費(fèi)者數(shù)量、未消費(fèi)消息數(shù)等信息。我第一次練手時(shí)故意不寫消費(fèi)者往隊(duì)列里發(fā)了幾條消息然后跑到控制臺去看隊(duì)列里確實(shí)能看到消息堆積那個(gè)畫面讓我對“Broker 中轉(zhuǎn)”立刻有了直觀認(rèn)知。如果啟動(dòng)報(bào)端口占用通常是因?yàn)?61616OpenWire 端口或 8161Web 控制臺被占用修改 conf/jetty.xml 里的端口配置或者直接換個(gè)空閑端口。注意ActiveMQ 5.x 默認(rèn)監(jiān)聽 0.0.0.0生產(chǎn)環(huán)境必須做訪問控制和防火墻限制不能把管理控制臺和 OpenWire 端口裸奔到公網(wǎng)這是個(gè)非?，F(xiàn)實(shí)的安全點(diǎn)。3.2 生產(chǎn)落地的兩個(gè)關(guān)鍵配置文件部署到服務(wù)器時(shí)最常用的兩個(gè)配置是 conf/activemq.xml 和 conf/jetty.xml。activemq.xml 負(fù)責(zé)核心配置包括持久化適配器、傳輸連接器、認(rèn)證插件等。jetty.xml 則是 Web 管理控制臺的配置。最常見的調(diào)整就是把默認(rèn)的 KahaDB 存儲路徑改到有足夠磁盤空間的目錄防止磁盤寫滿導(dǎo)致 Broker 崩潰persistenceAdapter kahaDB directory/data/activemq/kahadb / /persistenceAdapter如果你要開啟連接密碼認(rèn)證則在 broker 標(biāo)簽下加一個(gè)簡單認(rèn)證插件配置用戶名和密碼plugins simpleAuthenticationPlugin users authenticationUser usernameapp-user passwordapp-pass groupsusers/ /users /simpleAuthenticationPlugin /plugins這些配置不復(fù)雜但每一條都可能直接影響線上穩(wěn)定性。改完配置文件都需要重啟 Broker 才生效測試環(huán)境改完務(wù)必驗(yàn)證一遍。4. 動(dòng)手實(shí)戰(zhàn)Producer 和 Consumer 的關(guān)鍵代碼與運(yùn)行細(xì)節(jié)本地環(huán)境準(zhǔn)備好了就可以開始寫代碼了。我先用 Maven 建一個(gè)普通 Java 工程引入 ActiveMQ 客戶端依賴。版本號要和 Broker 版本匹配我用的版本組合是 ActiveMQ 5.18.x 對應(yīng)客戶端依賴 5.18.x不要混用大版本否則可能出現(xiàn)協(xié)議兼容問題。dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId version5.18.2/version /dependency4.1 生產(chǎn)端代碼發(fā)送一條消息到隊(duì)列先寫一個(gè)最簡生產(chǎn)者public class SimpleProducer { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME demo.queue; public static void main(String[] args) throws Exception { // 1. 創(chuàng)建連接工廠 ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); // 2. 創(chuàng)建連接并啟動(dòng) try (ActiveMQConnection connection (ActiveMQConnection) factory.createConnection()) { connection.start(); // 3. 第一個(gè)參數(shù) true 表示開啟事務(wù) Session session connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 4. 創(chuàng)建隊(duì)列目標(biāo) Destination destination session.createQueue(QUEUE_NAME); // 5. 創(chuàng)建生產(chǎn)者 MessageProducer producer session.createProducer(destination); // 6. 創(chuàng)建文本消息 TextMessage textMessage session.createTextMessage(hello jms activemq); // 7. 發(fā)送 producer.send(textMessage); // 8. 提交事務(wù)消息才會真正進(jìn)入隊(duì)列 session.commit(); System.out.println(消息發(fā)送成功: textMessage.getText()); } } }這里有幾個(gè)細(xì)節(jié)要注意都是新手容易踩的坑。第一個(gè)是事務(wù)模式下一定要記得 commit。很多人代碼寫完了消息似乎也 send 成功了去控制臺看卻什么都沒有原因往往是忘了 commit。沒有提交的事務(wù)消息不會真正被 Broker 接收。第二個(gè)是連接工廠的 URL。默認(rèn)tcp://localhost:61616是 ActiveMQ Classic 的 OpenWire 協(xié)議端口不要用 61616 去連 8161這是很常見的混淆。8161 是 Web 控制臺61616 才是消息端口。第三個(gè)是 try-with-resources 的使用。Connection 和 Session 都實(shí)現(xiàn)了 Closeable用 try-with-resources 能保證程序結(jié)束前正常釋放資源避免連接一直占著不釋放。提示生產(chǎn)者的 Connection 啟動(dòng)不是必須的但建議養(yǎng)成習(xí)慣統(tǒng)一connection.start()避免后續(xù)某些消費(fèi)者場景下忘記啟動(dòng)而收不到消息。4.2 消費(fèi)端代碼同步拉取與異步監(jiān)聽消費(fèi)者有兩個(gè)模型一個(gè)是同步阻塞式consumer.receive()一個(gè)是異步回調(diào)式setMessageListener。同步接收代碼結(jié)構(gòu)如下public class SimpleConsumer { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME demo.queue; public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); try (ActiveMQConnection connection (ActiveMQConnection) factory.createConnection()) { connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination session.createQueue(QUEUE_NAME); MessageConsumer consumer session.createConsumer(destination); // 同步阻塞1秒超時(shí) Message message consumer.receive(1000); if (message instanceof TextMessage) { TextMessage textMessage (TextMessage) message; System.out.println(收到消息: textMessage.getText()); } } } }這里我還是建議用receive(long timeout)而不是裸receive()因?yàn)槁?receive 會一直阻塞如果隊(duì)列里沒有消息線程就永遠(yuǎn)掛在那兒了。加個(gè)超時(shí)設(shè)置邏輯更可控代碼也更適合放在后臺任務(wù)里。異步監(jiān)聽是生產(chǎn)上更常用的方式。它的好處是消費(fèi)者線程不用自己寫 while 循環(huán)Broker 有消息推過來回調(diào)方法自動(dòng)執(zhí)行public class AsyncConsumer { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost:61616); try (ActiveMQConnection connection (ActiveMQConnection) factory.createConnection()) { connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination session.createQueue(demo.queue); MessageConsumer consumer session.createConsumer(destination); consumer.setMessageListener(message - { try { if (message instanceof TextMessage) { System.out.println(異步收到消息: ((TextMessage) message).getText()); } } catch (Exception e) { // 這里要特別小心異常必須被捕獲否則監(jiān)聽線程容易出問題 } }); // keep alive Thread.sleep(10000); } } }異步監(jiān)聽里最容易犯的一個(gè)錯(cuò)誤就是回調(diào)方法內(nèi)不處理異常。因?yàn)?JMS 的消息監(jiān)聽回調(diào)是在 JMS 服務(wù)線程里跑的如果回調(diào)拋出 RuntimeException線程可能被中斷消息也可能被重新投遞造成“看起來消費(fèi)者崩了”的問題。所以回調(diào)方法第一件事就是做好異常兜底。注意異步監(jiān)聽模式下的消費(fèi)者比較特殊代碼里不要用Thread.sleep來模擬長任務(wù)。如果回調(diào)業(yè)務(wù)處理時(shí)間長可以另外開工作線程池處理把消息轉(zhuǎn)發(fā)給線程池之后立即返回。否則 JMS 回調(diào)線程被占滿后續(xù)消息推送會被堵住。4.3 Queue 與 Topic 在代碼上的差異和實(shí)際表現(xiàn)很多人以為隊(duì)列和主題就是換一個(gè) API 的問題實(shí)際上行為差異很大。Queue 是點(diǎn)對點(diǎn)模型一條消息只會被一個(gè)消費(fèi)者消費(fèi)。你啟動(dòng)多個(gè)消費(fèi)者監(jiān)聽同一個(gè)隊(duì)列消息會分?jǐn)偟讲煌M(fèi)者手里。這是天然的負(fù)載均衡場景適合任務(wù)分發(fā)、訂單處理這類業(yè)務(wù)。Topic 是發(fā)布訂閱模型一條消息會被所有在線的訂閱者各收到一份。每個(gè)訂閱者相當(dāng)于一個(gè)獨(dú)立的消費(fèi)者組。適合廣播通知、實(shí)時(shí)行情推送這類場景。在代碼上兩者的差異其實(shí)就是一個(gè) API// 隊(duì)列 Destination destination session.createQueue(demo.queue); // 主題 Destination destination session.createTopic(demo.topic);但 Topic 有個(gè)很關(guān)鍵的注意事項(xiàng)普通消費(fèi)者如果先啟動(dòng)并訂閱了主題生產(chǎn)者再發(fā)消息能收到。如果消費(fèi)者晚于生產(chǎn)者啟動(dòng)或者中間斷開重連那就收不到那段時(shí)間內(nèi)的消息了。因?yàn)槠胀ㄓ嗛喪恰霸诰€訂閱”離線消息不保留。如果你想讓后啟動(dòng)的消費(fèi)者也能收到之前漏掉的消息JMS 提供了持久訂閱Durable Subscription機(jī)制。它要求消費(fèi)者必須設(shè)置一個(gè)持久化客戶端 IDBroker 會為它保留離線期間的消息等它重新上線后再推送。connection.setClientID(client-1); // 創(chuàng)建持久訂閱 TopicSubscriber subscriber session.createDurableSubscriber(topic, subscriber-1);這里有個(gè)細(xì)節(jié)持久訂閱的名字和連接客戶端的 ID 一旦固定就不要輕易改因?yàn)?Broker 靠這兩個(gè)標(biāo)識區(qū)分不同的訂閱者。換了名稱舊的消息就找不回來了就相當(dāng)于換了一個(gè)入坑身份。我自己的項(xiàng)目里Topic 用得不多大部分實(shí)際需求其實(shí)用 Queue 就夠。做多模塊廣播時(shí)會更傾向于讓每個(gè)接收方自行創(chuàng)建隊(duì)列并訂閱進(jìn)一步降低耦合。5. 消息可靠性背后的機(jī)制與實(shí)操抉擇對于業(yè)務(wù)系統(tǒng)來說消息不能丟、不能重復(fù)是剛需。JMS 這套體系設(shè)計(jì)了三個(gè)層面來保證可靠性消息確認(rèn)ACK、事務(wù)、持久化。我把學(xué)習(xí)時(shí)的理解都說清楚。5.1 ACK 確認(rèn)機(jī)制Broker 與消費(fèi)者之間的握手簡單說一下 ACK 的四種常見模式AUTO_ACKNOWLEDGE自動(dòng)確認(rèn)。消費(fèi)者順利收到消息沒有拋異常就自動(dòng)回執(zhí)告訴 Broker 可以刪了。這是在代碼里最常見的模式。CLIENT_ACKNOWLEDGE客戶端手動(dòng)確認(rèn)。消費(fèi)者需要顯式調(diào)用message.acknowledge()Broker 收到確認(rèn)后才會刪除消息。適合處理流程長、需要業(yè)務(wù)代碼控制確認(rèn)時(shí)機(jī)的場景。DUPS_OK_ACKNOWLEDGE允許重復(fù)確認(rèn)性能最好但可能丟消息。嚴(yán)格的業(yè)務(wù)系統(tǒng)一般不用。SESSION_TRANSACTED配合事務(wù)使用消息是否確認(rèn)取決于事務(wù)是否提交。我第一次看到 CLIENT_ACKNOWLEDGE 時(shí)有個(gè)疑問手動(dòng)確認(rèn)的意義到底在哪里后來在對接外部短信通道時(shí)明白了。消費(fèi)端拉取消息到本地調(diào)用外部接口發(fā)送短信外部接口超時(shí)或返回錯(cuò)誤但如果代碼里用的是 AUTO_ACKNOWLEDGE消息在消費(fèi)者“拿到”那一刻就已經(jīng)被確認(rèn)了超時(shí)錯(cuò)誤并不會觸發(fā)重發(fā)這條短信就悄悄丟了。改成 CLIENT_ACKNOWLEDGE 后我們可以先拿到消息執(zhí)行完整業(yè)務(wù)邏輯確認(rèn)外部通道成功后再手動(dòng) acknowledge。如果失敗就不確認(rèn)Broker 會重新投遞這條消息短信就不會丟。這是“至少一次投遞”模式下的常見做法。Session session connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); // 收到消息并成功處理業(yè)務(wù)后 message.acknowledge();注意acknowledge()是作用在當(dāng)前 Session 內(nèi)所有未確認(rèn)消息上的它會一次性確認(rèn)這個(gè) Session 里收到的所有消息而不是只確認(rèn)調(diào)用它的那條。如果你在同一 Session 里同時(shí)處理多條消息不想一次全確認(rèn)那就得考慮按消息拆分 Session或者用事務(wù)模式。5.2 事務(wù)模式消息的批量提交與回滾事務(wù)在 JMS 中的作用是保證一批消息的原子性。用createSession(true, ...)開啟事務(wù)模式下第一個(gè)參數(shù)是是否使用事務(wù)第二個(gè) ACK 參數(shù)會被忽略。事務(wù)模式最常見的場景是消費(fèi)端從隊(duì)列批量拉取一批消息比如 100 條全部處理成功后commit()只要有一條失敗就rollback()這 100 條消息全部會被重新投遞。它的好處是保證批量一致性壞處也很明顯回滾會造成大量消息重復(fù)投遞浪費(fèi)。所以事務(wù)粒度要把握好一次事務(wù)最好只處理一個(gè)小批次不要把幾萬條消息放進(jìn)一個(gè)事務(wù)里。生產(chǎn)端使用事務(wù)還有一層意義批量發(fā)送消息時(shí)不開啟事務(wù)每條消息都要和 Broker 做一次網(wǎng)絡(luò)請求性能較差。開啟事務(wù)后多條 send 可以攢在一起commit 時(shí)一次性發(fā)送吞吐量會有明顯提升。尤其是大量批量消息入庫的時(shí)候這個(gè)優(yōu)化非常有效。事務(wù)模式適合批量審批通知、大批量數(shù)據(jù)同步 非事務(wù)自動(dòng)確認(rèn)模式適合實(shí)時(shí)性要求高、單條消息處理快的場景5.3 持久化Broker 掛了消息還能找回ActiveMQ 的持久化解決的是“Broker 宕機(jī)后消息是否還能恢復(fù)”的問題。發(fā)送消息時(shí)可以在 producer 上設(shè)置發(fā)送模式來決定這條消息是否持久化producer.setDeliveryMode(DeliveryMode.PERSISTENT);如果設(shè)置成 PERSISTENT消息會被寫入磁盤存儲。Broker 意外重啟后未消費(fèi)的消息依然在。如果設(shè)置成 NON_PERSISTENT消息只存在內(nèi)存里Broker 重啟后消息就沒了但性能更好。ActiveMQ 5.x 默認(rèn)使用 KahaDB 作為持久化存儲它是一個(gè)基于日志文件的本地?cái)?shù)據(jù)庫專門針對消息存儲優(yōu)化過。有個(gè)生產(chǎn)經(jīng)驗(yàn)值得記住持久化和發(fā)送確認(rèn)是兩回事。即使消息已經(jīng)持久化發(fā)送端的send()方法也不過是告訴你在語義上消息已被收下而不是說“已經(jīng)入盤了”。如果你要的是權(quán)衡可靠性可以在doPublish這類方法里再結(jié)合事務(wù)使用。建議生產(chǎn)環(huán)境一律開啟 PERSISTENT不要為了省那點(diǎn)性能去關(guān)持久化。等真的遇到 Broker 重啟導(dǎo)致消息丟失時(shí)再提性能就晚了。6. 常見問題與排查技巧實(shí)錄學(xué)習(xí)和上線過程中的踩坑記錄我總結(jié)成一張速查表再挑幾個(gè)典型的詳細(xì)說。問題現(xiàn)象可能原因排查思路生產(chǎn)者連接超時(shí)端口寫錯(cuò)、Broker 沒啟動(dòng)、防火墻攔截telnet 測試 61616 端口查看 activemq.log消息發(fā)送了但消費(fèi)者收不到生產(chǎn)端忘了 commit、訂閱 Topic 錯(cuò)過消息、消費(fèi)者沒 start看控制臺隊(duì)列堆積檢查代碼 Session 是否開啟事務(wù)消費(fèi)者一直收消息但業(yè)務(wù)重復(fù)執(zhí)行消費(fèi)回調(diào)異常、ACK 超時(shí)導(dǎo)致重投檢查回調(diào)異常捕獲、檢查 ACK 模式Broker 突然內(nèi)存暴漲有大量非持久化消息堆積、消費(fèi)者卡住控制臺看待消費(fèi)消息數(shù)量檢查消費(fèi)者是否死掉監(jiān)聽回調(diào)不觸發(fā)了回調(diào)線程占滿、連接失敗未重連查看日志、檢查連接工廠 failover 配置下面展開幾個(gè)比較值得講的。6.1 消費(fèi)者回調(diào)里出現(xiàn)異常導(dǎo)致消息無限重投這是我調(diào)試時(shí)碰到的一個(gè)典型問題。消費(fèi)端代碼在回調(diào)里調(diào)用了遠(yuǎn)程接口遠(yuǎn)程接口拋了 RuntimeException如果沒有 try-catch 包住JMS 客戶端收到回調(diào)異常后消息會進(jìn)入重投邏輯導(dǎo)致同一條消息不斷被消費(fèi)日志刷屏業(yè)務(wù)數(shù)據(jù)也可能被重復(fù)寫入。解決思路分兩層第一層是回調(diào)方法內(nèi)必須做異常兜底把單個(gè)消息的失敗隔離出來。第二層是把消息的失敗語義明確了。如果這是一條業(yè)務(wù)上永遠(yuǎn)無法處理成功的“死信消息”一直重投沒有任何意義應(yīng)當(dāng)捕獲后記錄到錯(cuò)誤表或死信隊(duì)列讓后續(xù)人工介入。6.2 消費(fèi)者連接斷了以后沒有自動(dòng)重連默認(rèn)的tcp://localhost:61616連接方式網(wǎng)絡(luò)故障或 Broker 重啟后不會自動(dòng)恢復(fù)連接。生產(chǎn)環(huán)境要使用 failover 協(xié)議讓客戶端能在連接斷開后自動(dòng)重連到 BrokerString brokerURL failover:(tcp://localhost:61616)?initialReconnectDelay1000maxReconnectAttempts20; ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(brokerURL);failover 協(xié)議是 ActiveMQ 留給生產(chǎn)環(huán)境的“保命”配置它能在底層連接斷開時(shí)按照參數(shù)配置重連。配合異步監(jiān)聽模式重連后消費(fèi)者會自動(dòng)恢復(fù)訂閱業(yè)務(wù)代碼幾乎無感。6.3 手動(dòng)確認(rèn)模式下忘記 acknowledge 導(dǎo)致消息永遠(yuǎn)收不到新消息用 CLIENT_ACKNOWLEDGE 模式時(shí)如果消費(fèi)者拉取消息后代碼拋異常沒有調(diào)用 acknowledgeBroker 就不會移除消息。但這里有個(gè)隱蔽問題如果你每次連的是同一條消息一直拉取失敗那后續(xù)新消息就進(jìn)不來因?yàn)榍懊娴南⒍伦×岁?duì)列頭。換一批新消息之前要確保每一條都成功確認(rèn)或者用 try-finally 結(jié)構(gòu)確保在處理異常時(shí)也做出明確決策是確認(rèn)丟棄還是回滾重投。不能模棱兩可。7. 學(xué)習(xí)路徑建議與后續(xù)擴(kuò)展方向如果你剛開始學(xué) JMS 和 ActiveMQ我的建議路徑是這樣先把官方 Quick Start 跑一遍然后自己寫一個(gè)生產(chǎn)者、兩個(gè)消費(fèi)者同時(shí)監(jiān)聽隊(duì)列觀察消息如何分?jǐn)?。再寫一個(gè) Topic 的發(fā)布訂閱 demo對比兩種模型的行為差異。接著把 ACK 模式、事務(wù)模式、持久化三個(gè)可靠性維度分別驗(yàn)證一遍基本概念就牢固了。后續(xù)擴(kuò)展方面有幾個(gè)方向值得繼續(xù)深入第一個(gè)是 Spring 集成。Spring 提供了 JmsTemplate 和JmsListener注解封裝了大部分模板代碼。用 Spring Boot 做集成時(shí)需要配置 ConnectionFactory、JmsListenerContainerFactory然后一個(gè)注解就能監(jiān)聽隊(duì)列消息代碼非常簡潔。但從學(xué)習(xí)角度我強(qiáng)烈建議先寫好原生 JMS API 再進(jìn) Spring因?yàn)槟憷斫饬说讓佑每蚣軙r(shí)才能明白它替你做了哪些事遇到問題才知道去哪里排查。第二個(gè)是失效轉(zhuǎn)移和集群。ActiveMQ 可以配置共享存儲下的主從模式或者網(wǎng)絡(luò)連接器Network Bridge做多個(gè) Broker 互聯(lián)形成集群。集群引入后要考慮的不僅是消息怎么發(fā)送還有消費(fèi)者負(fù)載的均勻分配、訂閱恢復(fù)的機(jī)制。第三個(gè)是吞吐量壓測。在把系統(tǒng)放上線之前最好對隊(duì)列做一輪簡單的壓測把生產(chǎn)端的發(fā)送速度、消費(fèi)端的處理速度、消息大小都記錄下來然后再決定線程池大小、事務(wù)批次大小這些參數(shù)。我做完這個(gè)項(xiàng)目后最大的感觸是JMS 和 ActiveMQ 本身并不難難的是想清楚每一層機(jī)制在解決什么問題。ACK 是為了“消費(fèi)結(jié)果可感知”事務(wù)是為了“多消息操作可原子”持久化是為了“Broker 掛了數(shù)據(jù)不丟”。把這些機(jī)制放到真實(shí)業(yè)務(wù)里去理解代碼怎么寫都順手。