JMS之ActiveMQ基本應用

2019-11-14 11:29:07

字體：大中小

來源：轉載

供稿：網友

一、JMS：

概念：

JMS即java Message Service（Java消息服務）應用程序接口，主要用于兩個應用程序或分布式系統中的異步通信，JMS就像JDBC一樣是一個Java平臺的技術規范，JDBC就是用來訪問不同關系數據庫的API，而JMS就是用來進行消息收發的API。

體系架構：

1、JMS提供者：實現JMS規范的產品，下面講的ActiveMQ就是其中一個，當然除了ActiveMQ外還有很多其他的提供者。

2、JMS客戶：生產或消費消息的Java應用程序或對象。

3、JMS生產者：創建并發送消息的JMS客戶。

4、JMS消費者：接收消息的JMS客戶。

5、JMS消息：JMS客戶之間傳遞的數據對象，消息類型可以為多種類型。

6、JMS隊列：由于消息被生產者發送之后不會被消費者立即消費掉，所以需要先將消息放在JMS隊列里等待消費者收取，消息被消費者消費之后就會從隊列中移除。

7、JMS主題：支持發送消息給多個訂閱者的機制，簡單說就是生產者將消息發送到主題之后，主題可以將每個消息發給每個消費者。

對象模型：

1、連接工廠（ConnectionFactory）：用來創建JMS連接。

2、JMS連接（Connection）：JMS客戶端與消息服務器之間的連接，即JMS客戶端和消息服務器之間通信的高速公路。

3、JMS會話（session）：即JMS客戶端和消息服務器之間的一個線程，有了高速公路還需要有一輛卡車將貨物運送過去啊。需要注意在實際的應用程序中會話可以支持事務，當選擇支持事務的情況下，消息會先存放在會話中，當事務提交時會將消息一并發送，提交之前可以進行回滾操作。

4、JMS目的（Destination）：消息發布和接收的地點，可以是隊列或主題，至于是隊列還是主題要看消息通信的模型了（點對點模型與發布訂閱模型），下邊說明。

5、JMS生產者和消費者（MessagePRoducter，MessageConsumer）：消息發送和接收的客戶端對象。

JMS消息通信模型：

1、點對點：

一個生產者向特定隊列發送消息，一個消費者從該隊列收取消息。該模式下生產者無需在消費者消費消息的情況下處于運行狀態，消費者也無需在生產者發送消息的情況下處于運行狀態。

2、發布訂閱：

一個生產者向特定主題發送消息，0個或多個消費者從該主題獲取消息，多個消費者之間的消費互不影響。該模式下發布和訂閱者之間存在時間依賴性，訂閱者需要先訂閱對應主題的消息，否則將無法獲取消息，也就是說你今天訂閱人民日報的報紙，那么之前的報紙消息你是無法獲取到的。

二、ActiveMQ

上邊說了ActiveMQ是JMS規范的一個實現，那么也自然符合上邊的相關體系和模型。下邊通過代碼看ActiveMQ的實現過程。

點對點模型

1、消息生產者：

