activeMQ,JMS學(xué)習(xí)資料.doc

1 JMSJMS源于企業(yè)應(yīng)用對(duì)于消息中間件的需求，使應(yīng)用程序可以通過(guò)消息進(jìn)行異步處理而互不影響。Sun公司和它的合作伙伴設(shè)計(jì)的JMS API定義了一組公共的應(yīng)用程序接口和相應(yīng)語(yǔ)法，使得Java程序能夠和其他消息組件進(jìn)行通信。1.1 JMS的基本構(gòu)件1.1.1 連接工廠連接工廠是客戶用來(lái)創(chuàng)建連接的對(duì)象，例如ActiveMQ提供的ActiveMQConnectionFactory。1.1.2 連接JMS Connection封裝了客戶與JMS提供者之間的一個(gè)虛擬的連接。1.1.3 會(huì)話JMS Session是生產(chǎn)和消費(fèi)消息的一個(gè)單線程上下文。會(huì)話用于創(chuàng)建消息生產(chǎn)者（producer）、消息消費(fèi)者（consumer）和消息（message）等。會(huì)話提供了一個(gè)事務(wù)性的上下文，在這個(gè)上下文中，一組發(fā)送和接收被組合到了一個(gè)原子操作中。1.1.4 目的地目的地是客戶用來(lái)指定它生產(chǎn)的消息的目標(biāo)和它消費(fèi)的消息的來(lái)源的對(duì)象。JMS1.0.2規(guī)范中定義了兩種消息傳遞域：點(diǎn)對(duì)點(diǎn)（PTP）消息傳遞域和發(fā)布/訂閱消息傳遞域。點(diǎn)對(duì)點(diǎn)消息傳遞域的特點(diǎn)如下：l 每個(gè)消息只能有一個(gè)消費(fèi)者。l 消息的生產(chǎn)者和消費(fèi)者之間沒(méi)有時(shí)間上的相關(guān)性。無(wú)論消費(fèi)者在生產(chǎn)者發(fā)送消息的時(shí)候是否處于運(yùn)行狀態(tài)，它都可以提取消息。發(fā)布/訂閱消息傳遞域的特點(diǎn)如下：l 每個(gè)消息可以有多個(gè)消費(fèi)者。l 生產(chǎn)者和消費(fèi)者之間有時(shí)間上的相關(guān)性。訂閱一個(gè)主題的消費(fèi)者只能消費(fèi)自它訂閱之后發(fā)布的消息。JMS規(guī)范允許客戶創(chuàng)建持久訂閱，這在一定程度上放松了時(shí)間上的相關(guān)性要求。持久訂閱允許消費(fèi)者消費(fèi)它在未處于激活狀態(tài)時(shí)發(fā)送的消息。在點(diǎn)對(duì)點(diǎn)消息傳遞域中，目的地被成為隊(duì)列（queue）；在發(fā)布/訂閱消息傳遞域中，目的地被成為主題（topic）。1.1.5 消息生產(chǎn)者消息生產(chǎn)者是由會(huì)話創(chuàng)建的一個(gè)對(duì)象，用于把消息發(fā)送到一個(gè)目的地。1.1.6 消息消費(fèi)者消息消費(fèi)者是由會(huì)話創(chuàng)建的一個(gè)對(duì)象，它用于接收發(fā)送到目的地的消息。消息的消費(fèi)可以采用以下兩種方法之一：l 同步消費(fèi)。通過(guò)調(diào)用消費(fèi)者的receive方法從目的地中顯式提取消息。receive方法可以一直阻塞到消息到達(dá)。l 異步消費(fèi)?？蛻艨梢詾橄M(fèi)者注冊(cè)一個(gè)消息監(jiān)聽(tīng)器，以定義在消息到達(dá)時(shí)所采取的動(dòng)作。1.1.7 消息JMS消息由以下三部分組成：l 消息頭。每個(gè)消息頭字段都有相應(yīng)的getter和setter方法。l 消息屬性。如果需要除消息頭字段以外的值，那么可以使用消息屬性。l 消息體。JMS定義的消息類型有TextMessage、MapMessage、BytesMessage、StreamMessage和ObjectMessage。1.2 JMS的可靠性機(jī)制1.2.1 消息確認(rèn)JMS消息只有在被確認(rèn)之后，才認(rèn)為已經(jīng)被成功地消費(fèi)了。消息的成功消費(fèi)通常包含三個(gè)階段：客戶接收消息、客戶處理消息和消息被確認(rèn)。在事務(wù)性會(huì)話中，當(dāng)一個(gè)事務(wù)被提交的時(shí)候，確認(rèn)自動(dòng)發(fā)生。在非事務(wù)性會(huì)話中，消息何時(shí)被確認(rèn)取決于創(chuàng)建會(huì)話時(shí)的應(yīng)答模式（acknowledgement mode）。該參數(shù)有以下三個(gè)可選值：l Session.AUTO_ACKNOWLEDGE。當(dāng)客戶成功的從receive方法返回的時(shí)候，或者從MessageListener.onMessage方法成功返回的時(shí)候，會(huì)話自動(dòng)確認(rèn)客戶收到的消息。l Session.CLIENT_ACKNOWLEDGE?？蛻敉ㄟ^(guò)消息的acknowledge方法確認(rèn)消息。需要注意的是，在這種模式中，確認(rèn)是在會(huì)話層上進(jìn)行：確認(rèn)一個(gè)被消費(fèi)的消息將自動(dòng)確認(rèn)所有已被會(huì)話消費(fèi)的消息。例如，如果一個(gè)消息消費(fèi)者消費(fèi)了10個(gè)消息，然后確認(rèn)第5個(gè)消息，那么所有10個(gè)消息都被確認(rèn)。l Session.DUPS_ACKNOWLEDGE。該選擇只是會(huì)話遲鈍第確認(rèn)消息的提交。如果JMS provider失敗，那么可能會(huì)導(dǎo)致一些重復(fù)的消息。如果是重復(fù)的消息，那么JMS provider必須把消息頭的JMSRedelivered字段設(shè)置為true。1.2.2 持久性JMS 支持以下兩種消息提交模式：l PERSISTENT。指JMS provider持久保存消息，以保證消息不會(huì)因?yàn)镴MS provider的失敗而丟失。l NON_PERSISTENT。不要求JMS provider持久保存消息。1.2.3 優(yōu)先級(jí)可以使用消息優(yōu)先級(jí)來(lái)指示JMS provider首先提交緊急的消息。優(yōu)先級(jí)分10個(gè)級(jí)別，從0（最低）到9（最高）。如果不指定優(yōu)先級(jí)，默認(rèn)級(jí)別是4。需要注意的是，JMS provider并不一定保證按照優(yōu)先級(jí)的順序提交消息。1.2.4 消息過(guò)期可以設(shè)置消息在一定時(shí)間后過(guò)期，默認(rèn)是永不過(guò)期。1.2.5 本地事務(wù)在一個(gè)JMS客戶端，可以使用本地事務(wù)來(lái)組合消息的發(fā)送和接收。JMS Session接口提供了commit和rollback方法。事務(wù)提交意味著生產(chǎn)的所有消息被發(fā)送，消費(fèi)的所有消息被確認(rèn)；事務(wù)回滾意味著生產(chǎn)的所有消息被銷毀，消費(fèi)的所有消息被恢復(fù)并重新提交，除非它們已經(jīng)過(guò)期。事務(wù)性的會(huì)話總是牽涉到事務(wù)處理中，commit或rollback方法一旦被調(diào)用，一個(gè)事務(wù)就結(jié)束了，而另一個(gè)事務(wù)被開(kāi)始。關(guān)閉事務(wù)性會(huì)話將回滾其中的事務(wù)。需要注意的是，如果使用請(qǐng)求/回復(fù)機(jī)制，即發(fā)送一個(gè)消息，同時(shí)希望在同一個(gè)事務(wù)中等待接收該消息的回復(fù)，那么程序?qū)⒈粧炱穑驗(yàn)橹朗聞?wù)提交，發(fā)送操作才會(huì)真正執(zhí)行。需要注意的還有一個(gè)，消息的生產(chǎn)和消費(fèi)不能包含在同一個(gè)事務(wù)中。2 ActiveMQActiveMQ是apache旗下開(kāi)源消息中間件，是目前最流行的開(kāi)源的消息中間件。ActiveMQ功能特點(diǎn)：l 支持跨語(yǔ)言的客戶端，如：Java，C和C + +，C，Ruby，Perl，Python和PHP。l 在JMS客戶端和消息代理都全面支持企業(yè)集成模式。l 支持許多高級(jí)功能，如：信息組，虛擬目的地，通配符和復(fù)合目的地。l 完全支持JMS 1.1和J2EE1.4 。l 支持Spring，以便ActiveMQ可以很容易地嵌入到Spring應(yīng)用程序和使用Spring的XML配置機(jī)制 。l 通過(guò)了常見(jiàn)J2EE服務(wù)器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測(cè)試。 l 支持可插拔傳輸協(xié)議，如：in-VM，TCP, SSL, NIO, UDP, multicast, JGroups and JXTA 傳輸。l 支持通過(guò)JDBC和journal提供高速的消息持久化。l 設(shè)計(jì)基于高性能集群，客戶端服務(wù)器，對(duì)等通信。l Ajax支持網(wǎng)絡(luò)流媒體，支持使用純DHTML的Web瀏覽器，允許Web瀏覽器成為消息傳遞結(jié)構(gòu)的一部分。l 支持CXF和Axis，以便ActiveMQ可以很容易地進(jìn)入這些Web服務(wù)并提供可靠的消息傳遞。l 可作為一個(gè)在內(nèi)存中的JMS提供者，是JMS的單元測(cè)試的理想選擇。2.1 安裝ActiveMQ 可以從官方網(wǎng)站（/）下載最新版的ActiveMQ。（最新版的為：5.4.2）2.2 啟動(dòng)ActiveMQ在下載最新的ActiveMQ將其解壓到相應(yīng)的目錄就可以了。需要啟動(dòng)ActiveMQ只要找到$activemq.home/bin目錄，雙擊運(yùn)行activemq.bat就可以了(windows版本)。啟動(dòng)后的界面如上圖。（ActiveMQ5.3.0版本啟動(dòng)后的截圖，不同版本會(huì)有所不同）Listening for connections at: tcp:/74:61616（這里的IP顯示的是機(jī)器名稱）tcp:/74:61616表示監(jiān)聽(tīng)的端口地址，也就是編寫(xiě)程序時(shí)，獲取連接進(jìn)用到的URL。ActiveMQ Console at 74:8161/admin 74:8161/admin這個(gè)是新版本的activeMQ提供的管理工具訪問(wèn)地址，可以查看隊(duì)列詳情，生產(chǎn)者，消費(fèi)者等信息。2.3 配置ActiveMQ2.3.1 基本配置說(shuō)明ActiveMQ默認(rèn)使用的是XML格式配置，配置文件在$activemq.home/conf目錄下，文件名為activemq.xml file:$activemq.base/conf/perties !- The element is used to configure the ActiveMQ broker. - producerFlowControl=true memoryLimit=1mb producerFlowControl=true memoryLimit=1mb !- Use VM cursor for better latency For more information, see: /message-cursors.html - !- The systemUsage controls the maximum amount of space the broker will use before slowing down producers. For more information, see: /producer-flow-control.html - Kahadb配置說(shuō)明配置示例：KahaDB的各個(gè)可配置屬性：屬性默認(rèn)值描述directoryactivemq-data保存message store數(shù)據(jù)文件的目錄indexWriteBatchSize1000批量更新索引的閥值，當(dāng)要更新的索引到達(dá)這個(gè)索引時(shí)，批量更新到metadata store中indexCacheSize10000指定metadata cache的大小enableIndexWriteAsyncfalse寫(xiě)入索引文件到metadata store中的方式是否采用異步寫(xiě)入journalMaxFileLength32mb消息持久數(shù)據(jù)文件的大小enableJournalDiskSyncstrue如果為true，保證使用同步寫(xiě)入的方式持久化消息到j(luò)ournal文件中cleanupInterval30000清除（清除或歸檔）不再使用的journal 文件的時(shí)間周期（毫秒）。checkpointInterval5000寫(xiě)入索引信息到metadata store中的時(shí)間周期（毫秒）ignoreMissingJournalfilesfalse是否忽略丟失的journal文件。如果為false，當(dāng)丟失了journal文件時(shí)，broker啟動(dòng)時(shí)會(huì)拋異常并關(guān)閉checkForCorruptJournalFilesfalse如果為true，broker在啟動(dòng)的時(shí)候會(huì)檢測(cè)journal文件是否損壞，若損壞便嘗試恢復(fù)它。checksumJournalFilesfalse如果為true。KahaDB為journal文件生產(chǎn)一個(gè)checksum，以便能夠檢測(cè)journal文件是否損壞。archiveDataLogsfalse如果為true，當(dāng)達(dá)到cleanupInterval周期時(shí)，會(huì)歸檔journal文件而不是刪除directoryArchivenull指定歸檔journal文件存放的路徑databaseLockedWaitDelay10000在使用主從數(shù)據(jù)庫(kù)備份時(shí)，等待獲取DB上的lock的延遲時(shí)間。maxAsyncJobs10000等待寫(xiě)入journal文件的任務(wù)隊(duì)列的最大數(shù)量。應(yīng)該大于或等于最大并發(fā)producer的數(shù)量。配合并行存儲(chǔ)轉(zhuǎn)發(fā)屬性使用。concurrentStoreAndDispatchTransactionsfalse如果為true，轉(zhuǎn)發(fā)消息的時(shí)候同時(shí)提交事務(wù)concurrentStoreAndDispatchTopicsfalse如果為true，轉(zhuǎn)發(fā)Topic消息的時(shí)候同時(shí)存儲(chǔ)消息的message store中。concurrentStoreAndDispatchQueuestrue如果為true，轉(zhuǎn)發(fā)Queue消息的時(shí)候同時(shí)存儲(chǔ)消息到message store中。2.3.2 安全性配置說(shuō)明（以5.3.0的版本為例，各個(gè)版本配置會(huì)有所不同，需查閱相應(yīng)資料）訪問(wèn)activeMQ權(quán)限配置是在$activemq.home/conf目錄下的activemq.xml中配置的，默認(rèn)是沒(méi)有進(jìn)行配置，任何用戶都是可以訪問(wèn)activeMQ。如果需要對(duì)訪問(wèn)權(quán)限進(jìn)行控制可以進(jìn)行相應(yīng)配置，可以指定到相應(yīng)隊(duì)列的訪問(wèn)權(quán)限。1. 配置activemq.xml配置權(quán)限需要在，activemq.xml的元素中加入以下部分。 /queue=對(duì)應(yīng)的是所有隊(duì)列/ read=”admins” 表示admins組的所有用戶都可以消費(fèi)前面queue屬性指定的隊(duì)列的消息/ write=admins表示admins組的所有用戶都可以向前面queue屬性指定的隊(duì)列發(fā)送消息/ admin=admins表示admins組的所有用戶可以在前面queue屬性指定的隊(duì)列不存在的情況創(chuàng)建隊(duì)列 read=admins write=admins admin=admins / queue=TEST.表示隊(duì)列名稱所有以”TEST.”(包含“.”)開(kāi)頭的隊(duì)列,(這里建議隊(duì)列名稱以“.”作分隔符) read=users write=users admin=users / read=users write=users admin=users / write=guests,users admin=guests,users可以同時(shí)指定多個(gè)組 read=guests write=guests,users admin=guests,users /主題相關(guān)配置，與隊(duì)列配置類似 read=admins write=admins admin=admins / read=users write=users admin=users / read=guests write=guests,users admin=guests,users / read=guests,users write=guests,users admin=guests,users/ 2. 用戶配置在$activemq.home/conf目錄下增加 login.config、perties、perties三個(gè)文件。l login.config內(nèi)容activemq-domain org.apache.activemq.jaas.PropertiesLoginModule required debug=true perties.user=perties perties.group=perties; ;perties.user=perties指定用戶對(duì)應(yīng)的配置文件perties.group=perties;指定用戶組對(duì)應(yīng)的配置文件l perties內(nèi)容admins=systemusers=zengjun,llguests=guestusers=zengjun,llusers表示用戶組名，與activemq.xml安全性配置中的read,write,admin屬性值對(duì)應(yīng)。zengjun,ll表示兩個(gè)用戶名zengjun和ll，這里表示users組下有兩個(gè)用戶zengjun和ll。l perties內(nèi)容system=managerzengjun=zjll=llzengjun=zjzengjun表示用戶名，與perties配置用戶名對(duì)應(yīng)zj表示等號(hào)左邊用戶對(duì)應(yīng)的密碼。注：在配置安全性配置后，在寫(xiě)代碼創(chuàng)建連接時(shí)需要加上對(duì)應(yīng)的用戶名與密碼。如：Connection connection = connectionFactory.createConnection(zengjun,zj);2.4 點(diǎn)對(duì)點(diǎn)域2.4.1 生產(chǎn)消息生產(chǎn)消息都步驟如上圖的深色部分所示。首先需要從連接工廠中獲取到連接，然后通過(guò)連接來(lái)創(chuàng)建會(huì)話，再通過(guò)會(huì)話來(lái)創(chuàng)建目的地，再用會(huì)話與目的地來(lái)創(chuàng)建生產(chǎn)者。需要發(fā)送的消息也是通過(guò)會(huì)話來(lái)創(chuàng)建的。最后通過(guò)生產(chǎn)者來(lái)發(fā)送消息。示例代碼：/初使化連接工廠ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(tcp:/74:61616); Connection connection = null; Session session = null; MessageProducer producer = null;Destination destination = null;/創(chuàng)建連接connection = connectionFactory.createConnection();/創(chuàng)建會(huì)話 session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);/創(chuàng)目的地destination = session.createQueue(TEST_QUEUE_ZJ);/創(chuàng)生產(chǎn)者producer = session.createProducer(destination);/設(shè)置消息的持久模式producer.setDeliveryMode(DeliveryMode.PERSISTENT);connection.start();/創(chuàng)建消息TextMessage message = session.createTextMessage();/設(shè)置消息屬性 double d = Math.random();message.setStringProperty(ID, String.valueOf(d);message.setText(tttt111);message.setText(tttt22222222);/發(fā)送消息producer.send(message);2.4.2 消費(fèi)消息消費(fèi)消息如上圖深色部分所示。首先需要從連接工廠創(chuàng)建連接，然后再通過(guò)連接創(chuàng)建會(huì)話，然后通過(guò)會(huì)話創(chuàng)建目的地，再通過(guò)會(huì)話與目的地來(lái)創(chuàng)消費(fèi)者，然后消費(fèi)者調(diào)用接口來(lái)消費(fèi)消息。代碼示例：/初始連接工廠 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(tcp:/74:61616); Session session = null; MessageConsumer consumer = null; Connection connection = null; Message message = null;Destination destination=null;/創(chuàng)建連接connection = connectionFactory.createConnection();connection.start();/創(chuàng)建會(huì)話，指定消息的確認(rèn)模式，(確認(rèn)模式請(qǐng)參1.2.1)session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);/創(chuàng)建目的地destination = session.createQueue(TEST_QUEUE_ZJ);/創(chuàng)建消息者consumer = session.createConsumer(destination);/消息者調(diào)用接收消息接口來(lái)接收消息。/ 無(wú)時(shí)間參數(shù)表示一直等待，直到收到消息。/ message = consumer.receive();/ 有時(shí)間參數(shù)表示指定時(shí)間后沒(méi)有消息則結(jié)束時(shí),如果存在消息就在取完消息后結(jié)束message = consumer.receive(5 * 1000);/ 如果沒(méi)有收到消息，不會(huì)等待，立即往下執(zhí)行，不建議使用/ message = consumer.receiveNoWait();/判斷是否收到消息。if (message != null) /判斷消息的類型if (message instanceof TextMessage) TextMessage textMessage = (TextMessage) message;String text = textMessage.getText();System.out.println(TEXT: + text);/ 確認(rèn)消息textMessage.acknowledge(); else if (message instanceof StreamMessage) StreamMessage streamMessage = (StreamMessage) message;String strId = streamMessage.getStringProperty(ID);System.out.println(streammessage ID: + strId);/確認(rèn)消息streamMessage.acknowledge(); else System.out.println(沒(méi)有收到消息);還可以采用監(jiān)聽(tīng)的方式來(lái)消費(fèi)消息。采用監(jiān)聽(tīng)的方式，需要實(shí)現(xiàn)MessageListener接口，創(chuàng)建消費(fèi)者的方式與前面描消費(fèi)消息的步驟一致，在建創(chuàng)建好消費(fèi)者后需要設(shè)置實(shí)現(xiàn)MessageListener接口的監(jiān)聽(tīng)器，當(dāng)監(jiān)聽(tīng)器監(jiān)聽(tīng)到消費(fèi)者對(duì)應(yīng)的目的地上有消息時(shí)，會(huì)自動(dòng)調(diào)用onMessage方法，在onMessage方法中可以得到消息。代碼示例：/實(shí)現(xiàn)MessageListener接口public class AmqConsumerOnMessage implements MessageListener public static void main(String args) new AmqConsumerOnMessage().createConsumer(TEST_queuE);/實(shí)現(xiàn)onMessage方法 public void onMessage(Message message) try if (message != null) if (message instanceof TextMessage) TextMessage textMessage = (TextMessage) message; String text = textMessage.getText(); System.out.println(TEXT: + text); else if (message instanceof StreamMessage) StreamMessage streamMessage = (StreamMessage) message; String strId = streamMessage.getStringProperty(ID); System.out.println(streammessage ID: + strId); /確認(rèn)消息 message.acknowledge(); else System.out.println(沒(méi)有收到消息); catch (JMSException e) e.printStackTrace(); /創(chuàng)建消費(fèi)者 public void createConsumer(String queue) MessageConsumer consumer = null; Session session = null; Connection connection = null; try ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(tcp:/74:61616); connection = connectionFactory.createConnection(); connection.start(); session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); Destination objDestination = null; objDestination = session.createQueue(queue); consumer = session.createConsumer(objDestination);/設(shè)置監(jiān)聽(tīng)器，來(lái)監(jiān)聽(tīng)消息 consumer.setMessageListener(this); catch (JMSException e) e.printStackTrace(); finally ConnectionUtil.closeAll(connection, session, consumer); 2.5 發(fā)布/訂閱域2.5.1 發(fā)布消息發(fā)布消息與點(diǎn)對(duì)點(diǎn)域的生產(chǎn)消息類似。只是將隊(duì)列換成了主題，將生產(chǎn)者換成發(fā)布者代碼示例：/初始化連接工廠ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(tcp:/74:61616);TopicConnection connection = null;ActiveMQTopicSession session = null;ActiveMQTopicPublisher publisher = null;ActiveMQTopic topic = null; /創(chuàng)建連接connection = connectionFactory.createTopicConnection(zengjun, zj);/創(chuàng)建會(huì)話session = (ActiveMQTopicSession) connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);/創(chuàng)建主題topic = (ActiveMQTopic) session.createTopic(TEST.topic.zj);/創(chuàng)建發(fā)布者publisher = (ActiveMQTopicPublisher) session.createPublisher(topic);/設(shè)置消息持久方式，如果要實(shí)現(xiàn)持久訂閱，持久方法必須是DeliveryMode.PERSISTENTpublisher.setDeliveryMode(DeliveryMode.PERSISTENT);connection.start();TextMessage message = session.createTextMessage();mes