jms

提出的指在統(tǒng)一各種MOM（Message-Oriented Middleware ）系統(tǒng)接口的規(guī)范，只是接口，不包含實(shí)現(xiàn)，實(shí)現(xiàn)JMS 接口的消息中間件稱為JMS Provider。activemq是Apache出品的開源項(xiàng)目，它是JMS規(guī)范的一個(gè)實(shí)現(xiàn)

接口構(gòu)成:

ConnectionFactory:連接工廠接口用于建立連接

Connection:封裝了JMS 客戶端到JMS Provider 的連接

Destination:用來指定生產(chǎn)的消息的目標(biāo)和它消費(fèi)的消息的來源的對象

Session:單線程上下文創(chuàng)建生產(chǎn)者、消費(fèi)者。消息等

MessageProducer:由Session 對象創(chuàng)建的用來發(fā)送消息的對象

MessageConsumer:由Session 對象創(chuàng)建的用來接收消息的對象

消息結(jié)構(gòu):

消息頭:1、自動分配的消息頭：JMSDeliveryMode、JMSMessageID、JMSTimestamp、JMSExpiration、JMSRedelivered、JMSPriority2、開發(fā)者分配的消息頭：JMSReplyTo、JMSCorrelationID、JMSType

消息屬性:1. 應(yīng)用需要用到的屬性;2. 消息頭中原有的一些可選屬性;3. JMS Provider 需要用到的屬性

消息體:1、TextMessage2、MapMessage3、BytesMessage4、StreamMessage5、ObjectMessage6、Message

開發(fā)步驟:

1.創(chuàng)建ConnectionFactory2.創(chuàng)建Connection3.創(chuàng)建一個(gè)或多個(gè)JMS Session4.創(chuàng)建Queue或TopicSession5.創(chuàng)建MessageProducer或MessageConsumer6.發(fā)送或者接受消息:(1、異步接收MessageConsumer:setMessageListener(MessageListener m)2、同步接收MessageConsumer：receive() 以上兩種接收方式返回Message 對象)

activeMq

模型分析:

生產(chǎn)者通過TransportConnection對象發(fā)送信息,TransportConnection對象指向一個(gè)regionBroker,這個(gè)regionBroker(區(qū)域中間件)中包含了四種區(qū)域(隊(duì)列域(queueRegion)、主題域(topicRegion)、臨時(shí)隊(duì)列域(tempQueueRegion)、臨時(shí)主題域(tempTopicRegion)),而每個(gè)Region(域)中包含了N個(gè)destination(目的地對象),生產(chǎn)者的信息通過regionBroker傳給Region后又傳給destination,在destination中進(jìn)行持久化操作,并且"待發(fā)送消息指針"對象指向待發(fā)送消息的位置,然后當(dāng)有消息訂閱時(shí)destination將信息發(fā)送給訂閱者,(這里注意,訂閱信息也在region中),然后訂閱機(jī)制通過transportConnection發(fā)送消息給消費(fèi)者進(jìn)程

消息指針:通常消息在未發(fā)生或者發(fā)送后未收到消費(fèi)者的確認(rèn)信息時(shí)都會持久保存消息到存儲中。當(dāng)有消費(fèi)者來可以消費(fèi)消息時(shí)，broker會批量從存儲中取出消息，發(fā)送給消費(fèi)者。游標(biāo)就是指向下次批量獲取消息時(shí)的存儲位置,有如下幾種:

• Store-based cursors ?基于存儲的，慢速消費(fèi)者需要使用pending cursor來指定下一次批量讀取消息的位置
• VM cursors?內(nèi)存保存一些游標(biāo)
• File-based cursors? 內(nèi)存超過閾值，存儲到臨時(shí)文件。

通信機(jī)制:

消息終結(jié)條件:?Consume ack?消息過期 ?存儲設(shè)備溢出

消息發(fā)送過程如圖:

這里主要涉及到異步發(fā)送和同步發(fā)送,異步發(fā)送不會返回發(fā)送結(jié)果,但是效率較高,適用于大量并且允許丟失的數(shù)據(jù),同步發(fā)送會阻塞并返回發(fā)送結(jié)果,適用于不可丟失的情況

