Java訊息服務定義

Java訊息服務（Java Message Service）應用程式介面，是一個Java平臺中關於面向訊息中介軟體的API，用於在兩個應用程式之間，或分散式系統中傳送訊息，進行非同步通訊。

JMS相關概念

提供者：實現JMS規範的訊息中介軟體伺服器
客戶端：傳送或接收訊息的應用程式
生產者/釋出者：建立併發送訊息的客戶端
消費者/訂閱者：接收並處理訊息的客戶端
訊息：應用程式之間傳遞的資料內容

訊息模式：在客戶端之間傳遞訊息的模式，JMS中定義了主題和佇列兩種模式

JMS訊息模式：佇列模式（Queue模式）

客戶端包括生產者和消費者
佇列中的訊息只能被一個消費者消費

消費者可以隨時消費佇列中的訊息

佇列模型示意圖

JMS訊息模式：主題模型

客戶端包括髮布者和訂閱者
主題中的訊息被所有訂閱者消費

消費者不能消費訂閱之前就傳送到主題中的訊息

主題模型示意圖

佇列模式的訊息演示

使用JMS介面規範連線ActiveMQ

建立生產者
建立消費者
建立釋出者

建立訂閱者

JMS編碼介面之間的關係

程式碼演示

1.編寫AppProducer類

package com.myimooc.jms.queue;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * App 生產者-佇列模式
 * @author ZhangCheng on 2017-07-22
 *
 */
public class AppProducer {
    /** 指定ActiveMQ服務的地址 */
    private static final String URL = "tcp://127.0.0.1:61616";
    /** 指定佇列的名稱 */
    private static final String QUEUE_NAME = "queue-test";
    
    public static void main(String[] args) throws JMSException {
        
        // 1.建立ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        
        // 2.建立Connection
        Connection connection = connectionFactory.createConnection();
        
        // 3.啟動連線
        connection.start();
        
        // 4.建立會話（第一個引數：是否在事務中處理）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        
        // 5. 建立一個目標
        Destination destination = session.createQueue(QUEUE_NAME);
        
        // 6.建立一個生產者
        MessageProducer producer = session.createProducer(destination);
        
        for (int i = 0; i < 100; i++) {
            
            // 7.建立訊息
            TextMessage textMessage = session.createTextMessage("test" + i);
            
            // 8.釋出訊息
            producer.send(textMessage);
            
            System.out.println("訊息傳送：" + textMessage.getText());
        }
        
        // 9.關閉連線
        connection.close();
    }
    
}
 

2.編寫AppConsumer類

package com.myimooc.jms.queue;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * App 消費者-佇列模式
 * @author ZhangCheng on 2017-07-22
 *
 */
public class AppConsumer {
    /** 指定ActiveMQ服務的地址 */
    private static final String URL = "tcp://127.0.0.1:61616";
    /** 指定佇列的名稱 */
    private static final String QUEUE_NAME = "queue-test";
    
    public static void main(String[] args) throws JMSException {
        
        // 1.建立ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        
        // 2.建立Connection
        Connection connection = connectionFactory.createConnection();
        
        // 3.啟動連線
        connection.start();
        
        // 4.建立會話（第一個引數：是否在事務中處理）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        
        // 5.建立一個目標
        Destination destination = session.createQueue(QUEUE_NAME);
        
        // 6.建立一個消費者
        MessageConsumer consumer = session.createConsumer(destination);
        
        // 7.建立一個監聽器
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage)message;
                try {
                    System.out.println("接收訊息：" + textMessage.getText());
                } catch (JMSException e) {
                    System.out.println("接收訊息異常：");
                    e.printStackTrace();
                }
            }
        });
        
        // 8.關閉連線
        //connection.close();
    }
    
}
 

主題模式的訊息演示

程式碼演示

1.編寫AppProducer類

package com.myimooc.jms.topic;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * App 生產者-主題模式
 * @author ZhangCheng on 2017-07-22
 *
 */
public class AppProducer {
    /** 指定ActiveMQ服務的地址 */
    private static final String URL = "tcp://127.0.0.1:61616";
    /** 指定主題的名稱 */
    private static final String TOPIC_NAME = "topic-test";
    
    public static void main(String[] args) throws JMSException {
        
        // 1.建立ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        
        // 2.建立Connection
        Connection connection = connectionFactory.createConnection();
        
        // 3.啟動連線
        connection.start();
        
        // 4.建立會話（第一個引數：是否在事務中處理）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        
        // 5. 建立一個目標
        Destination destination = session.createTopic(TOPIC_NAME);
        
        // 6.建立一個生產者
        MessageProducer producer = session.createProducer(destination);
        
        for (int i = 0; i < 100; i++) {
            
            // 7.建立訊息
            TextMessage textMessage = session.createTextMessage("test" + i);
            
            // 8.釋出訊息
            producer.send(textMessage);
            
            System.out.println("訊息傳送：" + textMessage.getText());
        }
        
        // 9.關閉連線
        connection.close();
    }
    
}

2.編寫AppConsumer類

package com.myimooc.jms.topic;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * App 消費者-主題模式
 * @author ZhangCheng on 2017-07-22
 *
 */
public class AppConsumer {
    /** 指定ActiveMQ服務的地址 */
    private static final String URL = "tcp://127.0.0.1:61616";
    /** 指定主題的名稱 */
    private static final String TOPIC_NAME = "topic-test";
    
    public static void main(String[] args) throws JMSException {
        
        // 1.建立ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        
        // 2.建立Connection
        Connection connection = connectionFactory.createConnection();
        
        // 3.啟動連線
        connection.start();
        
        // 4.建立會話（第一個引數：是否在事務中處理）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        
        // 5.建立一個目標
        Destination destination = session.createTopic(TOPIC_NAME);
        
        // 6.建立一個消費者
        MessageConsumer consumer = session.createConsumer(destination);
        
        // 7.建立一個監聽器
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage)message;
                try {
                    System.out.println("接收訊息：" + textMessage.getText());
                } catch (JMSException e) {
                    System.out.println("接收訊息異常：");
                    e.printStackTrace();
                }
            }
        });
        
        // 8.關閉連線
        //connection.close();
    }
    
}