ActiveMQ 整合 Spring 實戰：消息隊列配置與生產消費模式

Spring 整合 ActiveMQ 完整指南：從依賴到異步消息收發

Apache ActiveMQ 是流行的開源消息中間件，完美支持 JMS 1.1 規範。本文詳細講解在 Spring 環境下整合 ActiveMQ 的完整配置，涵蓋點對點、發布訂閱兩種消息模型，並提供生產者/消費者代碼示例與常見問題排查。

一、Maven 依賴

<dependencies>
  <!-- Spring JMS 支持 -->
  <dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
    <version>5.3.21</version>
  </dependency>
  <!-- ActiveMQ 客戶端 -->
  <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
    <version>5.17.4</version>
  </dependency>
  <!-- 連接池優化 -->
  <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
    <version>5.17.4</version>
  </dependency>
</dependencies>

二、Spring XML 配置 (applicationContext-jms.xml)

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="https://www.springframework.org/schema/beans"
       xmlns:xsi="https://www.w3.org/2001/XMLSchema-instance"
       xmlns:jms="https://www.springframework.org/schema/jms"
       xsi:schemaLocation="
         https://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
         https://www.springframework.org/schema/jms https://www.springframework.org/schema/jms/spring-jms.xsd">

  <!-- 1. 真實連接工廠 -->
  <bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616"/>
    <property name="userName" value="admin"/>
    <property name="password" value="admin"/>
  </bean>

  <!-- 2. 連接池包裝（生產環境必須） -->
  <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
    <property name="connectionFactory" ref="amqConnectionFactory"/>
    <property name="maxConnections" value="100"/>
    <property name="maximumActiveSessionPerConnection" value="500"/>
  </bean>

  <!-- 3. Spring JMS 模板（單例，線程安全） -->
  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="pooledConnectionFactory"/>
    <property name="defaultDestinationName" value="queue.test"/>
    <property name="pubSubDomain" value="false"/> <!-- false=Queue, true=Topic -->
    <property name="deliveryPersistent" value="true"/>
  </bean>

  <!-- 4. 消費者監聽容器 -->
  <jms:listener-container connection-factory="pooledConnectionFactory" acknowledge="auto">
    <jms:listener destination="queue.test" ref="messageListener" method="onMessage"/>
  </jms:listener-container>

  <!-- 5. 業務監聽器 Bean -->
  <bean id="messageListener" class="com.example.jms.MessageListenerImpl"/>

</beans>

三、生產者代碼

@Component
public class MessageProducer {

    @Resource
    private JmsTemplate jmsTemplate;

    public void sendTextMessage(final String text) {
        jmsTemplate.send(session -> session.createTextMessage(text));
    }

    public void sendObjectMessage(final Serializable obj) {
        jmsTemplate.convertAndSend(obj); // 依賴 MessageConverter
    }

    public void sendToQueue(String queueName, String text) {
        jmsTemplate.convertAndSend(queueName, text);
    }
}

四、消費者代碼

@Component("messageListener")
public class MessageListenerImpl {

    private static final Logger LOG = LoggerFactory.getLogger(MessageListenerImpl.class);

    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                String text = ((TextMessage) message).getText();
                LOG.info("收到文本消息: {}", text);
                // 業務處理邏輯
                processBusiness(text);
            } else if (message instanceof ObjectMessage) {
                Serializable obj = ((ObjectMessage) message).getObject();
                LOG.info("收到對象消息: {}", obj);
            }
        } catch (JMSException e) {
            LOG.error("消息處理失敗", e);
            throw new RuntimeException(e); // 觸發重試/死信隊列
        }
    }

    private void processBusiness(String text) {
        // 具體業務邏輯
    }
}

五、發布/訂閱模型配置差異

六、常見問題與優化

1. 連接泄漏：必須使用 PooledConnectionFactory，並配置 maxConnections；關閉應用時調用 destroy() 釋放資源。
2. 消息丟失：生產端設置 deliveryPersistent=true；消費端使用事務會話或手動 ACK（acknowledge="transacted" 或 client）。
3. 性能調優：prefetchPolicy.queuePrefetch=1000 減少網絡往返；異步發送 jmsTemplate.setExplicitQosEnabled(true) + setDeliveryMode(DeliveryMode.NON_PERSISTENT) 用於非關鍵日志類消息。
4. 死信隊列 (DLQ)：配置 DeadLetterStrategy，超過重試次數的消息自動轉入 ActiveMQ.DLQ 便於排查。
5. Spring Boot 整合：僅需 spring-boot-starter-activemq，配置 spring.activemq.broker-url 即可自動裝配。

七、註解驅動配置 (Java Config)

@Configuration
@EnableJms
public class JmsConfig {

    @Value("${activemq.broker-url:tcp://localhost:61616}")
    private String brokerUrl;

    @Bean
    public ActiveMQConnectionFactory connectionFactory() {
        ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
        cf.setBrokerURL(brokerUrl);
        return cf;
    }

    @Bean
    public PooledConnectionFactory pooledConnectionFactory() {
        PooledConnectionFactory pcf = new PooledConnectionFactory();
        pcf.setConnectionFactory(connectionFactory());
        pcf.setMaxConnections(100);
        return pcf;
    }

    @Bean
    public JmsTemplate jmsTemplate() {
        JmsTemplate jt = new JmsTemplate();
        jt.setConnectionFactory(pooledConnectionFactory());
        jt.setPubSubDomain(false);
        return jt;
    }

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(pooledConnectionFactory());
        factory.setConcurrency("3-10");
        return factory;
    }
}

// 消費者改用 @JmsListener
@Component
public class AnnoListener {
    @JmsListener(destination = "queue.test", containerFactory = "jmsListenerContainerFactory")
    public void handle(String text) {
        System.out.println("收到: " + text);
    }
}

八、總結

Spring + ActiveMQ 整合核心在於：連接工廠 → 連接池 → JmsTemplate/ListenerContainer 三層封裝。掌握 Queue/Topic 兩種模型差異、持久化與確認機制、連接池參數調優，即可構建高可靠異步消息系統。新項目建議直接使用 Spring Boot Starter 簡化配置。