JMS即java Message Service（Java消息服務(wù)）應(yīng)用程序接口，主要用于兩個應(yīng)用程序或分布式系統(tǒng)中的異步通信，JMS就像JDBC一樣是一個Java平臺的技術(shù)規(guī)范，JDBC就是用來訪問不同關(guān)系數(shù)據(jù)庫的API，而JMS就是用來進(jìn)行消息收發(fā)的API。
1、JMS提供者：實現(xiàn)JMS規(guī)范的產(chǎn)品，下面講的ActiveMQ就是其中一個，當(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ù)對象，消息類型可以為多種類型。
6、JMS隊列：由于消息被生產(chǎn)者發(fā)送之后不會被消費(fèi)者立即消費(fèi)掉，所以需要先將消息放在JMS隊列里等待消費(fèi)者收取，消息被消費(fèi)者消費(fèi)之后就會從隊列中移除。
7、JMS主題：支持發(fā)送消息給多個訂閱者的機(jī)制，簡單說就是生產(chǎn)者將消息發(fā)送到主題之后，主題可以將每個消息發(fā)給每個消費(fèi)者。
1、連接工廠（ConnectionFactory）：用來創(chuàng)建JMS連接。
2、JMS連接（Connection）：JMS客戶端與消息服務(wù)器之間的連接，即JMS客戶端和消息服務(wù)器之間通信的高速公路。
3、JMS會話（session）：即JMS客戶端和消息服務(wù)器之間的一個線程，有了高速公路還需要有一輛卡車將貨物運(yùn)送過去啊。需要注意在實際的應(yīng)用程序中會話可以支持事務(wù)，當(dāng)選擇支持事務(wù)的情況下，消息會先存放在會話中，當(dāng)事務(wù)提交時會將消息一并發(fā)送，提交之前可以進(jìn)行回滾操作。
4、JMS目的（Destination）：消息發(fā)布和接收的地點，可以是隊列或主題，至于是隊列還是主題要看消息通信的模型了（點對點模型與發(fā)布訂閱模型），下邊說明。
5、JMS生產(chǎn)者和消費(fèi)者（MessagePRoducter，MessageConsumer）：消息發(fā)送和接收的客戶端對象。
1、點對點：
一個生產(chǎn)者向特定隊列發(fā)送消息，一個消費(fèi)者從該隊列收取消息。該模式下生產(chǎn)者無需在消費(fèi)者消費(fèi)消息的情況下處于運(yùn)行狀態(tài)，消費(fèi)者也無需在生產(chǎn)者發(fā)送消息的情況下處于運(yùn)行狀態(tài)。
2、發(fā)布訂閱：
一個生產(chǎn)者向特定主題發(fā)送消息，0個或多個消費(fèi)者從該主題獲取消息，多個消費(fèi)者之間的消費(fèi)互不影響。該模式下發(fā)布和訂閱者之間存在時間依賴性，訂閱者需要先訂閱對應(yīng)主題的消息，否則將無法獲取消息，也就是說你今天訂閱人民日報的報紙，那么之前的報紙消息你是無法獲取到的。
上邊說了ActiveMQ是JMS規(guī)范的一個實現(xiàn)，那么也自然符合上邊的相關(guān)體系和模型。下邊通過代碼看ActiveMQ的實現(xià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; // 會話 接受或者發(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表示開啟事務(wù) 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、消費(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; // 會話 接受或者發(fā)送消息的線程 Destination destination; // 消息的目的地 MessageConsumer messageConsumer; // 消息的消費(fèi)者 // 實例化連接工廠 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表示不開啟事務(wù) destination=session.createQueue("Queue1"); // 創(chuàng)建連接的消息隊列 messageConsumer=session.createConsumer(destination); // 創(chuàng)建消息消費(fèi)者 //定時從隊列中收取消息 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)者常用用法：
在上邊的實現(xiàn)中消費(fèi)者不斷的從消息隊列中收取消息，這樣太耗費(fèi)資源，通過我們會給消費(fèi)者注冊一個監(jiān)聽器，當(dāng)監(jiān)聽到有新的消息時就通知消費(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; // 會話 接受或者發(fā)送消息的線程 Destination destination; // 消息的目的地 MessageConsumer messageConsumer; // 消息的消費(fèi)者 // 實例化連接工廠 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)建消息消費(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; // 會話 接受或者發(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、消息訂閱者一
/** * 消息消費(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; // 會話 接受或者發(fā)送消息的線程 Destination destination; // 消息的目的地 MessageConsumer messageConsumer; // 消息的消費(fèi)者 // 實例化連接工廠 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)建消息消費(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; // 會話 接受或者發(fā)送消息的線程 Destination destination; // 消息的目的地 MessageConsumer messageConsumer; // 消息的消費(fèi)者 // 實例化連接工廠 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)建消息消費(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(); } }}
新聞熱點
疑難解答
