activemq in action學習筆記

時間 2019-11-19

原文原文鏈接

JMS基本概念

JMS 是 SUN 公司開發的一套訪問 MOM(Message-Oriented-Middleware) 消息服務中間件的標準 API。
html

MOM 提供消息接收和轉發的服務,對消息進行緩存和持久操做,保證消息的安全性,JMS讓開發都無須瞭解遠程過程調用的細節和網絡通訊協議的細節就能夠經過JMS向MOM發送消息,藉助消息咱們能夠鬆散耦合的方式集成不一樣的應用。spring

JMS(Java Message Service)即Java消息服務。它提供標準的產生、發送、接收消息的接口簡化企業應用的開發。它支持兩種消息通訊模型：點到點（point-to-point）（P2P）模型和發佈/訂閱（Pub/Sub）模型。P2P 模型規定了一個消息只能有一個接收者;Pub/Sub 模型容許一個消息能夠有多個接收者。
apache

對於點到點模型，消息生產者產生一個消息後，把這個消息發送到一個Queue（隊列）中，而後消息接收者再從這個Queue中讀取，一旦這個消息被一個接收者讀取以後，它就在這個Queue中消失了，因此一個消息只能被一個接收者消費。能夠定義多個消費者，可是同一個消息只會被一個消費者消費；數組
與點到點模型不一樣，發佈/訂閱模型中，消息生產者產生一個消息後，把這個消息發送到一個Topic中，這個Topic能夠同時有多個接收者在監聽，當一個消息到達這個Topic以後，全部消息接收者都會收到這個消息。緩存

點對對點消息傳遞域的特色：
安全

每一個消息只能有一個消費者；網絡
消息的生產者和消費者之間沒有時間上的相關性。不管消費者在生產者發送消息的時候是否處於運行狀態，它均可以提取消息。併發

發佈/訂閱消息傳遞域的特色：
app

每一個消息能夠有多個消費者；dom
生產者和消費者之間有時間上的關聯性。訂閱一個主題的消費者只能消費自它訂閱以後發佈的消息。JMS規範容許客戶建立持久閱讀，這在必定程度上放鬆了時間上的相關性要求。持久訂閱容許消費者消費它在未處於激活狀態時發送的消息。

幾個概念

Destination——消息發送的目的地，也就是前面說的Queue和Topic。建立好一個消息以後，只須要把這個消息發送到目的地，消息的發送者就能夠繼續作本身的事情，而不用等待消息被處理完成。至於這個消息何時，會被哪一個消費者消費，徹底取決於消息的接受者。
Connection——與JMS提供者（IBM Webshphere MQ, ActiveMQ等）創建的一個鏈接。能夠從這個鏈接建立一個會話，即Session。
ConnectionFactory——那如何建立一個Connection呢？ConnectionFactory。經過這個工廠類就能夠獲得一個與JMS提供者的鏈接，即Conection。
Session——與JMS提供者所創建的會話，經過Session咱們才能夠建立一個Message。
Producer——消息的生產者，要發送一個消息，必須經過這個生產者來發送。
MessageConsumer——與生產者相對應，這是消息的消費者或接收者，經過它來接收一個消息。
Message——從字面上就能夠看出是被髮送的消息。它有下面幾種類型：StreamMessage：Java 數據流消息，用標準流操做來順序的填充和讀取。MapMessage：一個Map類型的消息；名稱爲 string 類型，而值爲 Java 的基本類型。TextMessage：普通字符串消息，包含一個String。ObjectMessage：對象消息，包含一個可序列化的Java 對象BytesMessage：二進制數組消息，包含一個byte[]。XMLMessage: 一個XML類型的消息。最經常使用的是TextMessage和ObjectMessage。
JMS 事務——JMS 使用 Session 控制事務 , 如何須要事務能夠在建立 Session 時標註須要事務。

ConnectionFactory和Destination經過JMS提供者提供給咱們的接口獲得。

消息的消費能夠採用如下兩種方法之一：

同步消費——經過調用消費者的receive方法從目的地中顯式地提取消息。receive方法能夠一直阻塞到消息到達；
異步消費——客戶能夠爲消費者註冊一個消息監聽器，以定義在消息到達時所採起的動做。

JMS支持如下兩種消息提交模式：

PERSISTENT——指示JMS provider持久保存消息，以保證消息不會由於JMS provider的失敗而丟失。
NON_PERSISTENT——不要求JMS provider持久保存消息。

JMS消息只有在被確認以後，才認爲已經被成功消費。消息的成功消費一般包含三個階段：客戶接收消息、客戶處理消息和消息被確認。

在事務性會話中，當一個事務被提交時，確認自動發生。在非事務性會話中，消息什麼時候被確認取決於建立會話時的應答模式（acknowledgement mode）。該參數有如下三個可選值：

Session.Auto_ACKNOWLEDGE。當客戶成功地從receive方法返回的時候，或者從MessageListener.onMessage方法成功返回的時候，會話自動確認客戶收到的消息。
Session.CLIENT_ACKNOWLEDGE。客戶經過消息的acknowledge方法確認消息。須要注意的是，在這種模式中，確認是在會話層進行：確認一個被消費的消息將自動確認全部已被會話消費的消息。例如，若是一個消息消費者消費了10個消息，而後確認第5個消息，那個全部10個消息都被確認。
Session.DUPS_ACKNOWLEDGE。在會話遲鈍等狀況下確認消息。若是JMS provider失敗，那麼可能會致使一些重複的消息。若是是重複的消息，那麼JMS provider必須把消息頭的JMSRecelivered字段設置爲true。

