ActiveMQ分享（一）JMS簡(jiǎn)介文章分類(lèi):
Java編程一、概述
Message，即消息。人與人之間通過(guò)消息傳遞信息。言語(yǔ)、眼神、肢體動(dòng)作都可被視為消息體。當(dāng)然還有我們經(jīng)常用到的郵件、短信。計(jì)算機(jī)系統(tǒng)也由消息來(lái)主導(dǎo)運(yùn)行。每一條指令的執(zhí)行，每一個(gè)數(shù)據(jù)包的傳遞。軟件系統(tǒng)間的合作也不例外，消息告訴各個(gè)系統(tǒng)應(yīng)該怎樣協(xié)作。事件處理機(jī)制，也是消息傳送的過(guò)程。消息無(wú)處不在。
消息分為同步消息和異步消息。同步消息在接收到對(duì)方的返回前，需要掛起，直到返回或超時(shí)。異步消息只需要發(fā)送消息，不需要對(duì)方系統(tǒng)的立即反饋。
同步消息如java RPC調(diào)用，同步調(diào)用依賴(lài)于被調(diào)用方，如果被調(diào)用方失敗或網(wǎng)絡(luò)錯(cuò)誤，那么程序就沒(méi)辦法繼續(xù)執(zhí)行下去，造成一個(gè)系統(tǒng)最薄弱的環(huán)節(jié)依賴(lài)于對(duì)方系統(tǒng)。而多個(gè)系統(tǒng)通過(guò)同步調(diào)用方式耦合在一起的時(shí)候，那么可靠性取決于最薄弱的一方系統(tǒng)。而異步調(diào)用能增強(qiáng)一個(gè)系統(tǒng)的健壯性。當(dāng)然，不是任何情況都適合異步調(diào)用，還是那句話(huà)，能異步的地方，盡量異步。
異步消息，如同一個(gè)郵箱系統(tǒng)，我們把信件丟入郵桶，郵遞員會(huì)更具上面的地址，送達(dá)到這封信要去的地方。郵箱和信件格式由郵局提供定義，比如郵箱需要有一個(gè)口子投遞郵件，郵件需要有地址，郵政編碼等等。而這些郵箱具體的加工和制作均交由各自的廠(chǎng)商來(lái)完成。
在java消息領(lǐng)域，我們也有一個(gè)稱(chēng)為消息中間件的東西，來(lái)提供這樣一個(gè)服務(wù)。消息的發(fā)送、消費(fèi)接口、消息體的格式等都由JMS來(lái)定義，而具體的實(shí)現(xiàn)由各個(gè)消息中間件廠(chǎng)商來(lái)實(shí)現(xiàn)。JMS是sun公司對(duì)于消息中間件的一個(gè)規(guī)范。對(duì)java領(lǐng)域里的消息起到舉足輕重的作用。以前的消息交互，均各自實(shí)現(xiàn)一套格式，如同一個(gè)國(guó)家的人都用不同的方言跟另外來(lái)自不同省份的人交流一樣。自從規(guī)范了普通話(huà)，我們的交流成本降低了。這也正如JMS規(guī)范在整個(gè)java消息領(lǐng)域的作用。
二、JMS簡(jiǎn)介
JMS1.1規(guī)范定義了一些概念和一組API,可以使得在我們利用消息系統(tǒng)的過(guò)程中，不依賴(lài)于各個(gè)廠(chǎng)商的具體實(shí)現(xiàn)，便能寫(xiě)出消息代碼。由于不依賴(lài)具體廠(chǎng)商實(shí)現(xiàn)，這樣的代碼有很好的移植性。
JMS1.1定義了的部分概念：
1、JMS客戶(hù)端：接收或發(fā)送消息的java系統(tǒng)
2、JMS消息體：系統(tǒng)間發(fā)送的消息體
3、JMS提供商：JMS規(guī)范實(shí)現(xiàn)廠(chǎng)商
4、JMS管理對(duì)象：預(yù)先配置好的用于JMS客戶(hù)端的JMS對(duì)象。如ConnectionFactory跟JMS服務(wù)端的連接工廠(chǎng)，Destination 接收和發(fā)送消息的目標(biāo)地址等
5、JMS Domain：JMS定義了兩種域模型，一種是PTP（point-to-point）即點(diǎn)對(duì)點(diǎn)消息傳輸模型，一種是pub/sub(publish-subscribe)即發(fā)布訂閱模型。
PTP通過(guò)一個(gè)先進(jìn)先出的queue實(shí)現(xiàn)。很多人在這個(gè)PTP概念上有所誤解，所謂點(diǎn)對(duì)點(diǎn)不是指生產(chǎn)者和消費(fèi)者只有一個(gè)。PTP如下圖所示：
我們可以看到，一個(gè)或多個(gè)生產(chǎn)者發(fā)送消息，消息m2先抵達(dá)了queue，然后m1也發(fā)出了，并一同存在于一個(gè)先進(jìn)先出的queue里面。消費(fèi)者也存在一個(gè)或多個(gè)，對(duì)queue里的消息進(jìn)行消費(fèi)。但消息被隨機(jī)的一個(gè)消費(fèi)者消費(fèi)且僅消費(fèi)一次。
在pub/sub消息模型中，消息被廣播給所有訂閱者。如下圖：
JMS1.1定義了如下接口：
JMS通用接口JMS PTP接口JMS pub/sub 接口描述
ConnectionFactoryQueueConnectionFactoryTopicConnectionFactory創(chuàng)建與JMS服務(wù)的連接
ConnectionQueueConnectionTopicConnectionJMS服務(wù)的連接，可創(chuàng)建Session
DestinationQueueTopic發(fā)送接收消息的目標(biāo)，Queue和Topic均可視為一種Destination
SessionQueueSessionTopicSessionSession
MessageProducerQueueSenderTopicPublisher消息發(fā)送者
MessageConsumerQueueReceiver、QueueBrowserTopicSubscriber消息接收者/瀏覽者/訂閱者
接口關(guān)系如下：
可以看到一個(gè)JMS應(yīng)用的發(fā)送端的標(biāo)準(zhǔn)流程是：創(chuàng)建連接工廠(chǎng)>創(chuàng)建連接>創(chuàng)建session>創(chuàng)建發(fā)送者>創(chuàng)建消息體>發(fā)送消息到Destination（queue或topic）。
接收端則為：創(chuàng)建連接工廠(chǎng)>創(chuàng)建連接>創(chuàng)建session>創(chuàng)建接收者>創(chuàng)建消息監(jiān)聽(tīng)器監(jiān)聽(tīng)某Destination的消息>獲取消息并執(zhí)行業(yè)務(wù)邏輯
下面展示了JMS的編碼模板：
發(fā)送端：
Java代碼
public class QueueSender {
public static void main(String[] args) throws JMSException{
ConnectionFactory factory=new ActiveMQConnectionFactory("tcp://localhost:61616");//這里為簡(jiǎn)便，依賴(lài)ActiveMQ的JMS實(shí)現(xiàn)。現(xiàn)實(shí)程序中可以通過(guò)jndi查找方式查找factory，這樣徹底避免關(guān)聯(lián)特定JMS實(shí)現(xiàn)者。并擁有最大的可移植性。
Connection connection=factory.createConnection();
Session session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Message message=session.createTextMessage("Hello World!"+new Date().getTime());
Queue queue=new ActiveMQQueue("queue.somebody");//同上，這里依賴(lài)了ActiveMQ的JMS實(shí)現(xiàn)，現(xiàn)實(shí)中可以采用jndi查找方式獲得Queue。
MessageProducer producer=session.createProducer(queue);
producer.send(message);
session.close();
connection.close();
}
}
接收端：
Java代碼
public class QueueReceiver {
public static void main(String[] args) throws JMSException{
ConnectionFactory factory=new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection=factory.createConnection();
Session session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Queue queue=new ActiveMQQueue("queue.somebody");
MessageConsumer receiver=session.createConsumer(queue);//創(chuàng)建兩個(gè)接收者，同時(shí)消費(fèi)同一個(gè)queue的消息。queue里的消息派發(fā)且僅派發(fā)一次給唯一一個(gè)消費(fèi)者。
MessageConsumer receiver2=session.createConsumer(queue);
receiver.setMessageListener(new QueueMessageListener("1"));
receiver2.setMessageListener(new QueueMessageListener("2"));
connection.start();
}
}
消息偵聽(tīng)者：
Java代碼
public class QueueMessageListener implements MessageListener{
private String num;
public QueueMessageListener(String num){
super();
this.num=num;
}
public QueueMessageListener(){
super();
}
public void onMessage(Message message) {
System.out.println(message.toString()+num);
}
我們安裝了ApacheMQ之后，啟動(dòng)本地MQ服務(wù)，便能測(cè)試這幾個(gè)類(lèi)的執(zhí)行方式，觀(guān)察執(zhí)行過(guò)程。我們可以發(fā)現(xiàn)消息僅被消費(fèi)了一次，JMS提供商必須保證一條消息不會(huì)重復(fù)被消費(fèi)，當(dāng)然更不能丟失。
在PTP下，我們建立一個(gè)queue以后，如果沒(méi)有消費(fèi)者連入該queue，那么消息生產(chǎn)者發(fā)送的消息將堵塞在queue里面，直到有消費(fèi)者來(lái)消費(fèi)。如果消費(fèi)者出現(xiàn)故障，不再連入queue，消息還會(huì)保存在queue里面，等待下次消費(fèi)者的連入，再進(jìn)行消費(fèi)。在pub/sub模型下，如果消費(fèi)者出現(xiàn)故障，那么所有的消息，將不在為之保留，下次連接之時(shí)，只能消費(fèi)此時(shí)生產(chǎn)者發(fā)送的消息。JMS于是定義了另一種訂閱者，叫Durable Subscription（持久訂閱者），該訂閱者在第一次連上topic后，就注冊(cè)了該topic的消息，消息會(huì)在消費(fèi)者不可用的情況下為之保留消息，并在其再次連上topic后，重推消息給消費(fèi)者。
理解了JMS中的兩種消息模型以后，我們來(lái)說(shuō)說(shuō)JMS中的消息體格式。JMS對(duì)消息體定為三個(gè)部分 1)head 消息頭信息 2)properties 消息屬性值 3)body 消息內(nèi)容。如下圖所示：
其含義大致解釋如下：
頭類(lèi)型描述設(shè)置方
JMSDestination描述該消息發(fā)往的目的地在發(fā)送方法中設(shè)定
JMSDeliveryModeNON_PERSISTENT 非持久化 表示消息發(fā)往JMS消息服務(wù)器之后，保存在內(nèi)存中，不做持久化；PERSISTENT 持久化 消息發(fā)往JMS消息服務(wù)器之后，持久化數(shù)據(jù)。以保證消息服務(wù)器拓機(jī)造成的消息丟失發(fā)送方法中設(shè)定
JMSExpiration消息過(guò)期時(shí)間。消費(fèi)者在發(fā)送消息時(shí)，可設(shè)定消息的time-to-live時(shí)間，如producer.setTimeToLive(10000)，在消息發(fā)送后，該時(shí)間保留在消息的JMSExpiration字段中，如果在指定的這段時(shí)間內(nèi)消息未被消費(fèi)，該消息將會(huì)被丟棄。在發(fā)送方法中設(shè)定
JMSPriority消息優(yōu)先級(jí)，JMS把消息分為10個(gè)等級(jí)，0-4為普通優(yōu)先級(jí)，5-9為加快優(yōu)先級(jí)，ActiveMQ中默認(rèn)的消息優(yōu)先級(jí)為4在發(fā)送方法中設(shè)定
JMSMessageId消息唯一性ID，必須以“ID:”為前綴在發(fā)送方法中設(shè)定
JMSTimestamp消息發(fā)送時(shí)間，表示消息的發(fā)送時(shí)間點(diǎn)，而非傳送時(shí)間在發(fā)送方法中設(shè)定
JMSCorrelationID客戶(hù)端
JMSReplyTo客戶(hù)端
JMSType客戶(hù)端
JMSRedeliveredJMS提供商
屬性值含義
properties
消息體含義
body消息體分為1:StreamMessage2:MapMessage 3:TextMessage 4:ObjectMessage 5:BytesMessage
再談消息可靠性
在上面，談及消息體格式定義中，有個(gè)字段 JMSDeliveryMode用來(lái)表示該消息發(fā)送后，JMS提供商應(yīng)該怎么處理消息。PERSISTENT（持久化）的消息在JMS服務(wù)器中持久化。接收端如果采用點(diǎn)對(duì)點(diǎn)的queue方式或者Durable Subscription（持久訂閱者）方式，那么消息可保證只且只有一次被成功接收。NON_PERSISTENT(非持久化)的消息在JMS服務(wù)器關(guān)閉或宕機(jī)時(shí)，消息丟失。根據(jù)發(fā)送端和接收端采用的方式，列出如下可靠性表格，以作參考。
引用
注意：以下的可靠性不包括JMS服務(wù)器由于資源關(guān)系，造成的消息不能持久化等因素引起的不可靠（該類(lèi)不可靠應(yīng)該是JMS提供商或硬件引起的的資源和處理能力的極限問(wèn)題，應(yīng)該由管理人員解決）。也不包括消息由于超時(shí)時(shí)間造成的銷(xiāo)毀丟失。
消息發(fā)送端消息接收端可靠性及因素
PERSISTENTqueue receiver/durable subscriber消費(fèi)一次且僅消費(fèi)一次?？煽啃宰詈?，但是占用服務(wù)器資源比較多。
PERSISTENTnon-durable subscriber最多消費(fèi)一次。這是由于non-durable subscriber決定的，如果消費(fèi)端宕機(jī)或其他問(wèn)題導(dǎo)致與JMS服務(wù)器斷開(kāi)連接，等下次再聯(lián)上JMS服務(wù)器時(shí)的一系列消息，不為之保留。
NON_PERSISTENTqueue receiver/durable subscriber最多消費(fèi)一次。這是由于服務(wù)器的宕機(jī)會(huì)造成消息丟失
NON_PERSISTENTnon-durable subscriber最多消費(fèi)一次。這是由于服務(wù)器的宕機(jī)造成消息丟失，也可能是由于non-durable subscriber的性質(zhì)所決定
消息的通知確認(rèn)
在客戶(hù)端接收了消息之后，JMS服務(wù)怎樣有效確認(rèn)消息是否已經(jīng)被客戶(hù)端接收呢？Session session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);這段代碼創(chuàng)建一個(gè)非事務(wù)性的session，并采用auto_acknowledge方式通知JMS服務(wù)器。如果采用事務(wù)性session時(shí)，通知會(huì)伴隨session的commit/rollback同時(shí)發(fā)送通知。在我們采用非事務(wù)session時(shí)，有三種通知方式。
通知方式效果
DUPS_OK_ACKNOWLEDGEsession延遲通知。如果JMS服務(wù)器宕機(jī)，會(huì)造成重復(fù)消息的情況。程序必須保證處理重復(fù)消息而不引起程序邏輯的混亂。
AUTO_ACKNOWLEDGE當(dāng)receive或MessageListener方法成功返回后自動(dòng)通知。
CLIENT_ACKNOWLEDGE客戶(hù)端調(diào)用消息的acknowledge方法通知
JMS就先簡(jiǎn)要的介紹到這里，詳細(xì)資料請(qǐng)參看JMS1.1規(guī)范。關(guān)于ActiveMQ的文章文章請(qǐng)繼續(xù)關(guān)注 ActiveMQ分享（二）
