點選上方“芋道原始碼”，選擇“置頂公眾號”

技術文章第一時間送達！

原始碼精品專欄

摘要: 原創出處 https://juejin.im/post/5ad46f34518825651d08265c 「預流」歡迎轉載，保留摘要，謝謝！

簡介

ActiveMQ 特點

ActiveMQ 是由 Apache 出品的一款開源訊息中介軟體，旨在為應用程式提供高效、可擴充套件、穩定、安全的企業級訊息通訊。它的設計標的是提供標準的、面向訊息的、多語言的應用整合訊息通訊中介軟體。ActiveMQ 實現了 JMS 1.1 並提供了很多附加的特性，比如 JMX 管理、主從管理、訊息組通訊、訊息優先順序、延遲接收訊息、虛擬接收者、訊息持久化、訊息佇列監控等等。其主要特性有：

支援包括 Java、C、C++、C#、Ruby、Perl、Python、PHP 等多種語言的客戶端和協議。協議包含 OpenWire、Stomp、AMQP、MQTT 。
提供了像訊息組通訊、訊息優先順序、延遲接收訊息、虛擬接收者、訊息持久化之類的高階特性
完全支援 JMS 1.1 和 J2EE 1.4規範（包括持久化、分散式事務訊息、事務）
對 Spring 框架的支援，ActiveMQ 可以透過 Spring 的配置檔案方式很容易嵌入到 Spring 應用中
透過了常見的 J2EE 伺服器測試，比如 TomEE、Geronimo、JBoss、GlassFish、WebLogic
連線方式的多樣化，ActiveMQ 提供了多種連線樣式，例如 in-VM、TCP、SSL、NIO、UDP、多播、JGroups、JXTA
支援透過使用 JDBC 和 journal 實現訊息的快速持久化
為高效能叢集、客戶端-伺服器、點對點通訊等場景而設計
提供了技術和語言中立的 REST API 介面
支援 Ajax 方式呼叫 ActiveMQ
ActiveMQ 可以輕鬆地與 CXF、Axis 等 Web Service 技術整合，以提供可靠的訊息傳遞
可用作為記憶體中的 JMS 提供者，非常適合 JMS 單元測試

基本概念

因為 ActiveMQ 是完整支援 JMS 1.1 的，所以從 Java 使用者的角度其基本概念與 JMS 1.1 規範是一致的。

訊息傳送模型

點對點模型（Point to Point）使用佇列（Queue）作為訊息通訊載體，滿足生產者與消費者樣式，一條訊息只能被一個消費者使用，未被消費的訊息在佇列中保留直到被消費或超時。
釋出訂閱模型（Pub/Sub）使用主題作為訊息通訊載體，類似於廣播樣式，釋出者釋出一條訊息，該訊息透過主題傳遞給所有的訂閱者，在一條訊息廣播之後才訂閱的使用者則是收不到該條訊息的。

基本元件

ActiveMQ 使用時包含的基本元件各與 JMS 是相同的：

Broker，訊息代理，表示訊息佇列伺服器物體，接受客戶端連線，提供訊息通訊的核心服務。
Producer，訊息生產者，業務的發起方，負責生產訊息並傳輸給 Broker 。
Consumer，訊息消費者，業務的處理方，負責從 Broker 獲取訊息併進行業務邏輯處理。
Topic，主題，釋出訂閱樣式下的訊息統一彙集地，不同生產者向 Topic 傳送訊息，由 Broker 分發到不同的訂閱者，實現訊息的廣播。
Queue，佇列，點對點樣式下特定生產者向特定佇列傳送訊息，消費者訂閱特定佇列接收訊息併進行業務邏輯處理。
Message，訊息體，根據不同通訊協議定義的固定格式進行編碼的資料包，來封裝業務資料，實現訊息的傳輸。

由於這些概念在 JMS 中已介紹過，這裡不再詳細介紹。

聯結器