判定異步發(fā)送還是同步發(fā)送是在mq代碼中判斷的,非持久化消息是異步發(fā)送的，持久化消息并且是在非事務(wù)模式下是同步發(fā)送的。但是在開啟事務(wù)的情況下，消息都是異步發(fā)送。至于持久化消息非持久化消息,以及事務(wù)模式在后邊"可靠機(jī)制"章節(jié)中講解

消息接收流程:

1,服務(wù)器發(fā)送消息有一個(gè)隊(duì)列,如果隊(duì)列滿了需要等待消息的回執(zhí)(ack)來清理消息,后繼續(xù)發(fā)送,消費(fèi)者接收信息后需要返回回執(zhí)信息ack

2.消費(fèi)者客戶端用receive方法可以定義欲獲取消息個(gè)數(shù),如果大于0則通過拉取的方式獲取相應(yīng)數(shù)量的信息,如果小于0則通過推送的方式獲取消息,獲取隊(duì)列中沒有消息則阻塞

3.消費(fèi)者客戶端用messagelistener方法異步獲取消息時(shí)調(diào)用onmessage方法處理消息

可靠機(jī)制:

指定消息傳送模式:

1．NON_PERSISTENT（非持久性消息）

??? 保證這些消息最多被傳送一次。對于這些消息，可靠性并非主要的考慮因素。此模式并不要求持久性的數(shù)據(jù)存儲，也不保證消息服務(wù)由于某種原因?qū)е率『笙⒉粫G失。

2．PERSISTENT（持久性消息）

??????? 這是ActiveMQ的默認(rèn)傳送模式，此模式保證這些消息只被傳送一次和成功使用一次。對于這些消息，可靠性是優(yōu)先考慮的因素。可靠性的另一個(gè)重要方面是確保持久性消息傳送至目標(biāo)后，消息服務(wù)在向消費(fèi)者傳送它們之前不會丟失這些消息。

有兩種方法指定傳送模式：

1．使用setDeliveryMode方法，這樣所有的消息都采用此傳送模式；

2．使用send方法為每一條消息設(shè)置傳送模式；

默認(rèn)存儲方式:kahadb,文件消息存儲器

事務(wù)和消息的簽收ack:

簽收可以由ActiveMQ發(fā)起，也可以由客戶端發(fā)起，取決于Session簽收模式的設(shè)置。

?在帶事務(wù)的Session中，簽收自動發(fā)生在事務(wù)提交時(shí)。如果事務(wù)回滾，所有已經(jīng)接收的消息將會被再次傳送。Session配置為SESSION_TRANSACTED。

?在不帶事務(wù)的Session中，一條消息何時(shí)和如何被簽收取決于Session的設(shè)置控制消息的簽收

• 1．Session.AUTO_ACKNOWLEDGE（自動簽收）
• 2．Session.CLIENT_ACKNOWLEDGE（客戶端簽收）
• 3．Session.DUPS_OK_ACKNOWLEDGE（不必確保對傳送消息的簽收）

???????消息的重傳:

?觸發(fā)條件 CLIENT_ACKNOWLEDGE+recover(),事務(wù)rollback(),AUTO_ACKNOWLEDGE異常

?RedeliveryPolicy 配置對應(yīng)的重發(fā)機(jī)制

?重傳次數(shù)配置

?重發(fā)最終失敗會丟棄到死信隊(duì)列

處理失敗的過期的、處理失敗的消息，將會被ActiveMQ置入“ActiveMQ.DLQ”這個(gè)死信隊(duì)列中 死信隊(duì)列可以拿到,通過監(jiān)聽死信隊(duì)列獲取這些信息? 死信隊(duì)列支持配置 比如消息過期的不放進(jìn)來

設(shè)置消息優(yōu)先級:

普通情況下可以確保將單個(gè)會話向目標(biāo)發(fā)送的所有消息按其發(fā)送順序傳送至消費(fèi)者。然而，如果為這些消息分配了不同的優(yōu)先級，消息傳送系統(tǒng)將首先嘗試傳送優(yōu)先級較高的消息。

