歡迎光臨
每天分享高質量文章

ActiveMQ · 基礎篇

(給ImportNew加星標,提高Java技能)

 

轉自:cyfonly

 

過了個春節,回到公司的成小胖變成了成大胖。但是你們千萬別以為他那個大肚子裡面裝的都是肥肉,裡面的墨水也多了不少嘞,畢竟成小胖利用春節的半個月時間專心學習並研究了 ActiveMQ,嘿嘿……

 

這不,為了檢驗下自己的學習成果,上班的第一天成小胖就去找架構師老王交流 ActiveMQ 相關的知識,還順便向老王討了個紅包,可把成小胖給高興壞了。

 

“來,根據你的瞭解說下 ActiveMQ 是什麼。”

 

“這個簡單,ActiveMQ 是一個 MOM,具體來說是一個實現了 JMS 規範的系統間遠程通信的訊息代理。它……”

 

“等等,先解釋下什麼是 MOM。”

 

“好。MOM 就是面向訊息中間件(Message-oriented middleware),是用於以分佈式應用或系統中的異步、松耦合、可靠、可擴展和安全通信的一類軟體。MOM 的總體思想是它作為訊息發送器和訊息接收器之間的訊息中介,這種中介提供了一個全新水平的松耦合。”

 

“JMS呢?”

 

成小胖是個追求極致的人,為瞭解釋得更通俗易懂,索性搬來一塊白板邊畫邊說。

 

“JMS 叫做 Java 訊息服務(Java Message Service),是 Java 平臺上有關面向 MOM 的技術規範,旨在通過提供標準的產生、發送、接收和處理訊息的 API 簡化企業應用的開發,類似於 JDBC 和關係型資料庫通信方式的抽象。”

 

“嗯,很好。下麵的這些概念你也需要特別理解下”:

 

  • Provider:純 Java 語言編寫的 JMS 接口實現(比如 ActiveMQ 就是)
  • Domains:訊息傳遞方式,包括點對點(P2P)、發佈/訂閱(Pub/Sub)兩種
  • Connection factory:客戶端使用連接工廠來創建與 JMS provider 的連接
  • Destination:訊息被尋址、發送以及接收的物件

 

“你來說說這其中 P2P 和 Pub/Sub 的區別吧”,老王給成小胖丟擲了一個問題。

 

成小胖可不是吃素的,畢竟要是吃素的話他也吃不到這麼胖……這些基本概念對他來說都是小事一樁:

 

P2P (點對點)訊息域使用 queue 作為 Destination,訊息可以被同步或異步的發送和接收,每個訊息只會給一個 Consumer 傳送一次。

 

Consumer 可以使用 MessageConsumer.receive() 同步地接收訊息,也可以通過使用MessageConsumer.setMessageListener() 註冊一個 MessageListener 實現異步接收。

 

多個 Consumer 可以註冊到同一個 queue 上,但一個訊息只能被一個 Consumer 所接收,然後由該 Consumer 來確認訊息。並且在這種情況下,Provider 對所有註冊的 Consumer 以輪詢的方式發送訊息。

 

Pub/Sub(發佈/訂閱,Publish/Subscribe)訊息域使用 topic 作為 Destination,發佈者向 topic 發送訊息,訂閱者註冊接收來自 topic 的訊息。發送到 topic 的任何訊息都將自動傳遞給所有訂閱者。接收方式(同步和異步)與 P2P 域相同。

除非顯式指定,否則 topic 不會為訂閱者保留訊息。當然,這可以通過持久化(Durable)訂閱來實現訊息的儲存。這種情況下,當訂閱者與 Provider 斷開時,Provider 會為它儲存訊息。當持久化訂閱者重新連接時,將會受到所有的斷連期間未消費的訊息。

 

“嗯,總結的很不錯,上面的這些知識是學習 ActiveMQ 的理論基礎,是必須要掌握的。”

 

“既然 JMS 是一個通用的規範,那麼使用它創建應用程式肯定也有一個通用的步驟吧?”老王追問道。

 

“有的有的。要不您來說說這個通用步驟?就當我考考您,哈哈!”成小胖故作聰明,自以為老王作為架構師不會關註這些太具體的實現細節。

 

然而老王平日里親力親為,至今還常常擼代碼,怎麼會被這種小 case 所難倒?於是老王分分鐘給出答案:

 

  • 獲取連接工廠
  • 使用連接工廠創建連接
  • 啟動連接
  • 從連接創建會話
  • 獲取 Destination
  • 創建 Producer,或
  1. 創建 Producer
  2. 創建 message
  • 創建 Consumer,或發送或接收message發送或接收 message
  1. 創建 Consumer
  2. 註冊訊息監聽器(可選)
  • 發送或接收 message
  • 關閉資源(connection, session, producer, consumer 等)

 

