Java SE基礎鞏固（十四）：JMS

1 概述

JMS即Java Message Service（Java消息服務），它是一組規範，提供了方便的而且通用的方式來讓Java程序建立，發送，接受和讀取消息。

在中大型的項目中，不少都用到了被叫作面向消息的中間件（Message Oriented Middleware），一般簡稱消息中間件，如今流行的消息中間件產品主要有ActiveMQ，RabbitMQ，Kafka，RocketMQ等等，不少大型的商業軟件使用這些產品構建應用程序的消息系統，消息在應用程序中的各個模塊（或者服務）之間傳遞，以此達到解耦，削峯，提升應用程序的吞吐量等目的。

由於產品衆多，若是沒有一個成熟的規範來約束，那麼各個產商消息中間件的具體實現可能會截然不同，使得開發者在切換產品的時候很是麻煩，不得不從新編寫大量代碼來適應產品。JMS就是這麼一個規範，它定義了一個消息中間件產品應該具備哪些組件，哪些接口等等，若是消息中間件都能根據規範定義的接口來實現的話，開發者在構建消息系統的時候只須要付出不多的代價來切換不一樣的產品，由於接口都是同樣的，幾乎不須要如何修改代碼。

JMS規範定義的幾個組件以下所示：

• JMS Provider。提供消息服務的組件，用來爲JMS客戶端提供服務，JMS客戶端能夠有生產者客戶端和消費者客戶端等。
• JMS Message。即消息傳遞實體。
• JMS Domains。即消息傳遞的目的地。

JMS Message包含了以下幾個部分：

• Header。全部的消息都有相同的Header字段集合，相似HTTP Header，這些字段用於在客戶端和服務端（provider）之間作路由或者身份認證等，包括JMSDestination，JMSMessageID，JMSTimestamp等。
• Properties。即屬性，是可選項，一個屬性由一個鍵值對構成，鍵是屬性名，值是屬性的值。應用程序能夠自定義屬性，並將其附在Message上和Header以及Body一塊兒發送出去。
• Body，即消息體主體，例如對於普通的文本消息，消息主題就是本文內容，例如「Hello」字符串。

以上是對JMS的簡單介紹，更多關於JMS規範的內容，建議參看JSR 914文檔（直接在搜索引擎中搜索該關鍵字就能夠找到了），下面我將介紹JMS規範定義的接口並使用ActiveMQ來構建一個小Demo做爲實踐，最後會介紹兩種常見消息傳送模型：點對點傳輸模型和發佈/訂閱傳輸模型。

注意，這裏所說的消息不是狹義的消息，即不特指一句話或者一個郵件等，消息不只能夠是字符串形式的，還能夠是一個Java對象等。

2 JMS接口

上圖是JMS規範定義的接口，最左側是JMS通用接口，右側是兩種不一樣的消息傳送模型的接口，下面來介紹通用接口：

• ConnectionFactory，即鏈接工廠，Connection須要經過ConnectionFactory來建立。
• Connection，即鏈接，表示到JMS Provider的鏈接。
• Destination，即目的地，表示消息發送的終點，例如在發郵件的時候接受者的郵箱。
• Session，即會話，用於發送和接受消息的一個單線程上下文。
• MessageProducer，即消息生產者，消息生產者是消息的發送方，負責建立和發送消息都JMS Provider。
• MessageConsumer，即消息消費者，消息消費者是消息的接收方，負責從JMS Provider接收和讀取（消費）消息。

下圖是上述幾個接口的關係圖：

下面會看到，咱們在使用ActiveMQ編寫代碼的時候幾乎就是根據這麼一個關係圖來寫的。

3 使用ActiveMQ

標準的Java SE裏沒有實現JMS規範，因此標準JDK裏沒有上述提到的幾個接口和對應的實現，但Java EE裏有實現，不過搭建Java EE相對比較麻煩（只是一個Demo而已），因此這裏就直接使用ActiveMQ這個產品。ActiveMQ實現了JMS的規範，提供了JMS規範定義的接口，之因此使用ActiveMQ而不是其餘的產品，是由於ActiveMQ的實現比較「標準」，很是貼合JMS規範，且學習門檻叫其餘幾個產品低。

3.1 安裝並運行ActiveMQ

安裝很簡單，直接到ActiveMQ官網去下載便可，我這裏使用的是5.15.6這個版本，在windows下，下載apache-activemq-5.15.6-bin.zip這個壓縮包（在Unix系統下，下載apache-activemq-5.15.6-bin.tar.gz），而後解壓，在命令行裏進入bin目錄，輸入activemq start便可。

建議到官網的Get Start部分看看，安裝的啓動都講得很詳細，這裏我就很少說了。

啓動以後，能夠在瀏覽器輸入http://localhost:8161來查看ActiveMQ的各項數據，以下所示：

點擊Manage ActiveMQ broker便可進入控制檯，帳號和密碼默認都是admin，進入以後以下所示：

