ActiveMQ 實戰指南：安裝部署、JMS/Queue/Topic 三種模式、Spring 整合與生產環境調優

前言：為什麼選擇 ActiveMQ？

Apache ActiveMQ 是最成熟、功能最全的開源消息中間件之一：

• 完整支持 JMS 1.1 / J2EE 1.4 規範
• 多語言客戶端：Java、C/C++、C#、Python、PHP、Ruby、Perl
• 多協議：OpenWire、STOMP、REST、WS Notification、XMPP、AMQP、MQTT
• 高可用：Master/Slave、Network of Brokers、JDBC/LevelDB/KahaDB 持久化
• 易運維：Web Console、JMX、REST API、與 Spring/JCA 深度整合

一、快速安裝與啟動

1.1 環境準備

• JDK 1.8+
• 下載：apache-activemq-5.17.x-bin.zip（官網最新版）

1.2 解壓與目錄結構

apache-activemq-5.17.0/
├── bin/            # activemq 脚本、activemq.bat
├── conf/           # activemq.xml、jetty.xml、log4j2.properties、users.properties
├── data/           # KahaDB 持久化文件、日誌
├── docs/           # 文檔
├── examples/       # 示例代碼
├── lib/            # 核心 jar、依賴 jar
├── webapps/        # admin console (hawtio)、demo
└── webapps-demo/   # 額外示例

1.3 啟動 Broker

# 前台運行（調試用）
bin/activemq console

# 後台運行（生產用）
bin/activemq start

# 指定配置文件
bin/activemq console xbean:file:/opt/activemq/conf/activemq.xml

成功啟動後訪問：

• Web Console: https://localhost:8161/admin/ (默認 admin/admin)
• JMX: service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi
• OpenWire 端口: tcp://localhost:61616

1.4 常見啟動問題

二、核心概念：三種消息模型

2.1 統一 API 體系

現代代碼請統一使用 JMS 1.1 統一接口（ConnectionFactory、Destination、Session、MessageProducer/Consumer），避免耦合特定域接口。

2.2 Point-to-Point (Queue) —— 負載均衡、單消費

• 一條消息僅被一個消費者處理
• 多消費者並發消費實現負載均衡
• 消費者離線時消息持久化保存，上線後接收
• 無時間依賴：生產者隨時發送，消費者隨時啟動消費

2.3 Publish/Subscribe (Topic) —— 廣播、多消費

• 一條消息分發給所有訂閱者
• 訂閱者必須在線才能收到消息（除非持久訂閱 Durable Subscription）
• 有時間依賴：訂閱創建後發送的消息才會收到
• 兩種接收方式：consumer.receive() 同步拉取 / setMessageListener() 異步回調

三、JMS 標準代碼範例（統一 API）

3.1 依賴坐標

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
    <version>5.17.0</version>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
    <version>5.17.0</version>
</dependency>  

3.2 生產者通用模板

public class JmsProducer {
    private static final String BROKER_URL = "tcp://localhost:61616";
    private static final String DESTINATION = "demo.queue";  // 或 demo.topic

    public static void main(String[] args) throws JMSException {
        // 1. 創建連接工廠（生產環境用 PooledConnectionFactory）
        ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
        // 生產環境建議：
        // PooledConnectionFactory pooled = new PooledConnectionFactory(factory);
        // pooled.setMaxConnections(100);

        try (Connection conn = factory.createConnection();
             Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE)) {

            conn.start();
            Destination dest = session.createQueue(DESTINATION);  // 或 createTopic
            MessageProducer producer = session.createProducer(dest);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 或 PERSISTENT

            for (int i = 1; i <= 5; i++) {
                TextMessage msg = session.createTextMessage("Message #" + i);
                producer.send(msg);
                System.out.println("Sent: " + msg.getText());
            }
        }
    }
}

3.3 消費者：同步拉取 vs 異步監聽

// 同步拉取（簡單、阻塞）
public class SyncConsumer {
    public static void main(String[] args) throws JMSException {
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        try (Connection conn = factory.createConnection();
             Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            conn.start();
            MessageConsumer consumer = session.createConsumer(session.createQueue("demo.queue"));
            while (true) {
                Message msg = consumer.receive(5000); // 5s 超時
                if (msg instanceof TextMessage) System.out.println("Received: " + ((TextMessage)msg).getText());
                else break;
            }
        }
    }
}