有兩種方法設(shè)置消息的優(yōu)先級：

1．使用setDeliveryMode方法，這樣所有的消息都采用此傳送模式；

2．使用send方法為每一條消息設(shè)置傳送模式；

注意：消息級別分為0至9，其中0至4是普通級別，5至9為加急消息

允許消息過期:

默認(rèn)情況下，消息永不會過期。如果消息在特定周期內(nèi)失去意義，那么可以設(shè)置過期時(shí)間。

有兩種方法設(shè)置消息的過期時(shí)間，時(shí)間單位為毫秒：

1．使用setTimeToLive方法為所有的消息設(shè)置過期時(shí)間；

2．使用send方法為每一條消息設(shè)置過期時(shí)間；

注意：是在消息發(fā)送端調(diào)用setTimeToLive方法。

創(chuàng)建臨時(shí)目標(biāo)

ActiveMQ通過createTemporaryQueue和createTemporaryTopic創(chuàng)建臨時(shí)目標(biāo)，這些目標(biāo)持續(xù)到創(chuàng)建它的Connection關(guān)閉。只有創(chuàng)建臨時(shí)目標(biāo)的Connection所創(chuàng)建的客戶端才可以從臨時(shí)目標(biāo)中接收消息，但是任何的生產(chǎn)者都可以向臨時(shí)目標(biāo)中發(fā)送消息。如果關(guān)閉了創(chuàng)建此目標(biāo)的Connection，那么臨時(shí)目標(biāo)被關(guān)閉，內(nèi)容也將消失。

Topic持久訂閱:

消息訂閱分為非持久訂閱(non-durable subscription)和持久訂閱(durable subscription)，非持久訂閱只有當(dāng)客戶端處于激活狀態(tài)，也就是和ActiveMQ保持連接狀態(tài)才能收到發(fā)送到某個(gè)主題的消息，而當(dāng)客戶端處于離線狀態(tài)，這個(gè)時(shí)間段發(fā)到主題的消息將會丟失，永遠(yuǎn)不會收到。

建立持久訂閱的步驟：

1 .? 生產(chǎn)者設(shè)置為持久化模式

producer.setDeliveryMode(DeliveryMode.PERSISTENT);

1．?消費(fèi)者 為連接設(shè)置一個(gè)客戶ID；

topicConnection.setClientID("clientc");

2．? 消費(fèi)者為訂閱的主題指定一個(gè)訂閱名稱；topicSession.createDurableSubscriber(mytopic,"clientc")

注意：上述組合必須唯一。

持久訂閱例子方法：TopicSubscriber createDurableSubscriber(Topic topic, String name)

非持久化訂閱例子：session.createSubscriber(topic);//無需向connection指定clientId

lMessage Groups 消息分組

?針對queue 一個(gè)隊(duì)列 對應(yīng)多個(gè)消費(fèi)者，默認(rèn)消費(fèi)者會去競爭獲取消息。

?利用消息分組，具有相同groupId的消息會被投送到同一個(gè)消費(fèi)者（只要這個(gè)consumer保持active）

?負(fù)載均衡? broken收到消息，檢查消息JMSXGroupID屬性。如果存在，那么broker會檢查是否有某個(gè)consumer擁有這個(gè)message group。如果沒有，那么broker會選擇一個(gè)consumer，并將它關(guān)聯(lián)到這個(gè)message group。此后，這個(gè)consumer會接收這個(gè)message group的所有消息

• message.setStringProperty("JMSXGroupID","GroupA");

???????spring集成jms:

Spring框架提供一個(gè)模板機(jī)制JMSTemplate來隱藏Java API細(xì)節(jié)。參考源碼類JMSTemplate.java

Spring提供了JMSTemplate類，所以開發(fā)者不必為JMS實(shí)現(xiàn)編寫樣本代碼。何為模板模式,請參考https://blog.csdn.net/l450741881/article/details/88837064

總結(jié)

以上是生活随笔為你收集整理的jms规范以及activeMq相关介绍的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。