前言

ActiveMQ 是Apache出品，最流行的，能力強(qiáng)勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規(guī)范的 JMS Provider實(shí)現(xiàn)，盡管JMS規(guī)范出臺已經(jīng)是很久的事情了，但是JMS在當(dāng)今的J2EE應(yīng)用中間仍然扮演著特殊的地位。

在生產(chǎn)項目中，很多時候需要消息中間件來進(jìn)行分布式系統(tǒng)間的通信。它具有低耦合、可靠投遞、廣播、流量控制、最終一致性等一系列功能。本篇主要介紹ActiveMQ 相關(guān)概念以及安裝說明，后面會著重介紹 SpringBoot 集成實(shí)現(xiàn)秒殺消息隊列。

概念

JMS消息模式

點(diǎn)對點(diǎn)或隊列模式

包含三個角色：消息隊列（Queue），發(fā)送者(Sender)，接收者(Receiver)。每個消息都被發(fā)送到一個特定的隊列，接收者從隊列中獲取消息。隊列保留著消息，直到他們被消費(fèi)或超時。

• 每個消息只有一個消費(fèi)者（Consumer），即一旦被消費(fèi)，消息就不再在消息隊列中
• 發(fā)送者和接收者之間在時間上沒有依賴性，也就是說當(dāng)發(fā)送者發(fā)送了消息之后，不管接收者有沒有正在運(yùn)行，它不會影響到消息被發(fā)送到隊列
• 接收者在成功接收消息之后需向隊列應(yīng)答成功

Pub/Sub 發(fā)布/訂閱模式

包含三個角色：主題（Topic），發(fā)布者（Publisher），訂閱者（Subscriber） 。多個發(fā)布者將消息發(fā)送到Topic,系統(tǒng)將這些消息傳遞給多個訂閱者。

• 每個消息可以有多個消費(fèi)者
• 發(fā)布者和訂閱者之間有時間上的依賴性。針對某個主題（Topic）的訂閱者，它必須創(chuàng)建一個訂閱者之后，才能消費(fèi)發(fā)布者的消息。
• 為了消費(fèi)消息，訂閱者必須保持運(yùn)行的狀態(tài)。

為了緩和這樣嚴(yán)格的時間相關(guān)性，JMS允許訂閱者創(chuàng)建一個可持久化的訂閱。這樣，即使訂閱者沒有被激活（運(yùn)行），它也能接收到發(fā)布者的消息。

如果希望發(fā)送的消息可以不被做任何處理、或者只被一個消息者處理、或者可以被多個消費(fèi)者處理的話，那么可以采用Pub/Sub模型。

JMS消息基本組件

ConnectionFactory

創(chuàng)建Connection對象的工廠，針對兩種不同的jms消息模型，分別有QueueConnectionFactory和TopicConnectionFactory兩種?？梢酝ㄟ^JNDI來查找ConnectionFactory對象。

Destination

Destination的意思是消息生產(chǎn)者的消息發(fā)送目標(biāo)或者說消息消費(fèi)者的消息來源。對于消息生產(chǎn)者來說，它的Destination是某個隊列（Queue）或某個主題（Topic）;對于消息消費(fèi)者來說，它的Destination也是某個隊列或主題（即消息來源）。

所以，Destination實(shí)際上就是兩種類型的對象：Queue、Topic可以通過JNDI來查找Destination。

Connection

Connection表示在客戶端和JMS系統(tǒng)之間建立的鏈接（對TCP/IP socket的包裝）。Connection可以產(chǎn)生一個或多個Session。跟ConnectionFactory一樣，Connection也有兩種類型：QueueConnection和TopicConnection。

Session

Session是操作消息的接口。可以通過session創(chuàng)建生產(chǎn)者、消費(fèi)者、消息等。Session提供了事務(wù)的功能。當(dāng)需要使用session發(fā)送/接收多個消息時，可以將這些發(fā)送/接收動作放到一個事務(wù)中。同樣，也分QueueSession和TopicSession。

消息的生產(chǎn)者

