一、JMS：

       概念：

       JMS即java Message Service（Java消息服務(wù)）應(yīng)用程序接口，主要用于兩個(gè)應(yīng)用程序或分布式系統(tǒng)中的異步通信，JMS就像JDBC一樣是一個(gè)Java平臺(tái)的技術(shù)規(guī)范，JDBC就是用來訪問不同關(guān)系數(shù)據(jù)庫的API，而JMS就是用來進(jìn)行消息收發(fā)的API。

       體系架構(gòu)：

       1、JMS提供者：實(shí)現(xiàn)JMS規(guī)范的產(chǎn)品，下面講的ActiveMQ就是其中一個(gè)，當(dāng)然除了ActiveMQ外還有很多其他的提供者。

       2、JMS客戶：生產(chǎn)或消費(fèi)消息的Java應(yīng)用程序或?qū)ο蟆?/p>

       3、JMS生產(chǎn)者：創(chuàng)建并發(fā)送消息的JMS客戶。

       4、JMS消費(fèi)者：接收消息的JMS客戶。

       5、JMS消息：JMS客戶之間傳遞的數(shù)據(jù)對(duì)象，消息類型可以為多種類型。

       6、JMS隊(duì)列：由于消息被生產(chǎn)者發(fā)送之后不會(huì)被消費(fèi)者立即消費(fèi)掉，所以需要先將消息放在JMS隊(duì)列里等待消費(fèi)者收取，消息被消費(fèi)者消費(fèi)之后就會(huì)從隊(duì)列中移除。

       7、JMS主題：支持發(fā)送消息給多個(gè)訂閱者的機(jī)制，簡單說就是生產(chǎn)者將消息發(fā)送到主題之后，主題可以將每個(gè)消息發(fā)給每個(gè)消費(fèi)者。

       對(duì)象模型：

       1、連接工廠（ConnectionFactory）：用來創(chuàng)建JMS連接。

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

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

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

       5、JMS生產(chǎn)者和消費(fèi)者（MessagePRoducter，MessageConsumer）：消息發(fā)送和接收的客戶端對(duì)象。

       JMS消息通信模型：

       1、點(diǎn)對(duì)點(diǎn)：

       一個(gè)生產(chǎn)者向特定隊(duì)列發(fā)送消息，一個(gè)消費(fèi)者從該隊(duì)列收取消息。該模式下生產(chǎn)者無需在消費(fèi)者消費(fèi)消息的情況下處于運(yùn)行狀態(tài)，消費(fèi)者也無需在生產(chǎn)者發(fā)送消息的情況下處于運(yùn)行狀態(tài)。

       2、發(fā)布訂閱：

       一個(gè)生產(chǎn)者向特定主題發(fā)送消息，0個(gè)或多個(gè)消費(fèi)者從該主題獲取消息，多個(gè)消費(fèi)者之間的消費(fèi)互不影響。該模式下發(fā)布和訂閱者之間存在時(shí)間依賴性，訂閱者需要先訂閱對(duì)應(yīng)主題的消息，否則將無法獲取消息，也就是說你今天訂閱人民日報(bào)的報(bào)紙，那么之前的報(bào)紙消息你是無法獲取到的。

二、ActiveMQ

       上邊說了ActiveMQ是JMS規(guī)范的一個(gè)實(shí)現(xiàn)，那么也自然符合上邊的相關(guān)體系和模型。下邊通過代碼看ActiveMQ的實(shí)現(xiàn)過程。

       點(diǎn)對(duì)點(diǎn)模型

       1、消息生產(chǎn)者：

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