“66666,厲害啊我的王哥!”成小胖的小聰明被老王擊得粉碎!

 

“你嘴皮子耍夠了吧,還是多動動手吧。現在你手寫上面步驟對應的代碼實現吧”,老王給了成小胖一個眼神,讓他自己慢慢體會……

 

成小胖也不是省油的燈,馬上擦乾凈白板,現場擼了起來(是擼代碼,擼代碼,擼代碼,重要的事情說三遍):

 

public class JMSDemo {
        ConnectionFactory connectionFactory;
        Connection connection;
        Session session;
        Destination destination;
        MessageProducer producer;
        MessageConsumer consumer;
        Message message;
        boolean useTransaction = false;
        try {
                Context ctx = new InitialContext();
                connectionFactory = (ConnectionFactory) ctx.lookup("ConnectionFactoryName");
                //使用ActiveMQ時:connectionFactory = new ActiveMQConnectionFactory(user, password, getOptimizeBrokerUrl(broker));
                connection = connectionFactory.createConnection();
                connection.start();
                session = connection.createSession(useTransaction, Session.AUTO_ACKNOWLEDGE);
                destination = session.createQueue("TEST.QUEUE");
                //生產者發送訊息
                producer = session.createProducer(destination);
                message = session.createTextMessage("this is a test");
 
                //消費者同步接收
                consumer = session.createConsumer(destination);
                message = (TextMessage) consumer.receive(1000);
                System.out.println("Received message: " + message);
                //消費者異步接收
                consumer.setMessageListener(new MessageListener() {
                        @Override
                        public void onMessage(Message message) {
                                if (message != null) {
                                        doMessageEvent(message);
                                }
                        }
                });
        } catch (JMSException e) {
                ...
        } finally {
                producer.close();
                session.close();
                connection.close();
        }
}

 

 

老王滿意的點點頭:“還算不賴哈~ JMS 通用的規範咱們都聊完了,下麵就來聊點 ActiveMQ 更具體點的東西咯。”

 

“好啊好啊。要不我先基於自己的學習講講 ActiveMQ 的儲存,您看看我哪裡講的不對或者遺漏的,可好?”成小胖發揮了他一貫的積極主動的作風,當然內心裡還是想得到老王的贊許。

 

“行,那就開始吧。”

 

ActiveMQ 在 queue 中儲存 Message 時,採用先進先出順序(FIFO)儲存。同一時間一個訊息被分派給單個消費者,且只有當 Message 被消費並確認時,它才能從儲存中刪除。

 

對於持久化訂閱者來說,每個消費者獲得 Message 的副本。為了節省儲存空間,Provider 僅儲存訊息的一個副本。持久化訂閱者維護了指向下一個 Message 的指標,並將其副本分派給消費者。以這種方式實現訊息儲存,因為每個持久化訂閱者可能以不同的速率消費 Message,或者它們可能不是全部同時運行。此外,因每個 Message 可能存在多個消費者,所以在它被成功地傳遞給所有持久化訂閱者之前,不能從儲存中刪除。

 

很好,上面這段知識非常重要。其實我們可以通過表格來更清晰地展示”,老王補充道,併在白板上畫了以下表格:

 

 

成小胖雖然對以上特性做過實踐對比,但是並沒有想到去畫一個表格出來使對比更加清晰易懂。特別是當他看到老王隨時就畫出這個表格時便驚嘆不已,大聲喊道:“老王你太牛了,真是愛死你了!”

 

周圍的同事聽到後,都齊刷刷的往這邊看過來。

 

此情此景,老王也不好意思了:“誒誒誒,說話註意哈,不要讓人覺得我們在搞基。回歸正題,你再說說 ActiveMQ 常用的儲存方式吧。”

 

成小胖羞澀的點點頭,迅速地回歸原態,一五一十地說起來。

 

1.KahaDB

 

ActiveMQ 5.3 版本起的預設儲存方式。KahaDB儲存是一個基於檔案的快速儲存訊息,設計標的是易於使用且盡可能快。它使用基於檔案的訊息資料庫意味著沒有第三方資料庫的先決條件。

 

要啟用 KahaDB 儲存,需要在 activemq.xml 中進行以下配置:

 

"broker"

persistent=“true” useShutdownHook=“false”>

“${activemq.data}/kahadb” journalMaxFileLength=“16mb”/>

 

 

2.AMQ

 

