Apache ActiveMQ 入門實戰：P2P 模式生產者/消費者完整示例

ActiveMQ 核心特性速覽

• 全面支持 JMS 1.1 / Java EE 1.4 規範（持久化、XA 事務、消息確認）。
• 多語言客戶端：Java, C/C++, C#, Ruby, Python, PHP 等。
< li>多協議：OpenWire, STOMP, REST, WS Notification, XMPP, AMQP。
• 高性能持久化：JDBC / 內建 Journal / LevelDB。
• 輕量級嵌入：可內嵌 Spring / Tomcat / JUnit 單元測試。

環境準備（Windows）

1. 下載 apache-activemq-5.x.x-bin.zip 解壓。
2. 雙擊 bin/activemq.bat 啟動 Broker（默認 TCP 61616，Web 控制台 https://localhost:8161/admin）。
3. Eclipse 新建 Java Project，導入 activemq-all-5.x.x.jar。

P2P (Queue) 模式完整代碼

生產者 MsgSender

public class MsgSender {
    private static final String URL = "tcp://localhost:61616";
    private static final String QUEUE = "myQueue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory(URL);
        Connection conn = cf.createConnection();
        conn.start();
        Session session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
        Destination dest = session.createQueue(QUEUE);
        MessageProducer producer = session.createProducer(dest);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

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

        producer.close(); session.close(); conn.close();
    }
}

消費者 MsgReceiver

public class MsgReceiver {
    private static final String URL = "tcp://localhost:61616";
    private static final String QUEUE = "myQueue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory(URL);
        Connection conn = cf.createConnection();
        conn.start();
        Session session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
        Destination dest = session.createQueue(QUEUE);
        MessageConsumer consumer = session.createConsumer(dest);

        while (true) {
            TextMessage msg = (TextMessage) consumer.receive(1000); // 1s 超時
            if (msg != null) {
                System.out.println("Received: " + msg.getText());
            } else break; // 隊列空
        }
        session.commit();
        consumer.close(); session.close(); conn.close();
    }
}

關鍵配置細節

進階主題（後續探索）

• Topic (Pub/Sub)：session.createTopic("myTopic")，支持持久訂閱 createDurableSubscriber。
• Spring 整合：jms:listener-container + @JmsListener 註解驅動。
• 消息選擇器：SQL92 語法過濾 selector="priority > 5"。
• 集群/高可用：Master-Slave (Shared JDBC/LevelDB) 或 Network of Brokers。
• 監控指標：Web Console 查看隊列深度、消費速率、死信隊列。

常見問題排查

1. 連接被拒：檢查 61616 端口防火牆、Broker 是否啟動、activemq.xml transportConnector 配置。
2. 消息堆積：消費者數量不足、預取限制 prefetchPolicy、慢消費者處理。
3. 重複消費：關閉事務/自動確認時網絡抖動導致，改用 CLIENT_ACKNOWLEDGE 手動確認。
4. 類加載錯誤：ObjectMessage 要求雙方類路徑一致，生產環境建議用 TextMessage/BytesMessage + JSON/Protobuf。