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