一、什么是消息中間件

消息中間件顧名思義實(shí)現(xiàn)的就是在兩個(gè)系統(tǒng)或兩個(gè)客戶端之間進(jìn)行消息傳送

二、什么是ActiveMQ

ActiveMQ是一種開(kāi)源的基于JMS（Java Message Servie）規(guī)范的一種消息中間件的實(shí)現(xiàn)，ActiveMQ的設(shè)計(jì)目標(biāo)是提供標(biāo)準(zhǔn)的，面向消息的，能夠跨越多語(yǔ)言和多系統(tǒng)的應(yīng)用集成消息通信中間件。

三、什么時(shí)候需要用ActiveMQ

ActiveMQ常被應(yīng)用與系統(tǒng)業(yè)務(wù)的解耦，異步消息的推送，增加系統(tǒng)并發(fā)量，提高用戶體驗(yàn)。例如以我在工作中的使用，在比較耗時(shí)且異步的遠(yuǎn)程開(kāi)鎖操作時(shí)

四、如何使用ActiveMQ

1.AcitveMQ的數(shù)據(jù)傳送流程

2.ActiveMQ的兩種消息傳遞類型

（1）點(diǎn)對(duì)點(diǎn)傳輸，即一個(gè)生產(chǎn)者對(duì)應(yīng)一個(gè)消費(fèi)者，生產(chǎn)者向broke推送數(shù)據(jù)，數(shù)據(jù)存儲(chǔ)在broke的一個(gè)隊(duì)列中，當(dāng)消費(fèi)者接受該條隊(duì)列里的數(shù)據(jù)。

（2）基于發(fā)布/訂閱模式的傳輸，即根據(jù)訂閱話題來(lái)接收相應(yīng)數(shù)據(jù)，一個(gè)生產(chǎn)者可向多個(gè)消費(fèi)者推送數(shù)據(jù)，與MQTT協(xié)議的實(shí)現(xiàn)是類似的，對(duì)MQTT協(xié)議有興趣的可跳轉(zhuǎn)文末查看

兩種消息傳遞類型的不同，點(diǎn)對(duì)點(diǎn)傳輸消費(fèi)者可以接收到在連接之前生產(chǎn)者所推送的數(shù)據(jù)，而基于發(fā)布/訂閱模式的傳輸方式消費(fèi)者只能接收到連接之后生產(chǎn)者推送的數(shù)據(jù)。

3.ActiveMQ的安裝與啟動(dòng)

（1）官網(wǎng)下載對(duì)應(yīng)服務(wù)器版本

（2）解壓后進(jìn)入apache-activemq-5.15.9/bin目錄

（3）執(zhí)行./activemq start啟動(dòng)ActiveMQ

(4)瀏覽器輸入ActiveMQ啟動(dòng)的服務(wù)器ip:8161便可進(jìn)入web界面，點(diǎn)擊Manage ActiveMQ broker可以查看消息推送的狀態(tài)，默認(rèn)賬號(hào)密碼為admin,admin

（5）啟動(dòng)錯(cuò)誤分析

進(jìn)入/root/apache-activemq-5.15.9/data目錄查看activemq.log文件，根據(jù)錯(cuò)誤提示信息修改，例如端口號(hào)被占用等。

4.ActiveMQ的代碼測(cè)試

（1）構(gòu)建maven項(xiàng)目，引入依賴

<dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.9.0</version>
        </dependency>

（2）生產(chǎn)者類

/**
 * @Description 生產(chǎn)者
 * @Date 2019/7/20
 * @Created by yqh
 */
public class MyProducer {

    private static final String ACTIVEMQ_URL = "tcp://192.168.168.242:61616";

    public static void main(String[] args) throws JMSException {
        // 創(chuàng)建連接工廠
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 創(chuàng)建連接
        Connection connection = activeMQConnectionFactory.createConnection();
        // 打開(kāi)連接
        connection.start();
        // 創(chuàng)建會(huì)話
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 創(chuàng)建隊(duì)列目標(biāo),并標(biāo)識(shí)隊(duì)列名稱，消費(fèi)者根據(jù)隊(duì)列名稱接收數(shù)據(jù)
        Destination destination = session.createQueue("myQueue");
        // 創(chuàng)建一個(gè)生產(chǎn)者
        MessageProducer producer = session.createProducer(destination);
        // 向隊(duì)列推送10個(gè)文本消息數(shù)據(jù)
        for (int i = 1 ; i <= 10 ; i++){
            // 創(chuàng)建文本消息
            TextMessage message = session.createTextMessage("第" + i + "個(gè)文本消息");
            //發(fā)送消息
            producer.send(message);
            //在本地打印消息
            System.out.println("已發(fā)送的消息：" + message.getText());
        }
        //關(guān)閉連接
        connection.close();
    }

}

