您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

JMS 核心模型深度解析：P2P 與 Pub/Sub、編程 API、同步異步消費

編輯：Java 相關知識

Java Message Service (JMS) 是 Java EE 標準消息 API，實現應用間解耦、異步通訊、削峰填谷。本文系統梳理 JMS 兩大消息模型（點對點、發布訂閱）、優勢特性、編程模型六大核心接口、消息消費兩種模式及代碼範例。

一、JMS 核心優勢

異步：消費者無需輪詢，消息到達自動觸發 onMessage()，降低耦合與延遲。
可靠：PERSISTENT 消息 + 事務 + 確認機制，保證 Exactly Once 語義（需 Broker 支持）。

二、兩大消息模型

1. Point-to-Point (P2P) — 隊列模式


生產者 → Queue → 消費者 (單個)

一條消息僅一個消費者 處理（負載均衡）。
生產者/消費者時間解耦：消息持久化在隊列，消費者離線不影響發送，上線後拉取。
消費者處理成功需 ACK 確認，否則消息重回隊列或進入 DLQ。

2. Publish/Subscribe (Pub/Sub) — 主題模式


發布者 → Topic → 多個訂閱者 (廣播)

一條消息多個消費者 並行處理（廣播）。
時間耦合：訂閱者必須先創建訂閱並保持運行才能收到消息。
持久化訂閱：createDurableSubscriber()，訂閱者離線期間的消息會被保留，重新上線自動推送。

三、JMS 編程模型六大核心對象


ConnectionFactory (工廠) → Connection (連接) → Session (會話/事務上下文)
    ↓                                    ↓
Destination (Queue/Topic)          MessageProducer / MessageConsumer
                                                    ↓
                                            MessageListener (異步)

對象	職責	關鍵方法
ConnectionFactory	創建 Connection，JNDI 查找	`createConnection()`
Connection	封裝 TCP 連接，啟動/關閉	`start()`, `close()`, `createSession()`
Session	事務邊界、消息工廠、ACK 模式	`createProducer()`, `createConsumer()`, `createQueue/Topic()`, `commit/rollback()`
Destination	消息目的地：Queue/Topic	JNDI 查找或 Session 創建
MessageProducer	發送消息	`send(msg)`, `send(dest, msg)`, 設置優先級/持久化/TTL
MessageConsumer	同步接收消息	`receive()`, `receive(timeout)`, `receiveNoWait()`
MessageListener	異步回調接口	`onMessage(Message)`

四、消息確認模式

模式常量	行為	適用場景
`AUTO_ACKNOWLEDGE`	receive() 返回 / onMessage() 正常結束自動 ACK	允許極少量重複消費、高吞吐
`CLIENT_ACKNOWLEDGE`	調用 `message.acknowledge()` 手動確認（確認該消息及之前所有未確認）	業務需精確控制、批量確認
`DUPS_OK_ACKNOWLEDGE`	懶惰確認，允許重複，性能最好	非關鍵業務、日志收集、監控數據
`SESSION_TRANSACTED`	事務模式，session.commit() 統一確認	發送/接收多條消息需原子性

五、同步 vs 異步消費代碼對比

同步消費（阻塞輪詢）


Connection conn = factory.createConnection();
conn.start();
Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("test.queue");
MessageConsumer consumer = session.createConsumer(queue);

while (running) {
    Message msg = consumer.receive(1000); // 1s 超時
    if (msg instanceof TextMessage) {
        System.out.println("收到: " + ((TextMessage) msg).getText());
    }
}

異步消費（推薦生產）


Connection conn = factory.createConnection();
Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("test.topic");
MessageConsumer consumer = session.createConsumer(topic);

consumer.setMessageListener(message -> {
    try {
        if (message instanceof TextMessage) {
            System.out.println("異步收到: " + ((TextMessage) message).getText());
        }
    } catch (JMSException e) {
        // 異常處理：記錄日志、發送 DLQ、告警
        e.printStackTrace();
    }
});

conn.start(); // 必須啟動連接才開始分發
// 主線程可做其他事，或 await shutdown

六、完整 P2P 發送/接收示例（含 JNDI 註釋）


// 生產者
Context ctx = new InitialContext();
QueueConnectionFactory qcf = (QueueConnectionFactory) ctx.lookup("ConnectionFactory");
Queue queue = (Queue) ctx.lookup("queue/test");

QueueConnection conn = qcf.createQueueConnection();
QueueSession session = conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueSender sender = session.createSender(queue);

TextMessage msg = session.createTextMessage("Hello JMS");
sender.send(msg, DeliveryMode.PERSISTENT, 4, 0); // 持久化、優先級4、無過期
session.close();
conn.close();

// 消費者 (異步)
QueueConnection conn = qcf.createQueueConnection();
QueueSession session = conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueReceiver receiver = session.createReceiver(queue);
receiver.setMessageListener(msg -> { /* 處理 */ });
conn.start();