消息生產(chǎn)者由Session創(chuàng)建，并用于將消息發(fā)送到Destination。同樣，消息生產(chǎn)者分兩種類型：QueueSender和TopicPublisher?？梢哉{(diào)用消息生產(chǎn)者的方法（send或publish方法）發(fā)送消息。

消息消費(fèi)者

消息消費(fèi)者由Session創(chuàng)建，用于接收被發(fā)送到Destination的消息。兩種類型：QueueReceiver和TopicSubscriber?？煞謩e通過session的createReceiver(Queue)或createSubscriber(Topic)來創(chuàng)建。當(dāng)然，也可以session的creatDurableSubscriber方法來創(chuàng)建持久化的訂閱者。

MessageListener

消息監(jiān)聽器。如果注冊了消息監(jiān)聽器，一旦消息到達(dá)，將自動調(diào)用監(jiān)聽器的onMessage方法。EJB中的MDB（Message-Driven Bean）就是一種MessageListener。

Transport傳輸方式

ActiveMQ目前支持的Transport有：VM Transport、TCP Transport、NIO Transport、SSL Transport、Peer Transport、UDP Transport、Multicast Transport、HTTP and HTTPS Transport、WebSockets Transport、Failover Transport、Fanout Transport、Discovery Transport、ZeroConf Transport等。

• VM Transport：允許客戶端和Broker直接在VM內(nèi)部通信，采用的連接不是Socket連接，而是直接的方法調(diào)用，從而避免了網(wǎng)絡(luò)傳輸?shù)拈_銷。應(yīng)用場景也僅限于Broker和客戶端在同一JVM環(huán)境下。
• TCP Transport：客戶端通過TCP Socket連接到遠(yuǎn)程Broker。配置語法：
• tcp://hostname:port?transportOptions
• HTTP and HTTPS Transport：允許客戶端使用REST或者Ajax的方式進(jìn)行連接。這意味著可以直接使用Javascript向ActiveMQ發(fā)送消息。
• WebSockets Transport：允許客戶端通過HTML5標(biāo)準(zhǔn)的WebSockets方式連接到Broker。
• Failover Transport：青龍系統(tǒng)MQ采用的就是這種連接方式。這種方式具備自動重新連接的機(jī)制，工作在其他Transport的上層，用于建立可靠的傳輸。允許配置任意多個的URI，該機(jī)制將會自動選擇其中的一個URI來嘗試連接。配置語法：
• failover:(tcp://localhost:61616,tcp://localhost:61617,.....)?transportOptions
• Fanout Transport：主要適用于生產(chǎn)消息發(fā)向多個代理。如果多個代理出現(xiàn)環(huán)路，可能造成消費(fèi)者接收重復(fù)的消息。所以，使用該協(xié)議時，最好將消息發(fā)送給多個不相連接的代理。

 Persistence持久化存儲

AMQ Message Store

ActiveMQ 5.0 的缺省持久化存儲方式。

Kaha Persistence

這是一個專門針對消息持久化的解決方案。它對典型的消息使用模式進(jìn)行了優(yōu)化。

JDBC Persistence

目前支持的數(shù)據(jù)庫有：Apache Derby, Axion, DB2, HSQL, Informix, MaxDB, MySQL, Oracle, Postgresql, SQLServer, Sybase。

Disable Persistence

不應(yīng)用持久化存儲。

集群方案(Master / Slave)

Pure Master Slave

• 無單點(diǎn)故障；
• 不需要依賴共享文件系統(tǒng)或是共享數(shù)據(jù)庫，使用 KahaDB的方式持久化存儲；
• 一個Master只能帶一個Slave；
• Master工作期間，會將消息狀況自動同步到Slave；
• Master一旦崩潰，Slave自動接替其工作，已發(fā)送并尚未消費(fèi)的消息繼續(xù)有效；
• Slave接手后，必須停止Slave才能重啟先前的Master；

 Shared File System Master Slave

JDBC Master Slave

• 配置上，不存在Master和Slave的區(qū)分，多個共享數(shù)據(jù)源的Broker構(gòu)成JDBC Master Slave；
• 首先搶到資源（數(shù)據(jù)庫鎖）的Broker成為Master，其他Broker定期嘗試搶占資源；
• 一旦Master崩潰，其他Broker搶占資源，最終只有一臺搶到，立刻成為Master，之前的Master即便重啟成功，也只能作為Slave等待；

 安裝說明

這里使用Docker安裝，查詢Docker鏡像：

docker search activemq

下載Docker鏡像：

docker pull webcenter/activemq

創(chuàng)建&運(yùn)行ActiveMQ容器：

docker run -d --name myactivemq -p 61617:61616 -p 8162:8161 webcenter/activemq

61616是 activemq 的容器使用端口（映射為61617），8161是 web 頁面管理端口（對外映射為8162）

查看創(chuàng)建的容器，如果存在說明安裝成功：

docker ps

查看WEB管理頁面：

瀏覽器輸入 http://ip :8162 點(diǎn)擊Manage ActiveMQ broker使用默認(rèn)賬號/密碼：admin/admin進(jìn)入查看。

配置訪問密碼

進(jìn)入Docker容器：

docker exec -it myactivemq /bin/bash

控制臺界面設(shè)置用戶名和密碼：

# 位于根目錄 conf 目錄下
vi jetty-realm.properties

# 修改密碼
# username: password [,rolename ...]
admin: admin, admin

配置連接密碼

編輯activemq.xml文件，放置到 shutdownHooks 下方即可。

<!-- 添加訪問ActiveMQ的賬號密碼 -->
<plugins>
  <simpleAuthenticationPlugin>
    <users>
      <authenticationUser username="${activemq.username}" password="${activemq.password}" groups="users,admins"/>
    </users>
  </simpleAuthenticationPlugin>
</plugins>

修改conf中credentials.properties文件進(jìn)行密碼設(shè)置：

activemq.username=admin
activemq.password=123456
guest.password=123456

注意事項

如果是云服務(wù)器，記得開放相關(guān)端口(61617/8160)

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。