【文章內(nèi)容簡(jiǎn)介】 ty為消息優(yōu)先級(jí)，timeToLive為消息過(guò)期時(shí)間。例如：(message)。 MessageConsumer MessageConsumer是一個(gè)由Session創(chuàng)建的對(duì)象，用來(lái)從Destination接收消息。 創(chuàng)建 MessageConsumer Java客戶端： ActiveMQSession方法： MessageConsumer createConsumer(Destination destination)。 MessageConsumer createConsumer(Destination destination, String messageSelector)。 MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal)。 TopicSubscriber createDurableSubscriber(Topic topic, String name)。 TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal)。其中messageSelector為消息選擇器；noLocal標(biāo)志默認(rèn)為false，當(dāng)設(shè)置為true時(shí)限制消費(fèi)者只能接收和自己相同的連接（Connection）所發(fā)布的消息，此標(biāo)志只適用于主題，不適用于隊(duì)列；name標(biāo)識(shí)訂閱主題所對(duì)應(yīng)的訂閱名稱，持久訂閱時(shí)需要設(shè)置此參數(shù)。例如：MessageConsumer consumer = (destination)。 消息的同步接收是指客戶端主動(dòng)去接收消息，客戶端可以采用MessageConsumer的receive()方法去接收下一個(gè)消息。消息的異步接收是指當(dāng)消息到達(dá)時(shí)，ActiveMQ主動(dòng)通知客戶端?？蛻舳丝梢酝ㄟ^(guò)注冊(cè)一個(gè)實(shí)現(xiàn)MessageListener 接口的對(duì)象到MessageConsumer。MessageListener只有一個(gè)必須實(shí)現(xiàn)的方法 onMessage()，它只接收一個(gè)參數(shù)，即Message。在為每個(gè)發(fā)送到Destination的消息實(shí)現(xiàn)onMessage時(shí)，將調(diào)用該方法。 Java客戶端： ActiveMQMessageConsumer方法： Message receive() Message receive(long timeout) Message receiveNoWait()其中timeout為等待時(shí)間，單位為毫秒?；蛘邔?shí)現(xiàn)MessageListener接口，每當(dāng)消息到達(dá)時(shí)，ActiveMQ會(huì)調(diào)用MessageListener中的onMessage()函數(shù)。例如：Message message = ()。 JMS提供了一種機(jī)制，使用它，消息服務(wù)可根據(jù)消息選擇器中的標(biāo)準(zhǔn)來(lái)執(zhí)行消息過(guò)濾。生產(chǎn)者可在消息中放入應(yīng)用程序特有的屬性，而消費(fèi)者可使用基于這些屬性的選擇標(biāo)準(zhǔn)來(lái)表明對(duì)消息是否感興趣。這就簡(jiǎn)化了客戶端的工作，并避免了向不需要這些消息的消費(fèi)者傳送消息的開(kāi)銷。然而，它也使得處理選擇標(biāo)準(zhǔn)的消息服務(wù)增加了一些額外開(kāi)銷。消息選擇器是用于MessageConsumer的過(guò)濾器，可以用來(lái)過(guò)濾傳入消息的屬性和消息頭部分（但不過(guò)濾消息體），并確定是否將實(shí)際消費(fèi)該消息。按照J(rèn)MS文檔的說(shuō)法，消息選擇器是一些字符串，它們基于某種語(yǔ)法，而這種語(yǔ)法是SQL92的子集?？梢詫⑾⑦x擇器作為MessageConsumer創(chuàng)建的一部分。 Java客戶端：例如：public final String SELECTOR = “JMSType = ‘TOPIC_PUBLISHER’”。 該選擇器檢查了傳入消息的 JMSType屬性，并確定了這個(gè)屬性的值是否等于 TOPIC _PUBLISHER。如果相等，則消息被消費(fèi)；如果不相等，那么消息會(huì)被忽略。 Message JMS程序的最終目的是生產(chǎn)和消費(fèi)的消息能被其他程序使用，JMS的Message是一個(gè)既簡(jiǎn)單又不乏靈活性的基本格式，允許創(chuàng)建不同平臺(tái)上符合非JMS程序格式的消息。Message由以下幾部分組成：消息頭，屬性和消息體。 Java客戶端：ActiveMQSession方法： BlobMessage createBlobMessage(File file) BlobMessage createBlobMessage(InputStream in) BlobMessage createBlobMessage(URL url) BlobMessage createBlobMessage(URL url, boolean deletedByBroker) BytesMessage createBytesMessage() MapMessage createMapMessage() Message createMessage() ObjectMessage createObjectMessage() ObjectMessage createObjectMessage(Serializable object) TextMessage createTextMessage() TextMessage createTextMessage(String text)例如：下例演示創(chuàng)建并發(fā)送一個(gè)TextMessage到一個(gè)隊(duì)列： TextMessage message = ()。 (msg_text)。 // msg_text is a String (message)。 下例演示接收消息并轉(zhuǎn)換為合適的消息類型： Message m = ()。 if (m instanceof TextMessage) { TextMessage message = (TextMessage) m。 (Reading message: + ())。 } else { // Handle error } 可靠性機(jī)制發(fā)送消息最可靠的方法就是在事務(wù)中發(fā)送持久性的消息，ActiveMQ默認(rèn)發(fā)送持久性消息。結(jié)束事務(wù)有兩種方法：提交或者回滾。當(dāng)一個(gè)事務(wù)提交，消息被處理。如果事務(wù)中有一個(gè)步驟失敗，事務(wù)就回滾，這個(gè)事務(wù)中的已經(jīng)執(zhí)行的動(dòng)作將被撤銷。接收消息最可靠的方法就是在事務(wù)中接收信息，不管是從PTP模式的非臨時(shí)隊(duì)列接收消息還是從Pub/Sub模式持久訂閱中接收消息。對(duì)于其他程序，低可靠性可以降低開(kāi)銷和提高性能，例如發(fā)送消息時(shí)可以更改消息的優(yōu)先級(jí)或者指定消息的過(guò)期時(shí)間。消息傳送的可靠性越高，需要的開(kāi)銷和帶寬就越多。性能和可靠性之間的折衷是設(shè)計(jì)時(shí)要重點(diǎn)考慮的一個(gè)方面?？梢赃x擇生成和使用非持久性消息來(lái)獲得最佳性能。另一方面，也可以通過(guò)生成和使用持久性消息并使用事務(wù)會(huì)話來(lái)獲得最佳可靠性。在這兩種極端之間有許多選擇，這取決于應(yīng)用程序的要求。 基本可靠性機(jī)制 控制消息的簽收（Acknowledgment）客戶端成功接收一條消息的標(biāo)志是這條消息被簽收。成功接收一條消息一般包括如下三個(gè)階段： 1． 客戶端接收消息；2．客戶端處理消息；3．消息被簽收。簽收可以由ActiveMQ發(fā)起，也可以由客戶端發(fā)起，取決于Session簽收模式的設(shè)置。在帶事務(wù)的Session中，簽收自動(dòng)發(fā)生在事務(wù)提交時(shí)。如果事務(wù)回滾，所有已經(jīng)接收的消息將會(huì)被再次傳送。在不帶事務(wù)的Session中，一條消息何時(shí)和如何被簽收取決于Session的設(shè)置。 1．當(dāng)客戶端從receive或onMessage成功返回時(shí)，Session自動(dòng)簽收客戶端的這條消息的收條。在AUTO_ACKNOWLEDGE的Session中，同步接收receive是上述三個(gè)階段的一個(gè)例外，在這種情況下，收條和簽收緊隨在處理消息之后發(fā)生。 2．客戶端通過(guò)調(diào)用消息的acknowledge方法簽收消息。在這種情況下，簽收發(fā)生在Session層面：簽收一個(gè)已消費(fèi)的消息會(huì)自動(dòng)地簽收這個(gè)Session所有已消費(fèi)消息的收條。 3．此選項(xiàng)指示Session不必確保對(duì)傳送消息的簽收。它可能引起消息的重復(fù)，但是降低了Session的開(kāi)銷，所以只有客戶端能容忍重復(fù)的消息，才可使用（如果ActiveMQ再次傳送同一消息那么消息頭中的JMSRedelivered將被設(shè)置為true）。 Java客戶端：簽收模式分別為： 1． 2． 3． ActiveMQConnection方法： Session createSession(boolean transacted, int acknowledgeMode)。例如：Session session = (false, )。 對(duì)隊(duì)列來(lái)說(shuō)，如果當(dāng)一個(gè)Session終止時(shí)它接收了消息但是沒(méi)有簽收，那么ActiveMQ將保留這些消息并將再次傳送給下一個(gè)進(jìn)入隊(duì)列的消費(fèi)者。對(duì)主題來(lái)說(shuō)，如果持久訂閱用戶終止時(shí)，它已消費(fèi)未簽收的消息也將被保留，直到再次傳送給這個(gè)用戶。對(duì)于非持久訂閱，AtiveMQ在用戶Session關(guān)閉時(shí)將刪除這些消息。如果使用隊(duì)列和持久訂閱，并且Session沒(méi)有使用事務(wù)，那么可以使用Session的recover方法停止Session，再次啟動(dòng)后將收到它第一條沒(méi)有簽收的消息，事實(shí)上，重啟后Session一系列消息的傳送都是以上一次最后一條已簽收消息的下一條為起點(diǎn)。如果這時(shí)有消息過(guò)期或者高優(yōu)先級(jí)的消息到來(lái)，那么這時(shí)消息的傳送將會(huì)和最初的有所不同。對(duì)于非持久訂閱用戶，重啟后，ActiveMQ有可能刪除所有沒(méi)有簽收的消息。 指定消息傳送模式 ActiveMQ支持兩種消息傳送模式：PERSISTENT和NON_PERSISTENT兩種。 1．PERSISTENT（持久性消息）這是ActiveMQ的默認(rèn)傳送模式，此模式保證這些消息只被傳送一次和成使用一次。對(duì)于這些消息，可靠性是優(yōu)先考慮的因素?？煽啃缘牧硪粋€(gè)重要方面是確保持久性消息傳送至目標(biāo)后，消息服務(wù)在向消費(fèi)者傳送它們之前不會(huì)丟失這些消息。這意味著在持久性消息傳送至目標(biāo)時(shí)，消息服務(wù)將其放入持久性數(shù)據(jù)存儲(chǔ)。如果消息服務(wù)由于某種原因?qū)е率?，它可以恢?fù)此消息并將此消息傳送至相應(yīng)的消費(fèi)者。雖然這樣增加了消息傳送的開(kāi)銷，但卻增加了可靠性。 2．NON_PERSISTENT（非持久性消息）保證這些消息最多被傳送一次。對(duì)于這些消息，可靠性并非主要的考慮因素。此模式并不要求持久性的數(shù)據(jù)存儲(chǔ)，也不保證消息服務(wù)由于某種原因?qū)е率『笙⒉粫?huì)丟失。有兩種方法指定傳送模式： 1．使用 setDeliveryMode方法，這樣所有的消息都采用此傳送模式； 2．使用 send方法為每一條消息設(shè)置傳送模式； Java客戶端：傳送模式分別為： 1． 2． ActiveMQMessageProducer方法： void setDeliveryMode(int newDeliveryMode)?；蛘?void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive)。 void send(Message message, int deliveryMode, int priority, long timeToLive)。其中deliveryMode