與 KahaDB 儲存一樣,AMQ儲存使用戶能夠快速啟動和運行,因為它不依賴於第三方資料庫。AMQ 訊息儲存庫是可靠持久性和高性能索引的事務日誌組合,當訊息吞吐量是應用程式的主要需求時,該儲存是最佳選擇。但因為它為每個索引使用兩個分開的檔案,並且每個 Destination 都有一個索引,所以當你打算在代理中使用數千個佇列的時候,不應該使用它。

 


                        directory="${activemq.data}/kahadb"
                syncOnWrite="true"
                indexPageSize="16kb"
                indexMaxBinSize="100"
                maxFileLength="10mb" />

 

 

3.JDBC

 

選擇關係型資料庫,通常的原因是企業已經具備了管理關係型資料的專長,但是它在性能上絕對不優於上述訊息儲存實現。事實是,許多企業使用關係資料庫作為儲存,是因為他們更願意充分利用這些資料庫資源。

 


        "test-broker"

persistent=“true” xmlns=“http://activemq.apache.org/schema/core”>

“#mysql-ds”/>

“mysql-ds” class=“org.apache.commons.dbcp.BasicDataSource” destroy-method=“close”>
“driverClassName” value=“com.mysql.jdbc.Driver”/>
“url” value=“jdbc:mysql://localhost/activemq?relaxAutoCommit=true”/>
“username” value=“activemq”/>
“password” value=“activemq”/>
“maxActive” value=“200”/>
“poolPreparedStatements” value=“true”/>

 

 

4.記憶體儲存

 

記憶體訊息儲存器將所有持久訊息儲存在記憶體中。在僅儲存有限數量 Message 的情況下,記憶體訊息儲存會很有用,因為 Message 通常會被快速消耗。在 activema.xml 中將 broker 元素上的 persistent 屬性設置為 false 即可。

 

"test-broker"

persistent=“false” xmlns=“http://activemq.apache.org/schema/core”>

“tcp://localhost:61635″/>

 

 

老王聽完後露出贊許的笑容:“連配置都能寫的這麼詳細,看來你確實是做了不少功課,給你點個贊。”老王終究不會吝嗇自己的贊美,他也明白這些贊美對成小胖意味著什麼。

 

成小胖得到了老王的贊賞,心裡也是吃了蜜一般。

 

沒等成小胖說話,老王拿起筆走到白板前,說:“下麵就根據我在工作中的經歷,給你講講 ActiveMQ 的部署樣式。”

 

1.單例樣式

 

這個就不啰嗦了,略過。

 

2.無共享主從樣式

 

這是最簡單的 Provider 高可用性的方案,主從節點分別儲存 Message。從節點需要配置為連接到主節點,並且需要特殊配置其狀態。

 

所有訊息命令(訊息,確認,訂閱,事務等)都從主節點複製到從節點,這種複製發生在主節點對其接收的任何命令生效之前。並且,當主節點收到持久訊息,會等待從節點完成訊息的處理(通常是持久化到儲存),然後再自己完成訊息的處理(如持久化到儲存)後,再傳回對 Producer 的回執。

 

從節點不啟動任何傳輸,也不能接受任何客戶端或網絡連接,除非主節點失效。當主節點失效後,從節點自動成為主節點,並且開啟傳輸並接受連接。這是,使用 failover 傳輸的客戶端就會連接到該新主節點。

 

Broker 連接配置如下:

 

failover://(tcp://masterhost:61616,tcp://slavehost:61616)?randomize=false

 

 

但是,這種部署樣式有一些限制,

 

  • 主節點只會在從節點連接到主節點時複製其活動狀態,因此當從節點沒有連接上主節點之前,任何主節點處理的 Message 或者訊息確認都會在主節點失效後丟失。不過你可以通過在主節點設置 waitForSlave 來避免,這樣就強制主節點在沒有任何一個從節點連接上的情況下接受連接。
  • 就是主節點只能有一個從節點,並且從節點不允許再有其他從節點。
  • 把正在運行的單例配置成無共享主從,或者配置新的從節點時,你都要停止當前服務,修改配置後再重啟才能生效。

 

在可以接受一些故障停機時間的情況下,可以使用該樣式。

 

從節點配置:

 


        "tcp://remotehost:62001"

userName=“Rob” password=“Davies”/>

 

 

此外,可以配置 shutdownOnMasterFailure 項,表示主節點失效後安全關閉,保證沒有訊息丟失,允許管理員維護一個新的從節點。

 

3.共享儲存主從樣式

 

允許多個代理共享儲存,但任意時刻只有一個是活動的。這種情況下,當主節點失效時,無需人工干預來維護應用的完整性。另外一個好處就是沒有從節點數的限制。

 

有兩種細分樣式:

 

(1)基於資料庫

 

它會獲取一個表上的排它鎖,以確保沒有其他 ActiveMQ 代理可以同時訪問資料庫。其他未獲得鎖的代理則處於輪詢狀態,就會被當做是從節點,不會開啟傳輸也不會接受連接。

 