JMS消息頭

JMS Headers set automatically by the client’s send:

JMSDestination

JMSDeliveryMode 持久化/非持久化

JMSExpiration 消息過時時間

JMSMessageID

JMSPriority 消息優先級 0-9 9爲最高級

JMSTimestamp 消息被建立的時間

JMS Headers set optionally by the client:

MSCorrelationID Used to associate the current message with a previous message

JMSReplyTo Used to specify a destination where a reply should be sent

Headers set optionally by the JMS provider:

JMSRedelivered Used to indicate the liklihood that a message was previously delivered but not acknowledged.

JMS 消息屬性

CUSTOM PROPERTIES，自定義屬性

經過set or get 方法操做；

JMS-DEFINED PROPERTIES ，these properties is optional:

 JMSXAppID—Identifies the application sending the message

 JMSXConsumerTXID —The transaction identifier for the transaction within which this message was consumed

 JMSXDeliveryCount—The number of message delivery attempts

 JMSXGroupID—The message group of which this message is a part

 JMSXGroupSeq—The sequence number of this message within the group

 JMSXProducerTXID—The transaction identifier for the transaction within which this message was produced

 JMSXRcvTimestamp —The time the JMS provider delivered the message to the consumer

 JMSXState—Used to define a provider-specific state

 JMSXUserID —Identifies the user sending the message

JMS Message Selector 消息選擇器

選擇器語法：

Literals Booleans TRUE/FALSE; numbers such as 5, -10, +34; numbers with decimal or scientific notation such as 43.3E7, +10.5239
Identifiers A header or property field
Operators AND, OR, LIKE, BETWEEN, =, <>, <, >, <=, =>, +, -, *, /, IS NULL, IS NOT NULL

MESSAGE BODY

JMS defines six Java types for the message body, also known as the payload. Through the use of these objects, data and information can be sent via the message payload.

Message —The base message type. Used to send a message with no payload, only headers and properties. Typically used for simple event notification.

TextMessage —A message whose payload is a String. Commonly used to send simple textual and XML data.

MapMessage —Uses a set of name/value pairs as its payload. The names are of type String and the values are a Java primitive type.
BytesMessage —Used to contain an array of uninterpreted bytes as the payload.
StreamMessage—A message with a payload containing a stream of primitive Java types that’s filled and read sequentially.

ObjectMessage—Used to hold a serializable Java object as its payload. Usually used for complex Java objects. Also supports Java collections.

MESSAGE DURABILITY 和 MESSAGE PERSISTENCE 的區別

durable模式下，消息發送到隊列以前存在的全部訂閱者都消費以後，消息移除；若果訂閱者中間斷開，消息會保留，知道訂閱者再次鏈接後消費完成。
非durable模式下，當前存在訂閱者都消費以後，消息移除；

transport connectors：
支持協議類型：

 支持協議類型：TCP NIO and UDP broker.SSL HTTP(S)ker.VM

服務端定義
<transportConnectors>
<transportConnector  name="tcp" uri="tcp://localhost:61616?trace=true"/> 
<transportConnector  name="udp" uri="udp://localhost:61618?trace=true" /> 
</transportConnectors>

network connectors
·Static networks

 static:(uri1,uri2,uri3,...)?key=value
<transportConnectors>
<transportConnector  name="tcp" uri="tcp://localhost:61616?trace=true"/> 
</transportConnectors>
      <networkConnectors>
<networkConnector name="local network" uri="static://(tcp://remotehost1:61616,tcp://remotehost2:61616)"/>
 </networkConnectors>
說明：客戶端鏈接併發送消息到broker 61616，而後61616 broker會將消息轉發到remotehost1:61616 broker；


FAILOVER PROTOCOL，採用隨機的方式進行broker鏈接，能夠經過設置randomize=false使其默認鏈接uri1,只有當uri1不可用時，才鏈接uri2，實現Master/Slave模式；
failover:(uri1,...,uriN)?key=value

·Dynamic networks

MULTICAST CONNECTOR（服務端）

multicast://ipadaddress:port?key=value
<networkConnectors>
<networkConnector name="default-nc" uri="multicast://default"/>
</networkConnectors>
<transportConnectors>
<transportConnector name="openwire" uri="tcp://localhost:61616"  discoveryUri="multicast://default"/> 
</transportConnectors>

   DISCOVERY PROTOCOL（客戶端）    
       discovery:(discoveryAgentURI)?key=value
       discovery:(multicast://default)
   PEER PROTOCOL 對等協議，用於兩個embedded broker之間實現network connection，以下圖：

   FANOUT CONNECTOR 分列鏈接器，客戶端鏈接協議，只能用於生產者，將消息複製到多個broker上，且各個broker之間不相互鏈接，不然可能出現一個消息被同一個消費者屢次消費；
    fanout:(fanoutURI)?key=value
    fanout:(static:(tcp://host1:61616,tcp://host2:61616,tcp://host3:61616))
    fanout:(multicast://default)（前提是broker採用multicast 協議）

關於鏈接器總結：

message storage

KahaDB message store
AMQ message store internals

適用於高吞吐率場景，不用於存在大量destination的場景，由於每個destination會對應一個index文件。
默認使用NIO
基本配置方式：

<persistenceAdapter>
   <amqPersistenceAdapter  directory="target/Broker2-data/activemq-data" syncOnWrite="true" indexPageSize="16kb" 
indexMaxBinSize="100"  maxFileLength="10mb" /> 
</persistenceAdapter>