運(yùn)行結(jié)果：

已發(fā)送的消息：第1個(gè)文本消息

已發(fā)送的消息：第2個(gè)文本消息

已發(fā)送的消息：第3個(gè)文本消息

已發(fā)送的消息：第4個(gè)文本消息

已發(fā)送的消息：第5個(gè)文本消息

已發(fā)送的消息：第6個(gè)文本消息

已發(fā)送的消息：第7個(gè)文本消息

已發(fā)送的消息：第8個(gè)文本消息

已發(fā)送的消息：第9個(gè)文本消息

已發(fā)送的消息：第10個(gè)文本消息

測(cè)試查看web后臺(tái)顯示，有10條消息在隊(duì)列中等待消費(fèi)

（3）消費(fèi)者類

/**
 * @Description 消費(fèi)者類
 * @Date 2019/7/20 0020
 * @Created by yqh
 */
public class MyConsumer {

    private static final String ACTIVEMQ_URL = "tcp://192.168.168.242:61616";

    public static void main(String[] args) throws JMSException {
        // 創(chuàng)建連接工廠
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 創(chuàng)建連接
        Connection connection = activeMQConnectionFactory.createConnection();
        // 打開(kāi)連接
        connection.start();
        // 創(chuàng)建會(huì)話
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 創(chuàng)建隊(duì)列目標(biāo),并標(biāo)識(shí)隊(duì)列名稱，消費(fèi)者根據(jù)隊(duì)列名稱接收數(shù)據(jù)
        Destination destination = session.createQueue("myQueue");
        // 創(chuàng)建消費(fèi)者
        MessageConsumer consumer = session.createConsumer(destination);
        // 創(chuàng)建消費(fèi)的監(jiān)聽(tīng)
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("消費(fèi)的消息：" + textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

測(cè)試結(jié)果：

消費(fèi)的消息：第1個(gè)文本消息

消費(fèi)的消息：第2個(gè)文本消息

消費(fèi)的消息：第3個(gè)文本消息

消費(fèi)的消息：第4個(gè)文本消息

消費(fèi)的消息：第5個(gè)文本消息

消費(fèi)的消息：第6個(gè)文本消息

消費(fèi)的消息：第7個(gè)文本消息

消費(fèi)的消息：第8個(gè)文本消息

消費(fèi)的消息：第9個(gè)文本消息

消費(fèi)的消息：第10個(gè)文本消息

web后臺(tái)顯示有一個(gè)消費(fèi)者處于連接狀態(tài)，且已消費(fèi)了10個(gè)message，而該條隊(duì)列已沒(méi)有message待消費(fèi)了

（4）當(dāng)我們運(yùn)行兩個(gè)消費(fèi)者類，消息又是怎么被消費(fèi)的呢？是兩個(gè)消費(fèi)者都能收到生產(chǎn)者生產(chǎn)的message，還是只有其中一個(gè)消費(fèi)者能消費(fèi)呢？

我們先運(yùn)行兩個(gè)消費(fèi)者，在運(yùn)行一個(gè)生產(chǎn)者對(duì)目標(biāo)隊(duì)列生產(chǎn)10個(gè)message，會(huì)發(fā)現(xiàn)有以下情況

// Consumer1控制臺(tái)

消費(fèi)的消息：第1個(gè)文本消息

消費(fèi)的消息：第3個(gè)文本消息

消費(fèi)的消息：第5個(gè)文本消息

消費(fèi)的消息：第7個(gè)文本消息

消費(fèi)的消息：第9個(gè)文本消息

// Consumer2控制臺(tái)

消費(fèi)的消息：第2個(gè)文本消息

消費(fèi)的消息：第4個(gè)文本消息

消費(fèi)的消息：第6個(gè)文本消息

消費(fèi)的消息：第8個(gè)文本消息

消費(fèi)的消息：第10個(gè)文本消息

即隊(duì)列中的數(shù)據(jù)會(huì)平均的分給每一個(gè)消費(fèi)者消費(fèi)，且每一條數(shù)據(jù)只能被消費(fèi)一次

（5）以上是基于隊(duì)列點(diǎn)對(duì)點(diǎn)的傳輸類型，以下是基于發(fā)布/訂閱模式傳輸?shù)念愋蜏y(cè)試

/**
 * @Description 基于發(fā)布/訂閱模式傳輸類型的生產(chǎn)者測(cè)試
 * @Date 2019/7/20 0020
 * @Created by yqh
 */
public class MyProducerForTopic {

    private static final String ACTIVEMQ_URL = "tcp://192.168.168.242:61616";

    public static void main(String[] args) throws JMSException {
        // 創(chuàng)建連接工廠
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 創(chuàng)建連接
        Connection connection = activeMQConnectionFactory.createConnection();
        // 打開(kāi)連接
        connection.start();
        // 創(chuàng)建會(huì)話
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 創(chuàng)建隊(duì)列目標(biāo),并標(biāo)識(shí)隊(duì)列名稱，消費(fèi)者根據(jù)隊(duì)列名稱接收數(shù)據(jù)
        Destination destination = session.createTopic("topicTest");
        // 創(chuàng)建一個(gè)生產(chǎn)者
        MessageProducer producer = session.createProducer(destination);
        // 向隊(duì)列推送10個(gè)文本消息數(shù)據(jù)
        for (int i = 1 ; i <= 10 ; i++){
            // 創(chuàng)建文本消息
            TextMessage message = session.createTextMessage("第" + i + "個(gè)文本消息");
            //發(fā)送消息
            producer.send(message);
            //在本地打印消息
            System.out.println("已發(fā)送的消息：" + message.getText());
        }
        //關(guān)閉連接
        connection.close();
    }

}

/**
 * @Description 基于發(fā)布/訂閱模式傳輸類型的消費(fèi)者測(cè)試
 * @Date 2019/7/20 0020
 * @Created by yqh
 */
public class MyConsumerForTopic {

    private static final String ACTIVEMQ_URL = "tcp://192.168.168.242:61616";

    public static void main(String[] args) throws JMSException {
        // 創(chuàng)建連接工廠
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 創(chuàng)建連接
        Connection connection = activeMQConnectionFactory.createConnection();
        // 打開(kāi)連接
        connection.start();
        // 創(chuàng)建會(huì)話
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 創(chuàng)建隊(duì)列目標(biāo),并標(biāo)識(shí)隊(duì)列名稱，消費(fèi)者根據(jù)隊(duì)列名稱接收數(shù)據(jù)
        Destination destination = session.createTopic("topicTest");
        // 創(chuàng)建消費(fèi)者
        MessageConsumer consumer = session.createConsumer(destination);
        // 創(chuàng)建消費(fèi)的監(jiān)聽(tīng)
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("消費(fèi)的消息：" + textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

現(xiàn)在如果我們先啟動(dòng)生產(chǎn)者，再啟動(dòng)消費(fèi)者，會(huì)發(fā)現(xiàn)消費(fèi)者是無(wú)法接收到之前生產(chǎn)者之前所生產(chǎn)的數(shù)據(jù)，只有消費(fèi)者先啟動(dòng)，再讓生產(chǎn)者消費(fèi)才可以正常接收數(shù)據(jù)，這也是發(fā)布/訂閱的主題模式與點(diǎn)對(duì)點(diǎn)的隊(duì)列模式的一個(gè)明顯區(qū)別。

而如果啟動(dòng)兩個(gè)消費(fèi)者，那么每一個(gè)消費(fèi)者都能完整的接收到生產(chǎn)者生產(chǎn)的數(shù)據(jù)，即每一條數(shù)據(jù)都被消費(fèi)了兩次，這是發(fā)布/訂閱的主題模式與點(diǎn)對(duì)點(diǎn)的隊(duì)列模式的另一個(gè)明顯區(qū)別。

淺談MQTT

1、什么是MQTT

MQTT的全稱是“ Message Queuing Telemetry Transport”，即消息隊(duì)列遙測(cè)傳輸，是一種基于訂閱/發(fā)布模式的應(yīng)用層協(xié)議，而http是一種基于restful風(fēng)格的一種應(yīng)用層協(xié)議。

MQTT協(xié)議是一種輕量級(jí)協(xié)議，作為一種低開(kāi)銷、低帶寬占用的即時(shí)通訊協(xié)議，常被應(yīng)用于物聯(lián)網(wǎng)項(xiàng)目。同樣基于訂閱/發(fā)布模式的中間件有ActiveMQ，Kafka等消息中間件，歸根結(jié)底實(shí)現(xiàn)的都是消息的傳輸。

2、如何理解MQTT

MQTT的是一種應(yīng)用層協(xié)議，每一種協(xié)議都有其適用場(chǎng)景，而MQTT常被應(yīng)用于消息推送，消息采集。例如溫度檢測(cè)儀器定時(shí)上傳溫度、檢測(cè)礦洞氧氣濃度等。

MQTT是基于TCP/IP的一種應(yīng)用層協(xié)議，TCP/IP本身已實(shí)現(xiàn)了在不可靠的網(wǎng)絡(luò)環(huán)境提供可靠的網(wǎng)絡(luò)傳輸?shù)墓δ?，而MQTT協(xié)議也有其保障消息可靠傳輸?shù)牟呗浴?/p>

MQTT推送的消息有三種消息質(zhì)量

1.至多一次，即消息只推送一次，至于消息有沒(méi)有推送成功

2.至少一次，需要確認(rèn)消息到達(dá)，可能會(huì)導(dǎo)致收到重復(fù)數(shù)據(jù)(注：MQTT定義的重發(fā)機(jī)制與tcp的重復(fù)機(jī)制是不同的，tcp的重復(fù)機(jī)制是在限定時(shí)間內(nèi)如果沒(méi)有收到對(duì)應(yīng)序號(hào)的響應(yīng)報(bào)文，則會(huì)重新推送該序列號(hào)對(duì)應(yīng)的報(bào)文，而MQTT的重發(fā)機(jī)制是在客戶端重新建立連接時(shí)，

補(bǔ)發(fā)之前沒(méi)有對(duì)應(yīng)響應(yīng)報(bào)文的數(shù)據(jù)包，當(dāng)然客戶端可以選擇是否要接收這些之前沒(méi)有傳輸成功的數(shù)據(jù)包。最開(kāi)始使用netty實(shí)現(xiàn)MQTT服務(wù)器的時(shí)候就理解錯(cuò)了，以為MQTT的重復(fù)機(jī)制與tcp的重復(fù)機(jī)制一樣)

3.只有一次，確認(rèn)消息只到達(dá)一次，常用于對(duì)數(shù)據(jù)要求嚴(yán)格的場(chǎng)景，例如計(jì)費(fèi)場(chǎng)景，訂單場(chǎng)景

3、如何使用MQTT

MQTT的客戶端和服務(wù)端目前已有成熟的開(kāi)源產(chǎn)品，例如服務(wù)端有emqx，客戶端有Eclipse Paho Mqtt(Java)，都可以方面的引入相應(yīng)的庫(kù)快速的實(shí)現(xiàn)推送功能（具體可根據(jù)需求查看對(duì)應(yīng)的API）。

本質(zhì)上來(lái)將是客戶端與服務(wù)端建立一個(gè)Socket，然后根據(jù)MQTT協(xié)議規(guī)定發(fā)送響應(yīng)的報(bào)，例如建立socket后發(fā)送connet報(bào)文去建立連接，然后服務(wù)器會(huì)解析該連接報(bào)文，并保存該連接的相關(guān)信息。

我們可以把MQTT協(xié)議的規(guī)定當(dāng)成是我們實(shí)現(xiàn)web項(xiàng)目中所實(shí)現(xiàn)的業(yè)務(wù)邏輯。

4、MQTT協(xié)議的相關(guān)的名詞解析

1.訂閱（Subscription）

訂閱包含主題篩選器（Topic Filter）和最大服務(wù)質(zhì)量（QoS）。訂閱會(huì)與一個(gè)會(huì)話（Session）關(guān)聯(lián)。一個(gè)會(huì)話可以包含多個(gè)訂閱。每一個(gè)會(huì)話中的每個(gè)訂閱都有一個(gè)不同的主題篩選器。

2.、會(huì)話（Session）

每個(gè)客戶端與服務(wù)器建立連接后就是一個(gè)會(huì)話，客戶端和服務(wù)器之間有狀態(tài)交互。會(huì)話存在于一個(gè)網(wǎng)絡(luò)之間，也可能在客戶端和服務(wù)器之間跨越多個(gè)連續(xù)的網(wǎng)絡(luò)連接。

3.主題名（Topic Name）

連接到一個(gè)應(yīng)用程序消息的標(biāo)簽，該標(biāo)簽與服務(wù)器的訂閱相匹配。服務(wù)器會(huì)將消息發(fā)送給訂閱所匹配標(biāo)簽的每個(gè)客戶端。

4.主題篩選器（Topic Filter）

一個(gè)對(duì)主題名通配符篩選器，在訂閱表達(dá)式中使用，表示訂閱所匹配到的多個(gè)主題。

5.負(fù)載（Payload）

消息訂閱者所具體接收的內(nèi)容。

總結(jié)

到此這篇關(guān)于ActiveMQ的簡(jiǎn)單入門(mén)介紹與使用的文章就介紹到這了,更多相關(guān)ActiveMQ介紹與使用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論