深入掌握JMS--转
深入掌握JMS（一）：JSM基礎
1. JMS基本概念
???? JMS(Java Message Service) 即Java消息服務。它提供標準的產生、發送、接收消息的接口簡化企業?應用的開發。它支持兩種消息通信模型：點到點（point-to-point）（P2P）模型和發布/訂閱（Pub/Sub）模型。P2P 模型規定了一個消息只能有一個接收者;Pub/Sub 模型允許一個消息可以有多個接收者。
??? 對于點到點模型，消息生產者產生一個消息后，把這個消息發送到一個Queue（隊列）中，然后消息接收者再從這個Queue中讀取，一旦這個消息被一個接收者讀取之后，它就在這個Queue中消失了，所以一個消息只能被一個接收者消費。
??? 與點到點模型不同，發布/訂閱模型中，消息生產者產生一個消息后，把這個消息發送到一個Topic中，這個Topic可以同時有多個接收者在監聽，當一個消息到達這個Topic之后，所有消息接收者都會收到這個消息。
?
?
| 簡單的講，點到點模型和發布/訂閱模型的區別就是前者是一對一，后者是一對多。 |
?
2. 幾個重要概念?
???Destination?：消息發送的目的地，也就是前面說的Queue和Topic。創建好一個消息之后，只需要把這個消息發送到目的地，消息的發送者就可以繼續做自己的事情，而不用等待消息被處理完成。至于這個消息什么時候，會被哪個消費者消費，完全取決于消息的接受者。
? ??Message?：從字面上就可以看出是被發送的消息。它有下面幾種類型：
??????? StreamMessage：Java 數據流消息，用標準流操作來順序的填充和讀取。
??????? MapMessage：一個Map類型的消息；名稱為 string 類型，而值為 Java 的基本類型。
??????? TextMessage：普通字符串消息，包含一個String。
??????? ObjectMessage：對象消息，包含一個可序列化的Java 對象
??????? BytesMessage：二進制數組消息，包含一個byte[]。
??????? XMLMessage:? 一個XML類型的消息。
??? 最常用的是TextMessage和ObjectMessage。
???Session：?與JMS提供者所建立的會話，通過Session我們才可以創建一個Message。
???Connection：?與JMS提供者建立的一個連接。可以從這個連接創建一個會話，即Session。
???ConnectionFactory:?那如何創建一個Connection呢？這就需要下面講到的ConnectionFactory了。通過這個工廠類就可以得到一個與JMS提供者的連接，即Conection。
???Producer：?消息的生產者，要發送一個消息，必須通過這個生產者來發送。
???MessageConsumer：?與生產者相對應，這是消息的消費者或接收者，通過它來接收一個消息。
???前面多次提到JMS提供者，因為JMS給我們提供的只是一系列接口，當我們使用一個JMS的時候，還是需要一個第三方的提供者，它的作用就是真正管理?這些Connection，Session，Topic和Queue等。
??? 通過下面這個簡圖可以看出上面這些概念的關系。
? ConnectionFactory---->Connection--->Session--->Message
? Destination + Session------------------------------------>Producer
? Destination + Session------------------------------------>MessageConsumer?
?
??? 那么可能有人會問: ConnectionFactory和Destination 從哪兒得到?
??? 這就和JMS提供者有關了. 如果在一個JavaEE環境中, 可以通過JNDI查找得到, 如果在一個非JavaEE環境中, 那只能通過JMS提供者提供給我們的接口得到了.
深入掌握JMS（二）：一個JMS例子
??????? 前一講簡單的介紹了一下JMS的基本概念, 這一講結合一個例子讓大家深入理解前一講的基本概念. 首先需要做的是選擇一個JMS提供者, 如果在JavaEE環境中可以不用考慮這些. 我們選擇ActiveMQ, 官方地址: http://activemq.apache.org/. 網上有很多介紹ActiveMQ的文檔, 所以在這里就不介紹了.
按照上一講的這個簡圖,
? ConnectionFactory---->Connection--->Session--->Message
? Destination + Session------------------------------------>Producer
? Destination + Session------------------------------------>MessageConsumer
首先需要得到ConnectionFactoy和Destination，這里創建一個一對一的Queue作為Destination。
? ConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
? Queue queue = new ActiveMQQueue("testQueue");
然后又ConnectionFactory創建一個Connection, 再啟動這個Connection:
? Connection connection = factory.createConnection();
? connection.start();
接下來需要由Connection創建一個Session:
? Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)
??? 現在暫且不用管參數的含義, 以后會詳細講到.
下面就可以創建Message了,這里創建一個TextMessage。
? Message message = session.createTextMessage("Hello JMS!");
要想把剛才創建的消息發送出去，需要由Session和Destination創建一個消息生產者：
? MessageProducer producer = session.createProducer(queue);
下面就可以發送剛才創建的消息了：
? producer.send(message);
消息發送完成之后，我們需要創建一個消息消費者來接收這個消息：
? MessageConsumer comsumer = session.createConsumer(queue);
? Message recvMessage = comsumer.receive();
消息消費者接收到這個消息之后，就可以得到它的內容：
? System.out.println(((TextMessage)recvMessage).getText());
至此，一個簡單的JMS例子就完成了。下面是全部源碼 ：
import javax.jms.Connection; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; public class MessageSendAndReceive {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");Connection connection = factory.createConnection();connection.start();Queue queue = new ActiveMQQueue("testQueue");final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Message message = session.createTextMessage("Hello JMS!");MessageProducer producer = session.createProducer(queue);producer.send(message);System.out.println("Send Message Completed!");MessageConsumer comsumer = session.createConsumer(queue);Message recvMessage = comsumer.receive();System.out.println(((TextMessage)recvMessage).getText());} }?深入掌握JMS（三）：MessageListener
?
消息的消費者接收消息可以采用兩種方式：
? 1、consumer.receive() 或 consumer.receive(int timeout)；
? 2、注冊一個MessageListener。
采用第一種方式，消息的接收者會一直等待下去，直到有消息到達，或者超時。后一種方式會注冊一個監聽器，當有消息到達的時候，會回調它的onMessage()方法。下面舉例說明：
MessageConsumer comsumer = session.createConsumer(queue);comsumer.setMessageListener(new MessageListener(){@Overridepublic void onMessage(Message m) {TextMessage textMsg = (TextMessage) m;try {System.out.println(textMsg.getText());} catch (JMSException e) {e.printStackTrace();}}});?
深入掌握JMS（四）：實戰Queue
? Queue實現的是點到點模型，在下面的例子中，啟動2個消費者共同監聽一個Queue，然后循環給這個Queue中發送多個消息，我們依然采用ActiveMQ。
import javax.jms.Connection; import javax.jms.DeliveryMode; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue;public class QueueTest {public static void main(String[] args) throws Exception {ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");Connection connection = factory.createConnection();connection.start();//創建一個QueueQueue queue = new ActiveMQQueue("testQueue");//創建一個SessionSession session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//注冊消費者1MessageConsumer comsumer1 = session.createConsumer(queue);comsumer1.setMessageListener(new MessageListener(){public void onMessage(Message m) {try {System.out.println("Consumer1 get " + ((TextMessage)m).getText());} catch (JMSException e) {e.printStackTrace();}}});//注冊消費者2MessageConsumer comsumer2 = session.createConsumer(queue);comsumer2.setMessageListener(new MessageListener(){public void onMessage(Message m) {try {System.out.println("Consumer2 get " + ((TextMessage)m).getText());} catch (JMSException e) {e.printStackTrace();}}});//創建一個生產者，然后發送多個消息。MessageProducer producer = session.createProducer(queue);for(int i=0; i<10; i++){producer.send(session.createTextMessage("Message:" + i));}} }運行這個例子會得到下面的輸出結果：
?
[java]?view plaincopy
? 可以看出每個消息直被消費了一次，但是如果有多個消費者同時監聽一個Queue的話，無法確定一個消息最終會被哪一個消費者消費。
?
深入掌握JMS（五）：實戰Topic
?
??? 與Queue不同的是，Topic實現的是發布/訂閱模型，在下面的例子中，啟動2個消費者共同監聽一個Topic，然后循環給這個Topic中發送多個消息。
import javax.jms.Connection; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQTopic;public class TopicTest {public static void main(String[] args) throws Exception {ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");Connection connection = factory.createConnection();connection.start();//創建一個TopicTopic topic= new ActiveMQTopic("testTopic");Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//注冊消費者1MessageConsumer comsumer1 = session.createConsumer(topic);comsumer1.setMessageListener(new MessageListener(){public void onMessage(Message m) {try {System.out.println("Consumer1 get " + ((TextMessage)m).getText());} catch (JMSException e) {e.printStackTrace();}}});//注冊消費者2MessageConsumer comsumer2 = session.createConsumer(topic);comsumer2.setMessageListener(new MessageListener(){public void onMessage(Message m) {try {System.out.println("Consumer2 get " + ((TextMessage)m).getText());} catch (JMSException e) {e.printStackTrace();}}});//創建一個生產者，然后發送多個消息。MessageProducer producer = session.createProducer(topic);for(int i=0; i<10; i++){producer.send(session.createTextMessage("Message:" + i));}} }運行后得到下面的輸出結果：
[java]?view plaincopy
說明每一個消息都會被所有的消費者消費。
?
深入掌握JMS（六）：消息頭
??? 一個消息對象分為三部分：消息頭(Headers)，屬性（Properties）和消息體（Payload）。對于StreamMessage和MapMessage，消息本身就有特定的結構，而對于TextMessage，ObjectMessage和BytesMessage是無結構的。一個消息可以包含一些重要的數據或者僅僅是一個事件的通知。
??? 消息的Headers部分通常包含一些消息的描述信息，它們都是標準的描述信息。包含下面一些值：
JMSDestination?
?????? 消息的目的地，Topic或者是Queue。
JMSDeliveryMode?
??????? 消息的發送模式：persistent或nonpersistent。前者表示消息在被消費之前，如果JMS提供者DOWN了，重新啟動后消息仍然存在。后者在這種情況下表示消息會被丟失。可以通過下面的方式設置：
?????? Producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
???????JMSTimestamp?
?????? 當調用send()方法的時候，JMSTimestamp會被自動設置為當前事件。可以通過下面方式得到這個值：
?????? long timestamp = message.getJMSTimestamp();
JMSExpiration?
?????? 表示一個消息的有效期。只有在這個有效期內，消息消費者才可以消費這個消息。默認值為0，表示消息永不過期。可以通過下面的方式設置：
?????? producer.setTimeToLive(3600000); //有效期1小時 （1000毫秒 * 60秒 * 60分）
JMSPriority?
?????? 消息的優先級。0-4為正常的優先級，5-9為高優先級。可以通過下面方式設置：
?????? producer.setPriority(9);
JMSMessageID?
?????? 一個字符串用來唯一標示一個消息。
JMSReplyTo?
?????? 有時消息生產者希望消費者回復一個消息，JMSReplyTo為一個Destination，表示需要回復的目的地。當然消費者可以不理會它。
JMSCorrelationID?
?????? 通常用來關聯多個Message。例如需要回復一個消息，可以把JMSCorrelationID設置為所收到的消息的JMSMessageID。
JMSType?
?????? 表示消息體的結構，和JMS提供者有關。
JMSRedelivered?
?????? 如果這個值為true，表示消息是被重新發送了。因為有時消費者沒有確認他已經收到消息或者JMS提供者不確定消費者是否已經收到。
??? 除了Header，消息發送者可以添加一些屬性(Properties)。這些屬性可以是應用自定義的屬性，JMS定義的屬性和JMS提供者定義的屬性。我們通常只適用自定義的屬性。
??? 后面會講到這些Header和屬性的用法。
深入掌握JMS（七）：DeliveryMode例子
?
在下面的例子中，分別發送一個Persistent和nonpersistent的消息，然后關閉退出JMS。
import javax.jms.Connection; import javax.jms.DeliveryMode; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; public class DeliveryModeSendTest {public static void main(String[] args) throws Exception {ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");Connection connection = factory.createConnection();connection.start();Queue queue = new ActiveMQQueue("testQueue");Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);MessageProducer producer = session.createProducer(queue);producer.setDeliveryMode(DeliveryMode.PERSISTENT);producer.send(session.createTextMessage("A persistent Message"));producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);producer.send(session.createTextMessage("A non persistent Message"));System.out.println("Send messages sucessfully!");} }?運行上面的程序，當輸出“Send messages sucessfully!”時，說明兩個消息都已經發送成功，然后我們結束它，來停止JMS Provider。
??? 接下來我們重新啟動JMS Provicer，然后添加一個消費者：
運行上面的程序，可以得到下面的輸出結果：
Consumer get A persistent Message
可以看出消息消費者只接收到一個消息，它是一個Persistent的消息。而剛才發送的non persistent消息已經丟失了。
另外, 如果發送一個non persistent消息, 而剛好這個時候沒有消費者在監聽, 這個消息也會丟失.
?
?
?
深入掌握JMS（八）：JMSReplyTo
?
?
??? 在下面的例子中，首先創建兩個Queue，發送者給一個Queue發送，接收者接收到消息之后給另一個Queue回復一個Message，然后再創建一個消費者來接受所回復的消息。
import javax.jms.Connection; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; public class MessageSendReceiveAndReply {public static void main(String[] args) throws Exception {ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");Connection connection = factory.createConnection();connection.start();//消息發送到這個QueueQueue queue = new ActiveMQQueue("testQueue");//消息回復到這個QueueQueue replyQueue = new ActiveMQQueue("replyQueue");final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//創建一個消息，并設置它的JMSReplyTo為replyQueue。Message message = session.createTextMessage("Andy");message.setJMSReplyTo(replyQueue);MessageProducer producer = session.createProducer(queue);producer.send(message);//消息的接收者MessageConsumer comsumer = session.createConsumer(queue);comsumer.setMessageListener(new MessageListener(){public void onMessage(Message m) {try {//創建一個新的MessageProducer來發送一個回復消息。MessageProducer producer = session.createProducer(m.getJMSReplyTo());producer.send(session.createTextMessage("Hello " + ((TextMessage) m).getText()));} catch (JMSException e1) {e1.printStackTrace();}}});//這個接收者用來接收回復的消息MessageConsumer comsumer2 = session.createConsumer(replyQueue);comsumer2.setMessageListener(new MessageListener(){public void onMessage(Message m) {try {System.out.println(((TextMessage) m).getText());} catch (JMSException e) {e.printStackTrace();}}});} }首先消息生產者發送一個消息，內容為“Andy”， 然后消費者收到這個消息之后根據消息的JMSReplyTo，回復一個消息，內容為“Hello Andy‘。 最后在回復的Queue上創建一個接收回復消息的消費者，它輸出所回復的內容。
??? 運行上面的程序，可以得到下面的輸出結果：
?
??
原創的地址找不到，只找到轉載的地址：http://blog.csdn.net/zhangxs_3/article/details/4034713
?
?
轉載于:https://www.cnblogs.com/davidwang456/p/3821906.html
總結
以上是生活随笔為你收集整理的深入掌握JMS--转的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Realm Configuration
- 下一篇： JTA 深度历险 - 原理与实现---转