ActiveMQ Broker 的主要作用是為客戶端應用提供一種通訊機制，為此 ActiveMQ 提供了一種連線機制，並用聯結器（connector）來描述這種連線機制。ActiveMQ 中聯結器有兩種，一種是用於客戶端與訊息代理伺服器（client-to-broker）之間通訊的傳輸聯結器（transport connector），一種是用於訊息代理伺服器之間（broker-to-broker）通訊的網路聯結器（network connector）。connector 使用 URI（統一資源定位符）來表示，URI 格式為：:[?][#]schema name 表示協議，例如：foo://username:password@example.com:8042/over/there/index.dtb?type=animal&name;=narwhal#nose

其中 schema name 部分是 foo，hierarchical part 是 username:password@example.com:8042/over/there/index.dtb，query 是 type=animal&name;=narwhal，fragment 是 nose。

傳輸聯結器為了交換訊息，訊息生產者和訊息消費者（統稱為客戶端）都需要連線到訊息代理伺服器，這種客戶端和訊息代理伺服器之間的通訊就是透過傳輸聯結器（Transport connectors）完成的。很多情況下使用者連線訊息代理時的需求側重點不同，有的更關註效能，有的更註重安全性，因此 ActiveMQ 提供了一系列l連線協議供選擇，來改寫這些使用場景。從訊息代理的角度看，傳輸聯結器就是用來處理和監聽客戶端連線的，檢視 ActiveMQ demo 的配置檔案（/examples/conf/activemq-demo.xml），傳輸連線的相關配置如下：

        <transportConnectors>
            <transportConnector name="openwire" uri="tcp://localhost:61616" discoveryUri="multicast://default"/>
            <transportConnector name="ssl" uri="ssl://localhost:61617"/>
            <transportConnector name="stomp" uri="stomp://localhost:61613"/>
            <transportConnector name="ws" uri="ws://localhost:61614/" />
        transportConnectors>

傳輸聯結器定義在元素中，一個元素定義一個特定的聯結器，一個聯結器必須有自己唯一的名字和 URI 屬性，但discoveryUri屬性是可選的。目前在 ActiveMQ 最新的5.15版本中常用的傳輸聯結器連線協議有：vm、tcp、udp、multicast、nio、ssl、http、https、websocket、amqp、mqtt、stomp 等等

vm，允許客戶端和訊息伺服器直接在 VM 內部通訊，採用的連線不是 Socket 連線，而是直接的虛擬機器本地方法呼叫，從而避免網路傳輸的開銷。應用場景僅限於伺服器和客戶端在同一 JVM 中。
tcp，客戶端透過 TCP 連線到遠端的訊息伺服器。
udp，客戶端透過 UDP 連線到遠端的訊息伺服器。
multicast，允許使用組播傳輸的方式連線到訊息伺服器。
nio，nio 和 tcp 的作用是一樣的，只不過 nio 使用了 java 的 NIO包，這可能在某些場景下可提供更好的效能。
ssl，ssl 允許使用者在 TCP 的基礎上使用 SSL 。
http 和 https，允許客戶端使用 REST 或 Ajax 的方式進行連線，這意味著可以直接使用 Javascript 向 ActiveMQ 傳送訊息。
websocket，允許客戶端透過 HTML5 中的 WebSocket 方式連線到訊息伺服器。
amqp，5.8版本開始支援。
mqtt、stomp，5.6版本開始支援。

每個協議的具體配置見官網（http://activemq.apache.org/uri-protocols.html ）。除了以上這些基本協議之外 ActiveMQ 還支援一些高階協議也可以透過 URI 的方式進行配置，比如 Failover 和 Fanout 。

Failover 是一種重新連線的機制，工作於上面介紹的連線協議的上層，用於建立可靠的傳輸。其配置語法允許制定任意多個複合的 URI ，它會自動選擇其中的一個 URI 來嘗試建立連線，如果該連線沒有成功，則會繼續選擇其它的 URI 來嘗試。配置語法例如：failover:(tcp://localhost:61616,tcp://remotehost:61616)?initialReconnectDelay=100
Fanout 是一種重新連線和複製的機制，它也工作於其它連線的上層，採用複製的方式把訊息複製到多個訊息伺服器。配置語法例如：fanout:(tcp://localhost:61629,tcp://localhost:61639,tcp://localhost:61649)

網路聯結器很多情況下，我們要處理的資料可能是海量的，這種場景單臺伺服器很難支撐，這就要用到叢集功能，為此 ActiveMQ 提供了網路連線的樣式，簡單說就是透過把多個訊息伺服器實體連線在一起作為一個整體對外提供服務，從而提高整體對外的訊息服務能力。透過這種方式連線在一起的伺服器實體之間可共享佇列和消費者串列，從而達到分散式佇列的目的，網路聯結器就是用來配置伺服器之間的通訊。

如圖所示，伺服器 S1 和 S2 透過 NewworkConnector 相連，生產者 P1 傳送的訊息，消費者 C3 和 C4 都可以接收到，而生產者 P3 傳送的訊息，消費者 C1 和 C2 也可以接收到。要使用網路聯結器的功能需要在伺服器 S1 的 activemq.xml 中的 broker 節點下新增如下配置（假設192.168.11.23:61617 為 S2 的地址）：

<networkConnectors>      
          <networkConnector uri="static:(tcp://192.168.11.23:61617)"/>    
networkConnectors>

如果只是這樣，S1 可以將訊息傳送到 S2，但這隻是單方向的通訊，傳送到 S2 上的的訊息還不能傳送到 S1 上。如果想 S1 也收到從 S2 發來的訊息需要在 S2 的 activemq.xml 中的 broker 節點下也新增如下配置（假設192.168.11.45:61617為 S1 的地址）：

<networkConnectors>      
          <networkConnector uri="static:(tcp://192.168.11.45:61617)"/>    
networkConnectors>

這樣，S1和S2就可以雙向通訊了。目前在 ActiveMQ 最新的5.15版本中常用的網路聯結器協議有 static 和 multicast 兩種。

static，靜態協議，用於為一個網路中多個代理建立靜態配置，這種配置協議支援複合的 URI （即包含其他 URI 的 URI）。例如static://(tcp://ip:61616,tcp://ip2:61616)
multicast，多點傳送協議，訊息伺服器會廣播自己的服務，也會定位其他代理。這種方式用於伺服器之間實現動態識別，而不是配置靜態的 IP 組。

對這塊感興趣的話可以看官方檔案：http://activemq.apache.org/networks-of-brokers.html

訊息儲存

JMS 規範中訊息的分發方式有兩種：非持久化和持久化。對於非持久化訊息 JMS 實現者須保證盡最大努力分發訊息，但訊息不會持久化儲存；而持久化方式分發的訊息則必須進行持久化儲存。非持久化訊息常用於傳送通知或實時資料，當你比較看重系統效能並且即使丟失一些訊息並不影響業務正常運作時可選擇非持久化訊息。持久化訊息被髮送到訊息伺服器後如果當前訊息的消費者並沒有執行則該訊息繼續存在，只有等到訊息被處理並被訊息消費者確認之後，訊息才會從訊息伺服器中刪除。

對以上這兩種方式 ActiveMQ 都支援，並且還支援透過快取在記憶體中的中間狀態訊息的方式來恢復訊息。概括起來看 ActiveMQ 的訊息儲存有三種：儲存到記憶體、儲存到檔案、儲存到資料庫。具體使用上 ActiveMQ 提供了一個外掛式的訊息儲存機制，類似於訊息的多點傳播，主要實現瞭如下幾種：

AMQ，是 ActiveMQ 5.0及以前版本預設的訊息儲存方式，它是一個基於檔案的、支援事務的訊息儲存解決方案。在此方案下訊息本身以日誌的形式實現持久化，存放在 Data Log 裡。並且還對日誌裡的訊息做了取用索引，方便快速取回訊息。
KahaDB，也是一種基於檔案並具有支援事務的訊息儲存方式，從5.3開始推薦使用 KahaDB 儲存訊息，它提供了比 AMQ 訊息儲存更好的可擴充套件性和可恢復性。
JDBC，基於 JDBC 方式將訊息儲存在資料庫中，將訊息存到資料庫相對來說比較慢，所以 ActiveMQ 建議結合 journal 來儲存，它使用了快速的快取寫入技術，大大提高了效能。
記憶體儲存，是指將所有要持久化的訊息放到記憶體中，因為這裡沒有動態的快取，所以需要註意設定訊息伺服器的 JVM 和記憶體大小。
LevelDB，5.6版本之後推出了 LevelDB 的持久化引擎，它使用了自定義的索引代替常用的 BTree 索引，其持久化效能高於 KahaDB，雖然預設的持久化方式還是 KahaDB，但是 LevelDB 將是趨勢。在5.9版本還提供了基於 LevelDB 和 Zookeeper 的資料複製方式，作為 Master-Slave 方式的首選資料複製方案。

工程實體

Java 訪問 ActiveMQ 實體

JMS 規範中傳遞訊息的方式有兩種，一種是點對點模型的佇列（Queue）方式，另一種是釋出訂閱模型的主題（Topic）方式。下麵看下用 ActiveMQ 以主題方式傳遞訊息的 Java 示例。

引入依賴

Java 工程中需要引入 ActiveMQ 包的依賴，jar 包版本同你安裝 ActiveMQ 版本一致即可：

    <dependency>
        <groupId>org.apache.activemqgroupId>

<artifactId>activemq-allartifactId>
<version>5.15.2version>
dependency>

訊息生產者

package org.study.mq.activeMQ;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class TopicPublisher {
    /**
     * 預設使用者名稱
     */
    public static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    /**
     * 預設密碼
     */
    public static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    /**
     * 預設連線地址
     */
    public static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    public static void main(String[] args) {
        //建立連線工廠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
        try {
            //建立連線
            Connection connection = connectionFactory.createConnection();
            //開啟連線
            connection.start();
            //建立會話，不需要事務
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //建立 Topic，用作消費者訂閱訊息
            Topic myTestTopic = session.createTopic("activemq-topic-test1");
            //訊息生產者
            MessageProducer producer = session.createProducer(myTestTopic);
            for (int i = 1; i <= 3; i++) {
                TextMessage message = session.createTextMessage("傳送訊息 " + i);
                producer.send(myTestTopic, message);
            }
            //關閉資源
            session.close();
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

在 Topic 樣式中訊息生產者是用於釋出訊息的，絕大部分程式碼與 Queue 樣式中相似，不同的是本例中基於 Session 建立的是主題（Topic），該主題作為消費者消費訊息的目的地。

訊息消費者

package org.study.mq.activeMQ;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class TopicSubscriber {
    /**
     * 預設使用者名稱
     */
    public static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    /**
     * 預設密碼
     */
    public static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    /**
     * 預設連線地址
     */
    public static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    public static void main(String[] args) {
        //建立連線工廠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
        try {
            //建立連線
            Connection connection = connectionFactory.createConnection();
            //開啟連線
            connection.start();
            //建立會話，不需要事務
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //建立 Topic
            Topic myTestTopic = session.createTopic("activemq-topic-test1");
            MessageConsumer messageConsumer = session.createConsumer(myTestTopic);
            messageConsumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        System.out.println("消費者1 接收到訊息：" + ((TextMessage) message).getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            MessageConsumer messageConsumer2 = session.createConsumer(myTestTopic);
            messageConsumer2.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        System.out.println("消費者2 接收到訊息：" + ((TextMessage) message).getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            MessageConsumer messageConsumer3 = session.createConsumer(myTestTopic);
            messageConsumer3.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        System.out.println("消費者3 接收到訊息：" + ((TextMessage) message).getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            //讓主執行緒休眠100秒，使訊息消費者物件能繼續存活一段時間從而能監聽到訊息
            Thread.sleep(100 * 1000);
            //關閉資源
            session.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

為了展示主題樣式中訊息廣播給多個訂閱者的功能，這裡建立了三個消費者物件並訂閱了同一個主題，比較特殊的是最後讓主執行緒休眠了一段時間，這麼做的目的是讓消費者物件能繼續存活，從而使控制檯能打印出監聽到的訊息內容。

啟動 ActiveMQ 伺服器

在 ActiveMQ 的 bin 目錄下直接執行activemq start即啟動了 ActiveMQ

執行 TopicSubscriber

需要先執行 TopicSubscriber 類的 main 方法，這樣釋出者釋出訊息的時候訂閱者才能接收到訊息，如果將執行順序倒過來則訊息先釋出出去但沒有任何訂閱者在執行，則看不到訊息被消費了。

執行 TopicPublisher

接著執行 TopicPublisher 類的 main 方法，向主題中釋出3條訊息，然後可以在 TopicSubscriber 後臺看到接收到的訊息內容：

Spring 整合 ActiveMQ

在實際專案中如果使用原生的 ActiveMQ API 開發顯然比較囉嗦，這中間建立連線工廠、建立連線之類程式碼完全可以抽取出來由框架統一做，這些事情 Spring 也想到了並幫我們做了。ActiveMQ 完全支援基於 Spring 的方式配置 JMS 客戶端和伺服器，下麵的例子展示一下在 Spring 中如何使用佇列樣式和主題樣式傳遞訊息。

引入依賴

<dependency>
    <groupId>org.apache.activemqgroupId>

<artifactId>activemq-allartifactId>
<version>5.15.2version>
dependency>
<dependency>
<groupId>org.springframeworkgroupId>
<artifactId>spring-jmsartifactId>
<version>4.3.10.RELEASEversion>
dependency>
<dependency>
<groupId>org.apache.activemqgroupId>
<artifactId>activemq-poolartifactId>
<version>5.15.0version>
dependency>

工程中除了 activemq 的包之外還要新增 Spring 支援 JMS 的包。由於 connection、session、producer 的建立會消耗大量系統資源，為此這裡使用 連線池 來複用這些資源，所以還要新增 activemq-pool 的依賴。

Spring 配置檔案


<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context-3.0.xsd">
    <context:component-scan base-package="org.study.mq.activeMQ.spring"/>
    <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
        <property name="connectionFactory">
            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                <property name="brokerURL">
                    <value>tcp://localhost:61616value>

property>
bean>
property>
<property name=“maxConnections” value=“100”>property>
bean>
<bean id=“cachingConnectionFactory” class=“org.springframework.jms.connection.CachingConnectionFactory”>
<property name=“targetConnectionFactory” ref=“jmsFactory”/>
<property name=“sessionCacheSize” value=“1”/>
bean>
<bean id=“jmsTemplate” class=“org.springframework.jms.core.JmsTemplate”>
<property name=“connectionFactory” ref=“cachingConnectionFactory”/>
<property name=“messageConverter”>
<bean class=“org.springframework.jms.support.converter.SimpleMessageConverter”/>
property>
bean>
<bean id=“testQueue” class=“org.apache.activemq.command.ActiveMQQueue”>
<constructor-arg name=“name” value=“spring-queue”/>
bean>
<bean id=“testTopic” class=“org.apache.activemq.command.ActiveMQTopic”>
<constructor-arg index=“0” value=“spring-topic”/>
bean>
<bean id=“queueListener” class=“org.study.mq.activeMQ.spring.QueueListener”/>
<bean id=“topic1Listener” class=“org.study.mq.activeMQ.spring.Topic1Listener”/>
<bean id=“topic2Listener” class=“org.study.mq.activeMQ.spring.Topic2Listener”/>
<bean id=“queueContainer”
class=“org.springframework.jms.listener.DefaultMessageListenerContainer”>
<property name=“connectionFactory” ref=“cachingConnectionFactory”/>
<property name=“destination” ref=“testQueue”/>
<property name=“messageListener” ref=“queueListener”/>
bean>
<bean id=“topic1Container”
class=“org.springframework.jms.listener.DefaultMessageListenerContainer”>
<property name=“connectionFactory” ref=“cachingConnectionFactory”/>
<property name=“destination” ref=“testTopic”/>
<property name=“messageListener” ref=“topic1Listener”/>
bean>
<bean id=“topic2Container”
class=“org.springframework.jms.listener.DefaultMessageListenerContainer”>
<property name=“connectionFactory” ref=“cachingConnectionFactory”/>
<property name=“destination” ref=“testTopic”/>
<property name=“messageListener” ref=“topic2Listener”/>
bean>
beans>

下麵的專案示例中的 Java 程式碼採用註解的方式，這也是現在很多程式員的習慣用法，所以在配置檔案一開始定義註解掃描包路徑org.study.mq.activeMQ.spring，您可以根據自己實際情況修改包名稱，本例中的所有 Java 程式碼都放在該包之下。

接下來定義了一個 JMS 工廠 bean，採用的是池化連線工廠類org.apache.activemq.pool.PooledConnectionFactory，實際就是對內部的 ActiveMQ 連線工廠增加了連線池的功能，從其內部配置可以看到就是對org.apache.activemq.ActiveMQConnectionFactory的功能封裝，而ActiveMQConnectionFactory類則比較熟悉了，就是上面 Java 訪問 ActiveMQ 示例一開始建立連線工廠時使用的類。brokerURL 屬性配置的就是連線伺服器的協議和伺服器地址。接下來的 cachingConnectionFactory 是實際專案程式碼中常用的，對連線工廠的又一層增強，使用連線的快取功能以提升效率，讀者可酌情選擇使用。

jmsTemplate 就是 Spring 解決 JMS 訪問時冗長重覆程式碼的方案，它需要配置的兩個主要屬性是 connectionFactory 和 messageConverter，透過 connectionFactory 獲取連線、會話等物件，messageConverter 則是配置訊息轉換器，因為通常訊息在傳送前和接收後都需要進行一個前置和後置處理，轉換器便進行這個工作。這樣實際程式碼直接透過 jmsTemplate 來傳送和接收訊息，而每次傳送接收訊息時建立連線工廠、建立連線、建立會話等工作都由 Spring 框架做了。

有了 JMS 模板還需要知道佇列和主題作為實際傳送和接收訊息的目的地，所以接下來定義了 testQueue 和 testTopic 作為兩種樣式的示例。而非同步接收訊息時則需要提供 MessageListener 的實現類，所以定義了 queueListener 作為佇列樣式下非同步接收訊息的監聽器，topic1Listener 和 topic2Listener 作為主題樣式下非同步接收訊息的監聽器，主題樣式用兩個監聽器是為了演示多個消費者時都能收到訊息。最後的 queueContainer、topic1Container、topic2Container 用於將訊息監聽器系結到具體的訊息目的地上。

訊息服務類

下麵是使用 JMS 模板處理訊息的訊息服務類

package org.study.mq.activeMQ.spring;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.jms.*;
@Service
public class MessageService {
    @Resource(name = "jmsTemplate")
    private JmsTemplate jmsTemplate;
    @Resource(name = "testQueue")
    private Destination testQueue;
    @Resource(name = "testTopic")
    private Destination testTopic;
    //向佇列傳送訊息
    public void sendQueueMessage(String messageContent) {
        jmsTemplate.send(testQueue, new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage msg = session.createTextMessage();
                // 設定訊息內容
                msg.setText(messageContent);
                return msg;
            }
        });
    }
    //向主題傳送訊息
    public void sendTopicMessage(String messageContent) {
        jmsTemplate.send(testTopic, new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage msg = session.createTextMessage();
                // 設定訊息內容
                msg.setText(messageContent);
                return msg;
            }
        });
    }
}

@Service 將該類宣告為一個服務，實際專案中很多服務程式碼也類似。透過 Resource 註解直接將上面配置檔案中定義的 jmsTemplate 引入到 MessageService 類中就可以直接使用了，testQueue 和 testTopic 也是類似，服務類中直接引入配置檔案中定義好的佇列和主題。重點是下麵的兩個傳送訊息的方法，sendQueueMessage 向佇列傳送訊息，sendTopicMessage 向主題傳送訊息，兩種樣式都使用了 jmsTemplate 的 send 方法，send 方法第1個引數是javax.jms.Destination型別，表示訊息目的地。由於javax.jms.Queue和javax.jms.Topic都繼承了javax.jms.Destination介面，所以該方法對佇列樣式和主題樣式都適用。send 方法的第2個引數是org.springframework.jms.core.MessageCreator，這裡使用了匿名內部類的方式建立物件，從支援的 Session 物件中建立文字訊息，這樣就可以傳送訊息了。可以看到無論是佇列還是主題，透過 Spring 框架來傳送訊息的程式碼比之前的 Java 程式碼示例簡潔了很多。

訊息監聽器類

package org.study.mq.activeMQ.spring;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class QueueListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            try {
                TextMessage txtMsg = (TextMessage) message;
                String messageStr = txtMsg.getText();
                System.out.println("佇列監聽器接收到文字訊息：" + messageStr);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        } else {
            throw new IllegalArgumentException("只支援 TextMessage 型別訊息！");
        }
    }
}

佇列訊息監聽器在收到訊息時校驗是否是文字訊息型別，是的話則打印出內容。

package org.study.mq.activeMQ.spring;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class Topic1Listener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            try {
                TextMessage txtMsg = (TextMessage) message;
                String messageStr = txtMsg.getText();
                System.out.println("主題監聽器1 接收到文字訊息：" + messageStr);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        } else {
            throw new IllegalArgumentException("只支援 TextMessage 型別訊息！");
        }
    }
}

package org.study.mq.activeMQ.spring;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class Topic2Listener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            try {
                TextMessage txtMsg = (TextMessage) message;
                String messageStr = txtMsg.getText();
                System.out.println("主題監聽器2 接收到文字訊息：" + messageStr);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        } else {
            throw new IllegalArgumentException("只支援 TextMessage 型別訊息！");
        }
    }
}

主題監聽器的程式碼與佇列監聽器類似，只是列印時透過不同字串表示當前是不同監聽器接收的訊息。

啟動應用

為了演示例子，寫了一個 StartApplication 類，在 main 方法中載入 Spring ，獲取到 MessageService 服務之後呼叫 sendQueueMessage 和 sendTopicMessage 方法傳送訊息。

package org.study.mq.activeMQ.spring;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class StartApplication {
    public static void main(String[] args) {
        ApplicationContext ctx = new ClassPathXmlApplicationContext("spring-context.xml");
        MessageService messageService = (MessageService) ctx.getBean("messageService");
        messageService.sendQueueMessage("我的測試訊息1");
        messageService.sendTopicMessage("我的測試訊息2");
        messageService.sendTopicMessage("我的測試訊息3");
    }
}

啟動好 activeMQ 服務之後執行 StartApplication 類，在控制檯看到接收到文字訊息：

佇列監聽器監聽到了一條訊息，兩個主題監聽器分別監聽到了兩條訊息。

知識星球

目前在知識星球（https://t.zsxq.com/2VbiaEu）更新瞭如下 Dubbo 原始碼解析如下：

01. 除錯環境搭建
02. 專案結構一覽
03. API 配置（一）之應用
04. API 配置（二）之服務提供者
05. API 配置（三）之服務消費者
06. 屬性配置
07. XML 配置
08. 核心流程一覽

09. 拓展機制 SPI

10. 執行緒池

…
一共 60 篇++

【訊息佇列 MQ 專欄】訊息佇列之 ActiveMQ

簡介

ActiveMQ 特點

基本概念

訊息傳送模型

基本元件

聯結器

訊息儲存

工程實體

Java 訪問 ActiveMQ 實體

引入依賴

訊息生產者

訊息消費者

啟動 ActiveMQ 伺服器

執行 TopicSubscriber

執行 TopicPublisher

Spring 整合 ActiveMQ

引入依賴

Spring 配置檔案

訊息服務類

訊息監聽器類

啟動應用

相關推薦

熱門標籤

熱門文章

分享創造快樂