JMS 發布/訂閱模式實戰：Tomcat + JNDI + ActiveMQ 實現 Topic 消息廣播

JMS 發布/訂閱模式實戰：Topic 消息廣播完整教程

基於上篇「點對點隊列」教程，僅需將 JNDI 資源類型從 Queue 改為 Topic，即可實現「一條消息、多個訂閱者同時收到」的廣播模式。

架構差異

步驟一：Tomcat JNDI 配置 Topic

編輯 conf/context.xml：

<Resource name="topic/connectionFactory" 
          auth="Container"
          type="org.apache.activemq.ActiveMQConnectionFactory"
          description="JMS Connection Factory"
          factory="org.apache.activemq.jndi.JNDIReferenceFactory"
          brokerURL="failover:(tcp://localhost:61616)?initialReconnectDelay=100&maxReconnectAttempts=5"
          brokerName="LocalActiveMQBroker"
          useEmbeddedBroker="false" />

<Resource name="topic/topic0" 
          auth="Container"
          type="org.apache.activemq.command.ActiveMQTopic"
          description="My Topic"
          factory="org.apache.activemq.jndi.JNDIReferenceFactory"
          physicalName="TestTopic" />

關鍵點：

• type 指定 ActiveMQTopic
• physicalName 為 ActiveMQ 內部 Topic 名稱
• brokerURL 使用 failover: 協議支持自動重連

步驟二：發布者 Servlet (Publisher)

package pubSub;
import javax.jms.*;
import javax.naming.InitialContext;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.*;
import java.io.*;

@WebServlet("/Publish")
public class Publisher extends HttpServlet {
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException {
        PrintWriter out = resp.getWriter();
        try {
            InitialContext ctx = new InitialContext();
            
            Topic topic = (Topic) ctx.lookup("java:comp/env/topic/topic0");
            TopicConnectionFactory factory = (TopicConnectionFactory) 
                ctx.lookup("java:comp/env/topic/connectionFactory");
            
            TopicConnection conn = factory.createTopicConnection();
            TopicSession session = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            TopicPublisher publisher = session.createPublisher(topic);
            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            
            TextMessage msg = session.createTextMessage("Hello Topic World @ " + System.currentTimeMillis());
            publisher.publish(msg);
            
            out.write("Message Published: " + msg.getText());
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
            out.write("Error: " + e.getMessage());
        }
    }
}

步驟三：訂閱者 Servlet (Subscriber)

package pubSub;
import javax.jms.*;
import javax.naming.InitialContext;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.*;
import java.io.*;

@WebServlet("/Subscribe")
public class Subscriber extends HttpServlet {
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException {
        PrintWriter out = resp.getWriter();
        try {
            InitialContext ctx = new InitialContext();
            Topic topic = (Topic) ctx.lookup("java:comp/env/topic/topic0");
            TopicConnectionFactory factory = (TopicConnectionFactory) 
                ctx.lookup("java:comp/env/topic/connectionFactory");
            
            TopicConnection conn = factory.createTopicConnection();
            TopicSession session = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            TopicSubscriber subscriber = session.createSubscriber(topic);
            
            conn.start(); // 必須啟動！
            TextMessage msg = (TextMessage) subscriber.receive(5000); // 5秒超時避免永久阻塞
            
            if (msg != null) 
                out.write("Message Received: " + msg.getText());
            else 
                out.write("Timeout: No message received");
            
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
            out.write("Error: " + e.getMessage());
        }
    }
}

驗證流程與截圖

1. 啟動 ActiveMQ、Tomcat、部署應用
2. 打開多個瀏覽器標籤同時訪問 /Subscribe (模擬多訂閱者)
3. 
4. 新標籤訪問 /Publish 發送消息
5. 
6. 所有訂閱者標籤同時收到消息
7. 

關鍵注意事項

1. 訂閱者必須先啟動：Topic 非持久化，發布時無訂閱者消息直接丟失
2. conn.start() 必須在創建 Subscriber 後調用，否則 receive() 永久阻塞
3. 生產環境用持久化訂閱：

   TopicSubscriber sub = session.createDurableSubscriber(topic, "clientId-unique");

   需配置 clientId 於連接工廠或連接上
4. 超時機制：receive(long timeout) 防止 HTTP 請求掛死
5. JNDI 名稱隔離：Queue 用 queue/ 前綴，Topic 用 topic/，互不干擾

擴展：Spring JMS 模板簡化代碼

@Autowired JmsTemplate jmsTemplate;

public void publish(String msg) {
    jmsTemplate.convertAndSend("TestTopic", msg);
}

@JmsListener(destination = "TestTopic", containerFactory = "topicListenerFactory")
public void receive(String msg) {
    System.out.println("Received: " + msg);
}

Spring Boot 只需 spring-boot-starter-activemq 自動配置，極大簡化開發。

參考資源

HowToDoInJava JMS Publish/Subscribe Example