ActiveMQ 核心架構與 Java 整合實戰：P2P 隊列、Pub/Sub 主題、Spring Boot 自動配置最佳實踐

ActiveMQ 定位與核心模型

• JMS 1.1 規範完整實現，支援 Queue (P2P) 與 Topic (Pub/Sub) 兩種消息模型。
• 傳輸協議：OpenWire (TCP 默認 61616)、AMQP、STOMP、MQTT、HTTP/WS、HTTPS/SSL。
• 持久化：KahaDB (默認)、LevelDB、JDBC、Memory。
• 高可用：Master/Slave (共享存儲/複製)、Replicated LevelDB、JDBC Master/Slave。

兩大消息模型對比

Linux 快速安裝與驗證

# 1. 下載解壓
wget https://archive.apache.org/dist/activemq/5.18.3/apache-activemq-5.18.3-bin.tar.gz
tar -xzf apache-activemq-5.18.3-bin.tar.gz
cd apache-activemq-5.18.3

# 2. 啟動 (前台調試)
./bin/activemq console

# 3. 後台運行
./bin/activemq start

# 4. 驗證
ps -ef | grep activemq
# Web Console: https://host:8161/admin/  用戶/密碼: admin/admin

# 5. 修改端口 (conf/activemq.xml)
<transportConnectors>
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616"/>
</transportConnectors>

Spring Boot 3.x 整合 (推薦：spring-boot-starter-activemq)

依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<!-- 或 Spring JMS + ActiveMQ Pool -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
</dependency>

配置

spring:
  activemq:
    broker-url: tcp://localhost:61616
    user: admin
    password: admin
    pool:
      enabled: true
      max-connections: 50
      idle-timeout: 30000
  jms:
    template:
      default-destination: test.queue
    listener:
      acknowledge-mode: auto
      concurrency: 5-20

發送端

@Component @RequiredArgsConstructor
public class OrderProducer {
    private final JmsTemplate jmsTemplate;

    public void sendOrderCreated(OrderEvent event) {
        jmsTemplate.convertAndSend("order.created", event, msg -> {
            msg.setJMSType("ORDER_CREATED");
            msg.setStringProperty("traceId", TraceContext.getTraceId());
            return msg;
        });
    }
}

接收端

@Component
public class OrderConsumer {
    private static final Logger log = LoggerFactory.getLogger(OrderConsumer.class);

    @JmsListener(destination = "order.created", containerFactory = "jmsListenerContainerFactory")
    public void onOrderCreated(OrderEvent event, @Header("traceId") String traceId) {
        log.info("[{}] 收到訂單創建事件: {}", traceId, event);
        // 業務處理...
    }
}

配置類 (進階定制)

@Configuration @EnableJms
public class JmsConfig {
    @Bean
    public ActiveMQConnectionFactory connectionFactory(@Value("${spring.activemq.broker-url}") String url,
                                                       @Value("${spring.activemq.user}") String user,
                                                       @Value("${spring.activemq.password}") String pass) {
        ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(user, pass, url);
        cf.setTrustAllPackages(true); // 允許反序列化所有包，生產建議白名單
        return cf;
    }

    @Bean
    public JmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory cf) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(cf);
        factory.setConcurrency("5-20");
        factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE); // 手動確認，支援重試
        return factory;
    }
}

原生 JMS API 代碼對照 (理解底層)

// P2P 發送
ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");
try (Connection conn = cf.createConnection();
     Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
    Queue queue = session.createQueue("test.queue");
    MessageProducer producer = session.createProducer(queue);
    producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    TextMessage msg = session.createTextMessage("Hello Queue");
    producer.send(msg);
}

// Topic 持久訂閱接收
try (Connection conn = cf.createConnection()) {
    conn.setClientID("client-123"); // 持久訂閱必須
    Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic("news.topic");
    MessageConsumer consumer = session.createDurableSubscriber(topic, "sub-001");
    conn.start();
    Message msg = consumer.receive(5000);
}

監控與運維關鍵指標

生產級最佳實踐清單

1. 連接池化：必開 `spring.activemq.pool.enabled=true`，避免頻繁創建連接。
2. 持久化策略：關鍵業務 `DeliveryMode.PERSISTENT` + 消費者 `CLIENT_ACKNOWLEDGE` 手動確認，業務成功再 ack。
3. 冪等消費：消息帶唯一 `messageId`，消費者基於 Redis/DB 去重。
4. 死信處理：配置 `deadLetterStrategy`，超過重試次數進 DLQ，人工/定時任務補償。
5. 延遲/定時消息：ActiveMQ 5.x 需插件或利用 `scheduledDelivery` 屬性；ActiveMQ Artemis 原生支援。
6. 遷移建議：新專案優先考慮 **Apache RocketMQ** 或 **Apache Pulsar** / **Kafka**，生態更活躍、性能更強、雲原生支援更好。