// 異步監聽（高併發、事件驅動）
public class AsyncConsumer implements MessageListener {
    public static void main(String[] args) throws JMSException {
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        Connection conn = factory.createConnection();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        conn.start();
        MessageConsumer consumer = session.createConsumer(session.createQueue("demo.queue"));
        consumer.setMessageListener(new AsyncConsumer()); // 註冊回調
        // 主線程保持存活
        new CountDownLatch(1).await();
    }
    @Override public void onMessage(Message msg) {
        try { if (msg instanceof TextMessage) System.out.println("Async: " + ((TextMessage)msg).getText()); }
        catch (JMSException e) { e.printStackTrace(); }
    }
}

四、Queue/Topic 專用 API 示例（兼容舊代碼）

若維護舊系統需使用 QueueConnectionFactory 等域接口，參考原文代碼結構，核心差異僅在創建對象類型：

• Queue: QueueConnectionFactory → QueueConnection → QueueSession → QueueSender/QueueReceiver
• Topic: TopicConnectionFactory → TopicConnection → TopicSession → TopicPublisher/TopicSubscriber

建議：新代碼統一遷移至 JMS 1.1 統一 API。

五、Spring 整合：聲明式配置與 JmsTemplate

5.1 Maven 依賴

<dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>5.3.x</version> </dependency>
<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-pool</artifactId> <version>5.17.0</version> </dependency>

5.2 Spring XML 配置

<beans xmlns="https://www.springframework.org/schema/beans"
       xmlns:xsi="https://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="https://www.springframework.org/schema/beans
       https://www.springframework.org/schema/beans/spring-beans.xsd">

    <!-- 1. 連接池工廠（生產必備） -->
    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
        <property name="connectionFactory">
            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                <property name="brokerURL" value="tcp://localhost:61616"/>
            </bean>
        </property>
        <property name="maxConnections" value="100"/>
    </bean>

    <!-- 2. 目標 Queue/Topic -->
    <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="spring.demo.queue"/>
    </bean>

    <!-- 3. JmsTemplate -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="pooledConnectionFactory"/>
        <property name="defaultDestination" ref="destination"/>
        <property name="messageConverter">
            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
        </property>
    </bean>

    <!-- 4. 異步監聽容器（替代手動 setMessageListener） -->
    <bean id="messageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="pooledConnectionFactory"/>
        <property name="destination" ref="destination"/>
        <property name="messageListener" ref="myListener"/>
        <property name="concurrentConsumers" value="5"/>
        <property name="maxConcurrentConsumers" value="20"/>
    </bean>

    <bean id="myListener" class="com.example.MyMessageListener"/>
</beans>

5.3 發送與接收代碼

// 發送
@Component
public class SpringSender {
    @Autowired private JmsTemplate jmsTemplate;
    public void send(String text) {
        jmsTemplate.send(session -> session.createTextMessage(text));
    }
}

// 接收（實現 MessageListener）
@Component
public class MyMessageListener implements MessageListener {
    @Override public void onMessage(Message msg) {
        if (msg instanceof TextMessage) {
            try { System.out.println("Spring Received: " + ((TextMessage)msg).getText()); }
            catch (JMSException e) { throw new RuntimeException(e); }
        }
    }
}

六、生產環境關鍵調優清單

七、常見坑點避坑指南

1. 忘記 conn.start()：Connection 默認不啟動消息分發，消費者收不到消息。
2. 事務會話未 commit()：createSession(true, ...) 時必須顯式提交，否則消息不入隊/不確認。
3. 持久化 Topic 訂閱需 DurableSubscriber：session.createDurableSubscriber(topic, "subName")，否則離線消息丟失。
4. 對象消息 ObjectMessage 反序列化風險：ActiveMQ 5.13+ 默認拒絕非白名單類，配置 trustAllPackages=true 或精確白名單。
5. 關閉資源順序：Consumer → Producer → Session → Connection，或用 try-with-resources/JmsTemplate 托管。

總結：ActiveMQ 憑藉成熟的 JMS 實現、豐富的協議支持、靈活的部署拓撲與 Spring 生態無縫整合，仍是中小型項目、傳統企業應用、混合語言系統的首選消息中間件。掌握「統一 API、連接池、持久化策略、預取調優、高可用架構」五大核心，即可應對 90% 生產場景。