在這裏，能夠選擇各個標籤欄來查看不一樣的信息，例如Topic，Queue等，具體在這就很少說了，讀者能夠自行嘗試。

3.2 導入ActiveMQ依賴

若是使用的是Maven來構建項目，能夠加入以下依賴：

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.15.6</version>
</dependency>
複製代碼

若是沒有使用Maven，就須要本身往項目裏導入jar包了，在剛剛下載的壓縮包裏，包含了activemq-all-5.15.6.jar這個jar包，將其導入到項目的類路徑便可。

3.3 編寫Demo

準備工做作完以後，就能夠正式編寫代碼了，演示代碼以下所示：

JMSProducer

package top.yeonon.jms;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/** * @Author yeonon * @date 2018/10/20 0020 14:09 **/
public class JMSProducer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    private static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    public static void main(String[] args) throws JMSException {
        //鏈接工廠
        ConnectionFactory connectionFactory;
        //鏈接
        Connection connection;
        //Session會話
        Session session;
        //目的地
        Destination destination;

        //建立鏈接工廠實例
        connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD, BROKER_URL);
        //從鏈接工廠實例中建立鏈接
        connection = connectionFactory.createConnection();
        //從鏈接中建立Session，第一個參數是是否開啓事務，這裏選擇不開啓，傳入false
        // 第二個參數是消息確認模式，這裏選擇的是自動確認
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //建立了一個隊列，隊列Queue（javax.jms包裏的Queue）繼承了Destination接口
        destination = session.createQueue("Hello");
        //建立一個消息生產者，須要傳入一個destination，這樣這個producer就和destination綁定了
        MessageProducer producer = session.createProducer(destination);

        //配置完成以後，啓動鏈接
        connection.start();

        //發送消息
        for (int i = 0; i < 10; i++) {
            TextMessage message = session.createTextMessage("message " + i);
            System.out.println("producer send message : " + message.getText());
            producer.send(message);
        }

        //發送完成後，關閉鏈接
        connection.close();
    }
}

複製代碼

JMSConsumer

package top.yeonon.jms;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/** * @Author yeonon * @date 2018/10/20 0020 14:19 **/
public class JMSConsumer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    private static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    public static void main(String[] args) throws JMSException {
        //鏈接工廠
        ConnectionFactory connectionFactory;
        //鏈接
        Connection connection;
        //Session會話
        Session session;
        //目的地
        Destination destination;

        //建立鏈接工廠實例
        connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD, BROKER_URL);
        //從鏈接工廠實例中建立鏈接
        connection = connectionFactory.createConnection();
        //從鏈接中建立Session，第一個參數是是否開啓事務，這裏選擇不開啓，傳入false
        // 第二個參數是消息確認模式，這裏選擇的是自動確認
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //建立了一個隊列，隊列Queue（javax.jms包裏的Queue）繼承了Destination接口
        destination = session.createQueue("Hello");
        //建立一個消息生產者，須要傳入一個destination，這樣這個producer就和destination綁定了
        MessageConsumer consumer = session.createConsumer(destination);

        //配置完成以後，啓動鏈接
        connection.start();

        //接受消息
        while (true) {
            TextMessage message = (TextMessage) consumer.receive(10000);
            if (message != null)
                System.out.println("receive message : " + message.getText());
        }
    }
}
複製代碼

代碼中註釋寫的很清楚了，就只說說一個讓人感受疑惑的Destination，在演示代碼中，使用了Queue來表示destination，初次接觸消息隊列的話，可能會以爲有些奇怪，一個Queue怎麼就是一個目的地了？其實這是徹底能夠的，由於目的地並不就必定是ip地址或者郵箱之類的東西，只要是一個能夠接受消息的的載體便可，如今大多數的消息中間件都會或多或少只用隊列來做爲消息的載體，因此有時候直接把消息中間件簡稱爲消息隊列（Message Queue）。

先運行consumer，稍微等待一會，再運行producer，producer運行完畢以後能夠在producer的控制檯看到相似以下輸出：

producer send message : message 0
producer send message : message 1
producer send message : message 2
producer send message : message 3
producer send message : message 4
producer send message : message 5
producer send message : message 6
producer send message : message 7
producer send message : message 8
producer send message : message 9
複製代碼

而後去consumer的控制檯看看，發現有相似以下輸出：

receive message : message 0
receive message : message 1
receive message : message 2
receive message : message 3
receive message : message 4
receive message : message 5
receive message : message 6
receive message : message 7
receive message : message 8
receive message : message 9
複製代碼

若是能看到如上輸出，說明生產者已經順利將消息發送到JMS provider（ActiveMQ的服務）了，而且消費者也已經順利的從JMS provider裏取出消息並進行讀取，處理了。若是你開啓了多個consumer，還能看到producer發送的消息被多個consumer共同消費，例若有兩個consumer，可能會有以下輸出：

consumer1:

