本文基於 **Spring 4.1 + ActiveMQ 5.12 + Tomcat 7 + Maven** 實作完整的 JMS 整合範例，涵蓋 **Point-to-Point (Queue)** 與 **Publish/Subscribe (Topic)** 兩種模型、生產者/消費者服務、監聽器容器、REST 測試端點與壓力測試。
src/main/java
├── com.jayce
│ ├── config
│ │ ├── WebConfig.java # Spring MVC 配置
│ │ └── ActiveMQConfig.java # ActiveMQ 連線工廠、Queue/Topic、JmsTemplate、ListenerContainer
│ ├── controller
│ │ └── MessageController.java # REST 端點：發送/接收訊息
│ ├── service
│ │ ├── ProducerService.java # 發送訊息服務
│ │ └── ConsumerService.java # 手動接收訊息服務
│ └── listener
│ └── QueueMessageListener.java # 非同步監聽器 (實作 MessageListener)
@Configuration
@EnableJms
public class ActiveMQConfig {
@Value("${activemq.broker-url:tcp://localhost:61616}")
private String brokerUrl;
@Value("${activemq.user:admin}")
private String user;
@Value("${activemq.password:admin}")
private String password;
// 1. ActiveMQ 連線工廠
@Bean
public ActiveMQConnectionFactory amqConnectionFactory() {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
factory.setBrokerURL(brokerUrl);
factory.setUserName(user);
factory.setPassword(password);
factory.setTrustAllPackages(true); // 允許傳遞 ObjectMessage
return factory;
}
// 2. Spring 快取連線工廠 (效能關鍵)
@Bean
public CachingConnectionFactory connectionFactory() {
CachingConnectionFactory factory = new CachingConnectionFactory(amqConnectionFactory());
factory.setSessionCacheSize(100);
return factory;
}
// 3. 定義 Queue (點對點)
@Bean
public Queue demoQueue() {
return new ActiveMQQueue("jayce.queue.demo");
}
// 4. 定義 Topic (發布訂閱)
@Bean
public Topic demoTopic() {
return new ActiveMQTopic("jayce.topic.demo");
}
// 5. JmsTemplate (Queue 操作)
@Bean
public JmsTemplate jmsQueueTemplate() {
JmsTemplate template = new JmsTemplate(connectionFactory());
template.setDefaultDestination(demoQueue());
template.setReceiveTimeout(10000);
template.setPubSubDomain(false); // false=Queue, true=Topic
return template;
}
// 6. JmsTemplate (Topic 操作)
@Bean
public JmsTemplate jmsTopicTemplate() {
JmsTemplate template = new JmsTemplate(connectionFactory());
template.setDefaultDestination(demoTopic());
template.setPubSubDomain(true);
return template;
}
// 7. 非同步監聽器容器 (Queue)
@Bean
public DefaultMessageListenerContainer queueListenerContainer() {
DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setDestination(demoQueue());
container.setMessageListener(queueMessageListener());
container.setConcurrency("5-10"); // 5-10 消費者執行緒
return container;
}
@Bean
public QueueMessageListener queueMessageListener() {
return new QueueMessageListener();
}
}
@Service
public class ProducerService {
@Autowired @Qualifier("jmsQueueTemplate")
private JmsTemplate queueTemplate;
@Autowired @Qualifier("jmsTopicTemplate")
private JmsTemplate topicTemplate;
// 發送到預設 Queue
public void sendToQueue(final String msg) {
queueTemplate.send(session -> session.createTextMessage(msg));
}
// 發送到指定 Destination
public void sendToQueue(Destination dest, final String msg) {
queueTemplate.send(dest, session -> session.createTextMessage(msg));
}
// 發送到 Topic (廣播)
public void sendToTopic(final String msg) {
topicTemplate.send(session -> session.createTextMessage(msg));
}
}
@Service
public class ConsumerService {
@Autowired @Qualifier("jmsQueueTemplate")
private JmsTemplate queueTemplate;
public TextMessage receiveFromQueue(Destination dest) {
return (TextMessage) queueTemplate.receive(dest);
}
}
activemq:
broker-url: tcp://192.168.148.128:61616
user: admin
password: admin
server:
port: 8080
spring:
jms:
pub-sub-domain: false # 預設 Queue
# Linux/Mac
./bin/activemq start
# 管理主控台: https://localhost:8161/admin (admin/admin)
curl -X POST "https://localhost:8080/api/message/send-queue?msg=Hello%20Queue"
curl -X POST "https://localhost:8080/api/message/send-topic?msg=Hello%20Topic"
curl "https://localhost:8080/api/message/receive-queue"
發送後**無需呼叫 receive 端點**，後台日誌即顯示 `收到 Queue 訊息: ...`，證明 `DefaultMessageListenerContainer` 正常運作。
@Test
public void stressTest() throws Exception {
HttpClient client = HttpClients.createDefault();
int threadCount = 20;
int msgPerThread = 500;
CountDownLatch latch = new CountDownLatch(threadCount);
for (int t = 0; t < threadCount; t++) {
new Thread(() -> {
try {
for (int i = 0; i < msgPerThread; i++) {
HttpPost post = new HttpPost("https://localhost:8080/api/message/send-queue");
post.setEntity(new StringEntity("msg=" + UUID.randomUUID(), ContentType.APPLICATION_FORM_URLENCODED));
client.execute(post);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
latch.countDown();
}
}).start();
}
latch.await();
System.out.println("壓測完成: " + (threadCount * msgPerThread) + " 訊息");
}
觀察 ActiveMQ Enqueue/Dequeue 速率、JVM GC 情況、資料庫連線池使用率。
| 實踐 | 理由 |
|---|---|
| 使用 CachingConnectionFactory | 重用 Session/Producer，避免每次發送建立連線，效能提升 10x+ |
| 設定 ListenerContainer concurrency | 多執行緒並行消費，提升吞吐量 |
| 生產者用 JmsTemplate.send(MessageCreator) | 自動管理 Session 生命週期，異常自動回滾 |
| 啟用事務 (sessionTransacted=true) | 確保訊息處理與資料庫操作原子性 |
| 設定 Dead Letter Queue (DLQ) | 重試失敗訊息自動轉入 DLQ，避免丟失 |
| 監控 Queue 深度、消費延遲 | 及時發現積壓、擴容消費者 |
Spring + ActiveMQ 整合核心在於：**CachingConnectionFactory 重用連線、JmsTemplate 簡化發送/接收、DefaultMessageListenerContainer 非同步消費**。掌握 Queue/Topic 兩種模型差異、事務邊界、死信處理與監控指標，即可建構穩健的企業級訊息系統。
