ActiveMQ 實戰：從安裝部署到生產者消費者完整代碼實現

Apache ActiveMQ 是最流行的開源 JMS 提供者，支持多種協議、集群、持久化、Spring 整合。本文記錄 Linux 下安裝啟動、Web 控制台使用、Maven 依賴、生產者/消費者完整代碼（含事務、線程局部變量優化）、多線程壓測驗證及核心特性選型指南。

一、安裝與啟動

1. 下載解壓

wget https://archive.apache.org/dist/activemq/5.15.15/apache-activemq-5.15.15-bin.tar.gz
tar -zxvf apache-activemq-5.15.15-bin.tar.gz
cd apache-activemq-5.15.15

2. 目錄結構

3. 啟動服務

./bin/activemq start   # 後台啟動
# 或前台調試
./bin/activemq console

• 默認端口：61616 (OpenWire)、8161 (Web Console/Jetty)。
• Web Console：https://host:8161/admin/，默認用戶/密碼 admin/admin。

二、Maven 依賴

    org.apache.activemq
    activemq-client
    5.15.15


    org.apache.activemq
    activemq-pool
    5.15.15



    org.springframework
    spring-jms
    5.3.x

三、生產者實現（支持事務、線程安全複用 Session/Producer）

public class Producer {
    private static final String URL = "tcp://localhost:61616";
    private static final String USER = "admin";
    private static final String PASS = "admin";

    private ConnectionFactory factory;
    private Connection conn;
    private Session session;
    private final ThreadLocal producerLocal = new ThreadLocal<>();
    private final AtomicInteger counter = new AtomicInteger();

    public void init() throws JMSException {
        factory = new ActiveMQConnectionFactory(USER, PASS, URL);
        conn = factory.createConnection();
        conn.start();
        // 事務會話：手動 commit/rollback
        session = conn.createSession(true, Session.SESSION_TRANSACTED);
    }

    public void send(String queueName) {
        try {
            Queue queue = session.createQueue(queueName);
            MessageProducer producer = producerLocal.get();
            if (producer == null) {
                producer = session.createProducer(queue);
                producerLocal.set(producer);
            }

            while (true) {
                Thread.sleep(1000);
                int num = counter.getAndIncrement();
                TextMessage msg = session.createTextMessage(
                    Thread.currentThread().getName() + " 生產消息 count:" + num);
                producer.send(msg);
                session.commit(); // 事務提交
                System.out.println("已發送: " + msg.getText());
            }
        } catch (Exception e) {
            try { session.rollback(); } catch (JMSException ignored) {}
            e.printStackTrace();
        }
    }

    public void close() {
        try {
            if (session != null) session.close();
            if (conn != null) conn.close();
        } catch (JMSException e) { e.printStackTrace(); }
    }
}

四、消費者實現（自動確認、線程安全）

public class Consumer {
    private static final String URL = "tcp://localhost:61616";
    private static final String USER = "admin";
    private static final String PASS = "admin";

    private ConnectionFactory factory;
    private Connection conn;
    private Session session;
    private final ThreadLocal consumerLocal = new ThreadLocal<>();
    private final AtomicInteger counter = new AtomicInteger();

    public void init() throws JMSException {
        factory = new ActiveMQConnectionFactory(USER, PASS, URL);
        conn = factory.createConnection();
        conn.start();
        // 自動確認模式：消息處理完成自動 ACK
        session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }

    public void receive(String queueName) {
        try {
            Queue queue = session.createQueue(queueName);
            MessageConsumer consumer = consumerLocal.get();
            if (consumer == null) {
                consumer = session.createConsumer(queue);
                consumerLocal.set(consumer);
            }

            while (true) {
                Thread.sleep(1000);
                TextMessage msg = (TextMessage) consumer.receive(1000); // 1s 超時
                if (msg != null) {
                    System.out.println(Thread.currentThread().getName() 
                        + " 消費: " + msg.getText() + " ---> " + counter.getAndIncrement());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void close() {
        try {
            if (session != null) session.close();
            if (conn != null) conn.close();
        } catch (JMSException e) { e.printStackTrace(); }
    }
}

五、多線程壓測驗證

生產者測試（5 線程並發）

public class ProducerTest {
    public static void main(String[] args) throws Exception {
        Producer p = new Producer();
        p.init();
        Thread.sleep(1000);
        for (int i = 0; i < 5; i++) {
            new Thread(p::send, "P-" + i).start();
        }
    }
}

輸出片段：

P-0 生產消息 count:0
P-1 生產消息 count:1
P-2 生產消息 count:2
...

消費者測試（4 線程並發）

public class ConsumerTest {
    public static void main(String[] args) throws Exception {
        Consumer c = new Consumer();
        c.init();
        for (int i = 0; i < 4; i++) {
            new Thread(() -> c.receive("Jaycekon-MQ"), "C-" + i).start();
        }
    }
}

輸出片段：

C-0 消費: P-2 生產消息 count:4 ---> 0
C-1 消費: P-0 生產消息 count:5 ---> 1
...

六、Web Console 驗證

訪問 https://localhost:8161/admin/queues.jsp 查看：

• 隊列 Jaycekon-MQ 消息數、入隊/出隊速率。
• 點擊「Browse」查看消息內容、JMS 屬性。
• 「Delete」清理積壓測試數據。

七、ActiveMQ 核心特性與選型建議

八、適用場景

1. 異構系統集成：跨語言、跨平台、多協議適配。
2. 服務解耦：生產者不關心消費者，降低耦合，提升擴展性。
3. 前後端隔離/安全區隔離：DMZ 區通過 MQ 傳遞指令，避免直接數據庫訪問。
4. 削峰填谷：秒殺、大促流量緩衝，異步處理訂單、短信、積分。

九、JMS 參考

JMS 基礎模型（P2P/發布訂閱、Connection/Session/Producer/Consumer/Listener）詳見拙作：Java消息隊列--JMS概述