(2)基於檔案系統

 

需要獲取分佈式共享檔案鎖,linux 系統下推薦用 GFS2。            

 

看到這些乾貨,成小胖欣喜若狂一邊聽一邊記,等老王講完後他還沒記完。而老王則趁機喝了杯鐵觀音潤潤嗓子。

 

在記錄完老王所講的部署樣式後,成小胖也不好意思再讓老王繼續講下去了,畢竟他知道老王常年加班腰間盤突出,不能長時間站著。

 

“王哥您坐著休息下,我再給您講講我所理解的 ActiveMQ 的網絡連接,中不中?”

 

“中。沒事兒,我身體好著呢~”老王知道成小胖擔心他的腰,但他還是那個倔脾氣。成小胖也不敢多耽誤時間,立馬開講。

 

1.代理網絡

 

支持將 ActiveMQ 訊息代理鏈接到不同拓撲,這就是被人們熟知的代理網絡。

 

ActiveMQ 網絡使用儲存和轉發的概念,其中訊息總是儲存在本地代理中,然後通過網絡轉發到另一個代理。            

 

當連接建立後,遠程代理將把包含其所有持久和活動消費者目的地的信息傳遞給本地代理,本地代理根據信息決定遠程代理感興趣的 Message 並將它發送給遠程代理。

 

如果希望網絡是雙向的,您可以使用網絡連接器將遠程代理配置為指向本地代理,或將網絡連接器配置為雙工,以便雙向發送訊息。

 


        "static://(tcp://backoffice:61617)"

name=“bridge”
duplex=“true”
conduitSubscriptions=“true”
decreaseNetworkConsumerPriority=“false”>

 

 

註意,配置的順序很重要:

 

  1. 網絡連接——需要在訊息儲存前建立好連接,對應 networkConnectors 元素
  2. 訊息儲存——需要在傳輸前配置好,對應 persistenceAdapter 元素
  3. 訊息傳輸——最後配置,對應 transportConnectors 元素

 

2.網絡發現

 

(1)動態發現

 

使用多播來支持網絡動態發現。配置如下:

 


    "multicast://default"

/>

 

 

其中,multicast:// 中的預設名稱表示該代理所屬的組。因此使用此方式時,強烈推薦你使用一個獨特的組名,避免你的代理連接到其他不相關代理。

 

(2)靜態發現

 

靜態發現接受代理 URI 串列,並將嘗試按串列中確定的順序連接到遠程代理。

 


    "static:(tcp://remote-master:61617,tcp://remote-slave:61617)"

/>

 

 

相關配置如下:

 

  • initialReconnectDelay:預設值1000,表示嘗試連接前的時延。
  • maxReconnectDelay:預設值30000,表示連接失敗後到重新建立連接之間的時延,僅在 useExponentialBackOff 啟用時生效。
  • useExponentialBackOff:預設值 true,如果啟用,表示每次失敗後增加重建連接的時延。
  • backOffMultiplier:預設值2,表示啟用 useExponentialBackOff 後每次的時延增量需要註意的是,網絡連接將始終嘗試建立到遠程代理的連接。

 

需要註意的是,網絡連接將始終嘗試建立到遠程代理的連接。

 

(3)多連接場景           

 

當網絡負載高時,使用多連接很有意義。但是你需要確保不會重覆傳遞訊息,這可以通過過濾器來實現。

 


    "static://(tcp://remotehost:61617)"

name=“queues_only”
duplex=“true”

“>”/>

“static://(tcp://remotehost:61617)”
name=“topics_only”
duplex=“true”

<queue physicalName=“>”/>

 

 

講完後成小胖如釋重負,因為上面這些知識點雖然看起來很少,但他卻花了很多時間看了很多英文資料,同時反覆實踐才理解透的呢。

 

在成小胖講的這段時間,老王一直坐在轉椅上,這會兒他的腰也舒服了很多。老王站起來拍了拍成小胖的肩膀:“這個知識點雖然理解起來有點晦澀,但是你解釋得還是挺不錯的。通過今天的交流,可以看出你對 ActiveMQ 的基礎知識有了不錯的掌握,今後呢還是要多加深入實踐,這樣才能使用它提供高質量的服務。”

 

老王的手機突然響了,是要去開會了:“今天就到這兒吧,我有個會議要參加。”

 

“好,謝謝王哥的耐心指導。希望你多註意身體。”

 

老王鬼魅的一笑:“嗯,放心吧,隔壁有對年輕人剛結婚,聽Ta們說最近想要個小baby,所以我肯定會保養好自己的身體的。”

 

成小胖:“……”

 

“哈哈,開玩笑的!”

 

“……”

    赞(0)

    分享創造快樂