receive message : message 0
receive message : message 2
receive message : message 4
receive message : message 6
receive message : message 8
複製代碼

consumer2:

receive message : message 1
receive message : message 3
receive message : message 5
receive message : message 7
receive message : message 9
複製代碼

有點負載均衡的意思，這是由於如今有兩個consumer共享一個隊列Hello，而後輪流從隊列Hello中取出消息並消費（這是ActiveMQ的默認方式）。

4 消息傳輸模型

JMS規範定義了兩種消息傳輸模型：點對點傳輸模型和發佈/訂閱傳輸模型。

4.1 點對點傳輸模型

在點對點模型下，消息由生產者發送到隊列裏，隊列再將消息發送給消費者。在這個過程當中，消息的目的地是隊列，消費者監聽隊列，當隊列中有消息的時候，隊列就會把消息發送給監聽了隊列的消費者。能夠簡單的把隊列看作是一箇中轉站，負責接受生產者發送過來的消息，而後將消息再發送給消費者。

一個隊列能夠有多個生產者和消費者，可是一個消息只能被一個消費者接受並消費，JMS provider將根據「先來先服務」的原則肯定消息該發送到哪一個消費者中。若是沒有任何一個消費者在監聽隊列，那麼消息再會在隊列裏堆積，可堆積的消息多少由隊列的大小以及JMS provider的具體實現決定，若是超過了規定的大小，多餘的消息可能會被拋棄，也可能會被存儲到持久化存儲器中（這取決於具體實現）。

4.2 發佈/訂閱傳輸模型

在發佈/訂閱模型下，消息的目的地再也不是簡單的隊列，而是一個Topic（主題），消息由生產者發送到Topic中，而後再傳遞到訂閱了該主題的消費者中。一個Topic也能夠綁定多個生產者和多個消費者，和點對點模型不一樣的是，一個消息能夠被多個消費者消費，即假設有兩個消費者同時訂閱了一個Topic，那麼當有消息發往這個Topic的時候，Topic會將消息的兩個拷貝發送到兩個消費者中。

值得注意的是，這裏的Topic實際上是一個抽象的概念，其具體的形式也能夠是一個隊列或者多個隊列組成的隊列組（這取決於具體實現），只是在隊列上再打上了一個Topic標籤，生產者和消費者的目標再也不是簡單的某個隊列，而是一個Topic。舉個例子，如今咱們定義了一個名爲hello的Topic，生產者表示要將消息發送到這個hello Topic中，消費者也對這個hello Topic感興趣（即訂閱Topic），而這個Topic的底層是一個隊列，當生產者發送消息的時候，底層其實就是發送到這個打上了hello topic標籤的隊列裏，而後再把消息拷貝被傳遞到對hello topic感興趣的消費者中。

這裏的解釋可能有些不那麼合理，但願讀者能理解。不理解也沒事，下面我會寫一個發佈/訂閱模式的演示代碼，方便讀者理解這個模型。

在第3節中編寫的Demo其實就是點對點模型，從點貴點模型切換到發佈/訂閱模型很是簡單，對於上面的那個Demo，只須要修改兩行代碼便可：

//點對點模型中的producer
destination = session.createQueue("Hello");
//發佈/訂閱模型中的producer
destination = session.createTopic("Hello");


//點對點模型中的consumer
destination = session.createQueue("Hello");
//發佈/訂閱模型中的consumer
destination = session.createTopic("Hello");
複製代碼

修改完畢以後，運行兩個consumer，一個producer，而後看看控制檯的輸出。

producer的輸出大體以下所示：

producer send message : message 0
producer send message : message 1
producer send message : message 2
producer send message : message 3
producer send message : message 4
producer send message : message 5
producer send message : message 6
producer send message : message 7
producer send message : message 8
producer send message : message 9
複製代碼

consumer1:

receive message : message 0
receive message : message 1
receive message : message 2
receive message : message 3
receive message : message 4
receive message : message 5
receive message : message 6
receive message : message 7
receive message : message 8
receive message : message 9
複製代碼

consumer2:

receive message : message 0
receive message : message 1
receive message : message 2
receive message : message 3
receive message : message 4
receive message : message 5
receive message : message 6
receive message : message 7
receive message : message 8
receive message : message 9
複製代碼

發現consumer1和consumer2都收到了producer發佈的10個消息，總共20個消息。這和點對點模型中不同，點對點模型中不管有多少個消費者，總共都只能接受10個消息。

5 小結

本文簡單介紹了JMS規範的一些組件，接口，瞭解JMS能夠加深對各類消息中間件產品的理解。以後介紹了JMS定義的兩種消息傳輸模型：點對點模型和發佈/訂閱模型。二者最大的區別是：點對點模型中，不管有多少個消費者，一個消息只能被一個消費者接受並消費，發佈/訂閱模型中，一個消息能夠被多個消費者消費。文中還提供了一個Demo，讓你們直觀的感覺到這兩種模型的區別，加深理解。

6 參考資料

JSR 914