【正文】 nation, String messageSelector, boolean noLocal)。例如：MessageConsumer consumer = (destination)。MessageListener只有一個(gè)必須實(shí)現(xiàn)的方法 onMessage()，它只接收一個(gè)參數(shù)，即Message。例如：Message message = ()。然而，它也使得處理選擇標(biāo)準(zhǔn)的消息服務(wù)增加了一些額外開銷。 Java客戶端：例如：public final String SELECTOR = “JMSType = ‘TOPIC_PUBLISHER’”。Message由以下幾部分組成：消息頭，屬性和消息體。 下例演示接收消息并轉(zhuǎn)換為合適的消息類型： Message m = ()。結(jié)束事務(wù)有兩種方法：提交或者回滾。對(duì)于其他程序，低可靠性可以降低開銷和提高性能，例如發(fā)送消息時(shí)可以更改消息的優(yōu)先級(jí)或者指定消息的過期時(shí)間。另一方面，也可以通過生成和使用持久性消息并使用事務(wù)會(huì)話來獲得最佳可靠性。簽收可以由ActiveMQ發(fā)起，也可以由客戶端發(fā)起，取決于Session簽收模式的設(shè)置。 1．當(dāng)客戶端從receive或onMessage成功返回時(shí)，Session自動(dòng)簽收客戶端的這條消息的收條。 3．此選項(xiàng)指示Session不必確保對(duì)傳送消息的簽收。 對(duì)隊(duì)列來說，如果當(dāng)一個(gè)Session終止時(shí)它接收了消息但是沒有簽收，那么ActiveMQ將保留這些消息并將再次傳送給下一個(gè)進(jìn)入隊(duì)列的消費(fèi)者。如果這時(shí)有消息過期或者高優(yōu)先級(jí)的消息到來，那么這時(shí)消息的傳送將會(huì)和最初的有所不同。對(duì)于這些消息，可靠性是優(yōu)先考慮的因素。雖然這樣增加了消息傳送的開銷，但卻增加了可靠性。有兩種方法指定傳送模式： 1．使用 setDeliveryMode方法，這樣所有的消息都采用此傳送模式； 2．使用 send方法為每一條消息設(shè)置傳送模式； Java客戶端：傳送模式分別為： 1． 2． ActiveMQMessageProducer方法： void setDeliveryMode(int newDeliveryMode)。例如：()。然而，如果為這些消息分配了不同的優(yōu)先級(jí)，消息傳送系統(tǒng)將首先嘗試傳送優(yōu)先級(jí)較高的消息。其中deliveryMode為傳送模式，priority為消息優(yōu)先級(jí)，timeToLive為消息過期時(shí)間。JMS不要求嚴(yán)格按照這十個(gè)優(yōu)先級(jí)發(fā)送消息但必須保證加急消息要先于普通消息到達(dá)?；蛘?void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive)。 消息過期時(shí)間，send方法中的timeToLive 值加上發(fā)送時(shí)刻的GMT時(shí)間值。只有創(chuàng)建臨時(shí)目標(biāo)的Connection所創(chuàng)建的客戶端才可以從臨時(shí)目標(biāo)中接收消息，但是任何的生產(chǎn)者都可以向臨時(shí)目標(biāo)中發(fā)送消息。 某些客戶端需要一個(gè)目標(biāo)來接收對(duì)發(fā)送至其他客戶端的消息的回復(fù)。 Java客戶端：如下所示代碼段，將創(chuàng)建臨時(shí)的Destination，并將它放置在TextMessage的JMSReplyTo屬性中：Destination tempQueue = ()。它還展示了客戶端可以既是消息的生產(chǎn)者，又可以是消息的消費(fèi)者。 (REPLYTO_TEXT)。持久訂閱時(shí)，客戶端向ActiveMQ注冊(cè)一個(gè)識(shí)別自己身份的ID，當(dāng)這個(gè)客戶端處于離線時(shí)，ActiveMQ會(huì)為這個(gè)ID保存所有發(fā)送到主題的消息，當(dāng)客戶端再次連接到ActiveMQ時(shí)，會(huì)根據(jù)自己的ID得到所有當(dāng)自己處于離線時(shí)發(fā)送到主題的消息。name標(biāo)識(shí)訂閱主題所對(duì)應(yīng)的訂閱名稱，持久訂閱時(shí)需要設(shè)置此參數(shù)?？蛻舳舜a通過忽略異常、重試操作或回滾整個(gè)事務(wù)來處理異常。也就是說，可以將單個(gè)會(huì)話的上下文中執(zhí)行的一個(gè)或多個(gè)生產(chǎn)者或消費(fèi)者操作組成一個(gè)本地事務(wù)。類似地，單個(gè)事務(wù)可以用來接收隊(duì)列中的消息并將消息發(fā)送到主題上，反過來也可以。其中transacted為使用事務(wù)標(biāo)識(shí)，acknowledgeMode為簽收模式。 } catch (JMSException ex) { ()。在默認(rèn)大多數(shù)情況下，AcitveMQ是以異步模式發(fā)送消息。高性能的程序一般都能容忍在故障情況下丟失少量數(shù)據(jù)。 在 Connection層面配置異步發(fā)送，此層面的設(shè)置將覆蓋ConnectionFactory層面的設(shè)置: ((ActiveMQConnection)connection).setUseAsyncSend(true)。但是對(duì)于處理消息慢的消費(fèi)者，需要以異步模式分派。 consumer = (queue)。但是如果建立了多個(gè)Session和MessageConsumer，那么同一時(shí)刻多個(gè)線程同時(shí)從一個(gè)隊(duì)列中接收消息時(shí)就并不能保證處理時(shí)有序。如果消費(fèi)者發(fā)生故障，那么ActiveMQ將自動(dòng)故障轉(zhuǎn)移并選擇另一個(gè)消費(fèi)者。 3．在Session中使用CLIENT_ACKNOWLEDGE簽收模式，并且調(diào)用了recover()方法。例如可以利用混合目標(biāo)在同一操作中用向12個(gè)隊(duì)列發(fā)送同一條消息或者在同一操作中向一個(gè)主題和一個(gè)隊(duì)列發(fā)送同一條消息。 如果在一個(gè)目標(biāo)中混合不同類別的目標(biāo)，可以通過使用“queue://”和“topic://”前綴來識(shí)別不同的目標(biāo)。=10)?？梢詾槊總€(gè)消費(fèi)者指定消息預(yù)取。 consumer = (queue)。屬性默認(rèn)值alwaysSessionAsynctrueclientIDnullcloseTimeout15000(毫秒)copyMessageOnSendtruedisableTimeStampsByDefaultfasledispatchAsyncfaslenestedMapAndListEnabledtrueobjectMessageSerializationDeferedfasleoptimizeAcknowledgefalseoptimizedMessageDispatchtrueuseAsyncSendfalseuseCompressionfalseuseRetroactiveConsumerfalse 優(yōu)化優(yōu)化部分請(qǐng)參閱：6. ActiveMQ配置 配置文件 ActiveMQ配置文件：$AcrtiveMQ/conf/ 配置 ActiveMQ服務(wù)IP和端口 transportConnectors transportConnector name=openwire uri=tcp://localhost:61616 discoveryUri=mu lticast://default/ transportConnector name=ssl uri=ssl://localhost:61617/ transportConnector name=stomp uri=stomp://localhost:61613/ /transportConnectors 在transportConnectors標(biāo)識(shí)中配置ActiveMQ服務(wù)IP和端口，其中name屬性指定協(xié)議的名稱，uri屬性指定協(xié)議所對(duì)應(yīng)的協(xié)議名，IP地址和端口號(hào)。 配置 JXM broker brokerName=emv219 useJmx=true xmlns= managementContext managementContext connectorPort=1099 jmxDomainName=/ /managementContext /broker 配置JXM步驟如下： 1．設(shè)置broker標(biāo)識(shí)的useJmx屬性為true； 2．取消對(duì)managementContext標(biāo)識(shí)的注釋（系統(tǒng)默認(rèn)注釋 managementContext標(biāo)識(shí)），監(jiān)控的默認(rèn)端口為1099。但是此模式會(huì)造成消息重復(fù)接收。 目前版本的 C++客戶端僅支持 stomp協(xié)議目前版本的 C++客戶端程序（ActiveMQ CPP Release）僅支持stomp協(xié)議，因此傳輸消息的速度應(yīng)該沒有使用openwire協(xié)議的Java客戶端快。這時(shí)分布式的消息傳輸就會(huì)出現(xiàn)問題，此問題目前還沒找到原因。 分布式部署問題 ActiveMQ版本：ActiveMQ 測(cè)試選用上述兩個(gè)未正式發(fā)布的版本，未選用正式發(fā)布的ActiveMQ Release版本是因?yàn)榇税姹綽ug較多。 隊(duì)列消息堆積過多后有可能阻塞程序，這就意味著當(dāng)消息堆積超過20M后，程序可能出現(xiàn)問題。 7. 目前存在問題 C++客戶端丟失消息問題 ActiveMQ版本：ActiveMQ C++客戶端版本：ActiveMQ CPP Release 測(cè)試中發(fā)現(xiàn)，當(dāng) C++客戶端異常退出時(shí)（即沒有正常調(diào)用close函數(shù)關(guān)閉連接），ActiveMQ并不能檢測(cè)到C++客戶端的連接已經(jīng)中斷，這時(shí)如果向隊(duì)列中發(fā)送消息，那么第一條消息就會(huì)丟失，這時(shí)ActiveMQ才能檢測(cè)到這個(gè)連接是中斷的。Java客戶端默認(rèn)使用openwire協(xié)議，所以ActiveMQ服務(wù)地址為tcp://localhost:61616；目前C++客戶端僅支持stomp協(xié)議，所以ActiveMQ服務(wù)地址為tcp://localhost:61613。例如：當(dāng)要設(shè)置異步發(fā)送時(shí)，可以通過在 Configuration URI中使用 jms.$PROPERTY來設(shè)置。如果有少量的消息并且每條消息的處理都要花費(fèi)很長(zhǎng)的時(shí)間，那么可以設(shè)置預(yù)取值為1，這樣同一時(shí)間，ActiveMQ只會(huì)為這個(gè)消費(fèi)者分派一條消息。 消息預(yù)取 ActiveMQ的目標(biāo)之一就是高性能的數(shù)據(jù)傳送，所以ActiveMQ使用“預(yù)取限制”來控制有多少消息能及時(shí)的傳送給任何地方的消費(fèi)者。 (queue, someMessage)。 Java客戶端：例如：Queue queue = new ActiveMueue(,)。屬性默認(rèn)值collisionAvoidanceFactormaximumRedeliveries6initialRedeliveryDelay1000LuseCollisionAvoidancefalseuseExponentialBackOfffalsebackOffMultiplier5 目標(biāo)特色 復(fù)合目標(biāo)，ActiveMQ支持混合目標(biāo)技術(shù)。 consumer = (queue)。ActiveMQ4支持獨(dú)占的消費(fèi)。 consumer = (queue)。 在 Connection層面配置同步分派，此層面的設(shè)置將覆蓋ConnectionFactory層面的設(shè)置：((ActiveMQConnection)connection).setDispatchAsync(false)。這樣的意義：可以以異步模式向處理消息慢的消費(fèi)者分配消息；以同步模式向處理消息快的消費(fèi)者分配消息。 Java客戶端：使用 Connection URI配置異步發(fā)送: cf = new ActiveMQConnectionFactory(tcp://locahost:61616?=true)。在這種情況下，send方法都是同步的，并且一直阻塞直到ActiveMQ發(fā)回確認(rèn)消息：消息已經(jīng)存儲(chǔ)在持久性數(shù)據(jù)