/** * 消息生產者 */public class JMSProducer {	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默認的連接用戶名	private static final String PASSWord=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址	private static final int SENDNUM=10; // 發送的消息數量		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發送消息的線程		Destination destination; // 消息的目的地		MessageProducer messageProducer; // 消息生產者		// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);		try {			connection=connectionFactory.createConnection(); // 通過連接工廠獲取連接			connection.start(); // 啟動連接			session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 創建Session，參數為true表示開啟事務			destination=session.createQueue("Queue1"); // 創建消息隊列			messageProducer=session.createProducer(destination); // 創建消息生產者			sendMessage(session, messageProducer); // 發送消息			session.commit();		} catch (Exception e) {			e.printStackTrace();		} finally{			if(connection!=null){				try {					connection.close();				} catch (JMSException e) {					e.printStackTrace();				}			}		}	}		/**	 * 發送消息	 */	public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{		for(int i=0;i<JMSProducer.SENDNUM;i++){			TextMessage message=session.createTextMessage("ActiveMQ 發送的消息"+i);			messageProducer.send(message); //消息生產者發送消息		}	}}
       2、消費者基本用法：
 
/** * 消息消費者 */public class JMSConsumer {	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默認的連接用戶名	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發送消息的線程		Destination destination; // 消息的目的地		MessageConsumer messageConsumer; // 消息的消費者				// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);		try {			connection=connectionFactory.createConnection();  // 通過連接工廠獲取連接			connection.start(); // 啟動連接			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 創建Session，參數為false表示不開啟事務			destination=session.createQueue("Queue1");  // 創建連接的消息隊列			messageConsumer=session.createConsumer(destination); // 創建消息消費者			//定時從隊列中收取消息			while(true){				TextMessage textMessage=(TextMessage)messageConsumer.receive(100000);				if(textMessage!=null){					System.out.println("收到的消息為："+textMessage.getText());				}else{					break;				}			}		} catch (JMSException e) {			e.printStackTrace();		} 	}}
       3、消費者常用用法：
       在上邊的實現中消費者不斷的從消息隊列中收取消息，這樣太耗費資源，通過我們會給消費者注冊一個監聽器，當監聽到有新的消息時就通知消費者消費。
 
/** * 消息消費者 */public class JMSConsumer2 {	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默認的連接用戶名	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發送消息的線程		Destination destination; // 消息的目的地		MessageConsumer messageConsumer; // 消息的消費者				// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);			try {			connection=connectionFactory.createConnection();  // 通過連接工廠獲取連接			connection.start(); // 啟動連接			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 創建Session			destination=session.createQueue("Queue1");  // 創建連接的消息隊列			messageConsumer=session.createConsumer(destination); // 創建消息消費者			messageConsumer.setMessageListener(new Listener()); // 注冊消息監聽		} catch (JMSException e) {			e.printStackTrace();		} 	}}/** * 消息監聽 */public class Listener implements MessageListener{	@Override	public void onMessage(Message message) {		try {			System.out.println("收到的消息："+((TextMessage)message).getText());		} catch (JMSException e) {			e.printStackTrace();		}	}}
       訂閱模式
       1、消息發布者
/** * 消息生產者-消息發布者 */public class JMSProducer {	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默認的連接用戶名	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址	private static final int SENDNUM=10; // 發送的消息數量		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發送消息的線程		Destination destination; // 消息的目的地		MessageProducer messageProducer; // 消息生產者				// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);		try {			connection=connectionFactory.createConnection(); //通過連接工廠獲取連接			connection.start(); //啟動連接			session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); //創建Session			destination=session.createTopic("Topic1"); //這里創建的為消息主題			messageProducer=session.createProducer(destination); //創建消息生產者			sendMessage(session, messageProducer); //發送消息			session.commit();		} catch (Exception e) {			e.printStackTrace();		} finally{			if(connection!=null){				try {					connection.close();				} catch (JMSException e) {					e.printStackTrace();				}			}		}	}		/**	 * 發送消息	 */	public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{		for(int i=0;i<JMSProducer.SENDNUM;i++){			TextMessage message=session.createTextMessage("ActiveMQ 發送的消息"+i);			System.out.println("發送的消息："+"ActiveMQ 發布的消息"+i);			messageProducer.send(message); //發送消息		}	}}
       2、消息訂閱者一
/** * 消息消費者-消息訂閱者一 */public class JMSConsumer {	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默認的連接用戶名	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發送消息的線程		Destination destination; // 消息的目的地		MessageConsumer messageConsumer; // 消息的消費者				// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);		try {			connection=connectionFactory.createConnection();  // 通過連接工廠獲取連接			connection.start(); // 啟動連接			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 創建Session			destination=session.createTopic("Topic1"); //在這里創建的為消息主題			messageConsumer=session.createConsumer(destination); // 創建消息消費者			messageConsumer.setMessageListener(new Listener()); // 注冊消息監聽		} catch (JMSException e) {			e.printStackTrace();		} 	}}/** * 消息監聽-訂閱者一 */public class Listener implements MessageListener{	@Override	public void onMessage(Message message) {		try {			System.out.println("訂閱者一收到的消息："+((TextMessage)message).getText());		} catch (JMSException e) {			e.printStackTrace();		}	}}
       3、消息訂閱者二
/** * 消息消費者-消息訂閱者二 */public class JMSConsumer2 {	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默認的連接用戶名	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發送消息的線程		Destination destination; // 消息的目的地		MessageConsumer messageConsumer; // 消息的消費者				// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);						try {			connection=connectionFactory.createConnection();  // 通過連接工廠獲取連接			connection.start(); // 啟動連接			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 創建Session			destination=session.createTopic("Topic1");			messageConsumer=session.createConsumer(destination); // 創建消息消費者			messageConsumer.setMessageListener(new Listener2()); // 注冊消息監聽		} catch (JMSException e) {			e.printStackTrace();		} 	}}/** * 消息監聽-訂閱者二 */public class Listener2 implements MessageListener{	@Override	public void onMessage(Message message) {		try {			System.out.println("訂閱者二收到的消息："+((TextMessage)message).getText());		} catch (JMSException e) {			e.printStackTrace();		}	}}