一、JMS：

       概念：

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

       體系架構(gòu)：

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

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

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

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

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

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

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

       對象模型：

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

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

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

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

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

       JMS消息通信模型：

       1、點對點：

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

       2、發(fā)布訂閱：

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

二、ActiveMQ

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

       點對點模型

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

/** * 消息生產(chǎn)者 */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; // 發(fā)送的消息數(shù)量		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發(fā)送消息的線程		Destination destination; // 消息的目的地		MessageProducer messageProducer; // 消息生產(chǎn)者		// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);		try {			connection=connectionFactory.createConnection(); // 通過連接工廠獲取連接			connection.start(); // 啟動連接			session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 創(chuàng)建Session，參數(shù)為true表示開啟事務			destination=session.createQueue("Queue1"); // 創(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);			messageProducer.send(message); //消息生產(chǎn)者發(fā)送消息		}	}}
       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; // 會話 接受或者發(fā)送消息的線程		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); // 創(chuàng)建Session，參數(shù)為false表示不開啟事務			destination=session.createQueue("Queue1");  // 創(chuàng)建連接的消息隊列			messageConsumer=session.createConsumer(destination); // 創(chuàng)建消息消費者			//定時從隊列中收取消息			while(true){				TextMessage textMessage=(TextMessage)messageConsumer.receive(100000);				if(textMessage!=null){					System.out.println("收到的消息為："+textMessage.getText());				}else{					break;				}			}		} catch (JMSException e) {			e.printStackTrace();		} 	}}
       3、消費者常用用法：
       在上邊的實現(xiàn)中消費者不斷的從消息隊列中收取消息，這樣太耗費資源，通過我們會給消費者注冊一個監(jiān)聽器，當監(jiān)聽到有新的消息時就通知消費者消費。
 
/** * 消息消費者 */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; // 會話 接受或者發(fā)送消息的線程		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); // 創(chuàng)建Session			destination=session.createQueue("Queue1");  // 創(chuàng)建連接的消息隊列			messageConsumer=session.createConsumer(destination); // 創(chuàng)建消息消費者			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; // 默認的連接用戶名	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默認的連接密碼	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默認的連接地址	private static final int SENDNUM=10; // 發(fā)送的消息數(shù)量		public static void main(String[] args) {		ConnectionFactory connectionFactory; // 連接工廠		Connection connection = null; // 連接		Session session; // 會話 接受或者發(fā)送消息的線程		Destination destination; // 消息的目的地		MessageProducer messageProducer; // 消息生產(chǎn)者				// 實例化連接工廠		connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);		try {			connection=connectionFactory.createConnection(); //通過連接工廠獲取連接			connection.start(); //啟動連接			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、消息訂閱者一
/** * 消息消費者-消息訂閱者一 */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; // 會話 接受或者發(fā)送消息的線程		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); // 創(chuàng)建Session			destination=session.createTopic("Topic1"); //在這里創(chuàng)建的為消息主題			messageConsumer=session.createConsumer(destination); // 創(chuàng)建消息消費者			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、消息訂閱者二
/** * 消息消費者-消息訂閱者二 */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; // 會話 接受或者發(fā)送消息的線程		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); // 創(chuàng)建Session			destination=session.createTopic("Topic1");			messageConsumer=session.createConsumer(destination); // 創(chuàng)建消息消費者			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();		}	}}