前言

隨著互聯(lián)網(wǎng)開發(fā)模式的不斷普及，同時也伴隨著移動互聯(lián)網(wǎng)的不斷發(fā)展，給用戶發(fā)送消息成為每一款應(yīng)用都必不可少的組件之一，想想你的淘寶、微信、釘釘，你日常生活中肯定收到過很多很多的消息。當然，這也是為了提高用戶體驗，及時告知用戶業(yè)務(wù)辦理結(jié)果，給用戶推送消息這樣的功能需求自然也是必不可少（但是現(xiàn)在垃圾信息確實太多了）。正是在這樣的背景之下，消息隊列這樣的組件應(yīng)運而生，這樣的組件不僅滿足多端消息發(fā)送的需求，同時還可以作為服務(wù)器削峰去谷，降低系統(tǒng)壓力。今天我們就來簡單了解一款廣泛應(yīng)用的消息隊列組件——ActiveMQ。

ActiveMQ

ActiveMq什么

百度百科的介紹很簡單，但也說明了activeMq的特性：

Apache ActiveMQ是Apache軟件基金會所研發(fā)的開放源代碼消息中間件；由于ActiveMQ是一個純Java程序，因此只要操作系統(tǒng)支持Java虛擬機，ActiveMQ便可執(zhí)行。

• 支持Java消息服務(wù)(JMS) 1.1 版本
• [Spring Framework](https://baike.baidu.com/item/Spring Framework)
• 集群 (Clustering)
• 支持的編程語言包括：C、C++、C#、Delphi、Erlang、[Adobe Flash](https://baike.baidu.com/item/Adobe Flash)、Haskell、Java、JavaScript、Perl、PHP、Pike、Python和Ruby
• 協(xié)議支持包括：OpenWire、REST、STOMP、WS-Notification、MQTT、XMPP以及AMQP [1]

當然，作為一個開發(fā)組件，對我們而言最重要的就是如何使用它，至于它的特性以及底層的原理，不再我們目前的討論范圍，下來我們來看下如何安裝、部署它。

安裝

這一塊的內(nèi)容就太簡單了，但由于這是一篇科普類文字，所以我就稍微羅嗦一點。但是我個人覺得，只要你個人動手能力不是特別弱雞，那你大概看下官方文檔，應(yīng)該都可以安裝好，而且現(xiàn)在的組件都大同小異，不用看文檔你也能大概部署起來，無非就是找到bin文件夾，找到啟動腳本，找到config文件夾，看下配置，然后先試啟動下，一般都可以正常啟動了，這些我覺得應(yīng)該是作為一個開發(fā)人員的最基本修養(yǎng)，如果還不具備的小伙伴該好好面壁反思下了。

下載

訪問官方網(wǎng)站，選擇對應(yīng)的版本，然后下載

https://activemq.apache.org/

這里我選擇classic，也就是主流版本，Artemis是下一代的ActiveMQ，類似于開發(fā)版。

安裝

現(xiàn)在的開發(fā)組件，都是解壓即用的

啟動

進入bin目錄，直接雙擊運行，這時候我們會看到，有個cmd窗口一閃而過，說明報錯了，這時候我們打開cmd，這里有個小技巧，在當前文件夾直接輸入cmd即可打開cmd窗口：

然后我們在cmd中再次執(zhí)行activemq.bat，這時候會發(fā)現(xiàn)，cmd中回顯如下提示信息

根據(jù)提示信息，我們知道之前沒有添加參數(shù)(Usage才是最好的文檔，一定要習慣命令行，沒看明白的小伙伴繼續(xù)面壁)，這時候我們知道啟動的命令應(yīng)該是這樣的：

activemq.bat start

回車鍵入，完美啟動：

測試

我們在瀏覽器打開如下地址，如果正常訪問，說明ActiveMQ啟動沒有任何問題：

localhost:8161

用戶名和密碼都是admin，如何修改我們后面再說：

這時候，我們的ActiveMQ就正常啟動了。

docker安裝activemq

這里我們再補充下docker環(huán)境下，如何安裝activemq，很簡單只需要兩行命令即可：

$ docker pull webcenter/activemq:latest # 拉取最新版本的activemq
$ docker run -d --name myactivemq -p 61616:61616 -p 8161:8161 webcenter/activemq:latest # 啟動

關(guān)于docker的使用，我們前面已經(jīng)講過了，還想要了解的小伙伴可以翻一下前面的內(nèi)容。

java整合ActiveMQ

創(chuàng)建java項目

這里我們創(chuàng)建一個java項目，簡單測試下ActiveMQ，創(chuàng)建maven項目，引入如下依賴：

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

創(chuàng)建消息生產(chǎn)者

下面代碼中的注釋已經(jīng)夠詳細了，需要注意的是session.createQueue("test-queue")中的test-queue是我們要發(fā)送消息的目標隊列的名稱，消費者也必須是一樣的名稱，才能正常消費消息，否則是沒有辦法消費的。

session.createTextMessage("hello!test-queue")是發(fā)送消息，這里的"hello!test-queue是我們發(fā)送的消息內(nèi)容。

/**
 *
 * 消息生產(chǎn)者
 *
 * @author syske
 * @version 1.0
 * @date 2021-04-24 10:20:22
 */
public class ActiveMqProducer {

    public void mQProducerQueue() throws Exception{
        //1、創(chuàng)建工廠連接對象，需要制定ip和端口號，這里的端口就是activemq的端口，后面我們再來說如何配置
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //2、使用連接工廠創(chuàng)建一個連接對象
        Connection connection = connectionFactory.createConnection();
        //3、開啟連接
        connection.start();
        //4、使用連接對象創(chuàng)建會話（session）對象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用會話對象創(chuàng)建目標對象，包含queue和topic（一對一和一對多）
        Queue queue = session.createQueue("test-queue");
        //6、使用會話對象創(chuàng)建生產(chǎn)者對象
        MessageProducer producer = session.createProducer(queue);
        //7、使用會話對象創(chuàng)建一個消息對象
        TextMessage textMessage = session.createTextMessage("hello!test-queue");
        //8、發(fā)送消息
        producer.send(textMessage);
        //9、關(guān)閉資源
        producer.close();
        session.close();
        connection.close();
    }
}

創(chuàng)建消息消費者

基本上和消息生產(chǎn)的代碼一致，最大的區(qū)別在于消費者這里創(chuàng)建的是消費者session.createConsumer(queue)，同時啟動了一個消息監(jiān)聽器，實時監(jiān)聽指定隊列，只要隊列中有消息進來，消費者就會執(zhí)行MessageListener中的操作，將消息消費掉，這里的寫法采用了蘭姆達表達式

**
 *
 * 消息消費者
 *
 * @author syske
 * @version 1.0
 * @date 2021-04-24 10:22:21
 */
public class ActiveMqConsumer {
    public void mQConsumerQueue() throws Exception{
        //1、創(chuàng)建工廠連接對象，需要制定ip和端口號
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //2、使用連接工廠創(chuàng)建一個連接對象
        Connection connection = connectionFactory.createConnection();
        //3、開啟連接
        connection.start();
        //4、使用連接對象創(chuàng)建會話（session）對象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用會話對象創(chuàng)建目標對象，包含queue和topic（一對一和一對多）
        Queue queue = session.createQueue("test-queue");
        //6、使用會話對象創(chuàng)建生產(chǎn)者對象
        MessageConsumer consumer = session.createConsumer(queue);
        //7、向consumer對象中設(shè)置一個messageListener對象，用來接收消息
        consumer.setMessageListener(message-> {
            // TODO Auto-generated method stub
            if(message instanceof TextMessage){
                TextMessage textMessage = (TextMessage)message;
                try {
                    System.out.println("消息內(nèi)容：" + textMessage.getText());
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });
        //8、程序等待接收用戶消息
        System.in.read();
        //9、關(guān)閉資源
        consumer.close();
        session.close();
        connection.close();
    }
}

測試

生產(chǎn)者測試

首先我們啟動消費者，生產(chǎn)消息

/**
     * 生產(chǎn)消息
     *
     * @throws Exception
     */
    @Test
    public void produceMessage() throws Exception {
        new ActiveMqProducer().mQProducerQueue();
    }    

運行上面的測試代碼，然后打開ActiveMQ的管理頁面，我們會看到，多了一個消息隊列：

消息隊列名稱就是我們上面指定的test-queue，待消費消息（pending）數(shù)量1，因為我們暫時沒有啟動消費者（consumer），所以消費者數(shù)量是0，隊列中排隊的消息（enqueued）數(shù)量1，被消費消息（dequeued）數(shù)量0。如果你再次啟動生產(chǎn)者，消息隊列中的數(shù)據(jù)會變成2:

消費者測試

這時候，我們啟動消費者

/**
    * 消費消息
    * @throws Exception
    */
    @Test
    public void testConsumeMessage() throws Exception {
        new ActiveMqConsumer().mQConsumerQueue();
    }

然后我們會看到控制臺輸出如下信息：

這些消息就是我們在生產(chǎn)者中發(fā)送的消息。我們再訪問隊列管理頁面看一下：

和上面消費者啟動之前相比，消息已經(jīng)被消費，未消費消息數(shù)量0，已消費2，消費者數(shù)量1。

擴展

一般在實際項目中，我們會啟動異步線程發(fā)送消息（消息生產(chǎn)者），然后由消息中心集中對消息進行消費，并將消息推送至客戶端，比如微信公眾號、釘釘、安卓端、IOS端等。需要注意的是，不要在循環(huán)中調(diào)用異步線程，否則會導致線程池資源耗盡報錯，而且這種錯誤本地和測試環(huán)境很難復現(xiàn)。

我最近剛處理和優(yōu)化的一個問題就是for循環(huán)中調(diào)用了異步發(fā)送消息的方法，導致線程池資源耗盡報錯，但是本地和測試一直無法復現(xiàn)錯誤，直到最后在異步線程方法中加了sleep之后bug才復現(xiàn)，當然也說明線上環(huán)境復雜，消息隊列響應(yīng)慢，所以為了避免這類情況出現(xiàn)，一定在寫代碼的時候要盡可能規(guī)范合理，畢竟本地沒問題不代表線上沒問題，并發(fā)量為10的時候，接口一切響應(yīng)正常，但是如果并發(fā)量發(fā)生量級變化，變成100甚至1000，性能問題就出現(xiàn)了。

配置

這里我們再簡單說下activemq的配置，進入conf文件夾，可以看到如下文件：

端口配置

其中，端口的配置在jetty.xml中：

管理賬戶配置

管理用戶配置在jetty-realm.properties中，配置規(guī)則也很簡單:

用戶名: 密碼， 角色

其他配置

剩余的文件都是有關(guān)登錄驗證的配置，比如jaas、credentials，暫時未涉及，這里也就不做探討，等后面了解之后再來分享。

結(jié)語

今天的內(nèi)容已經(jīng)比較完整地介紹了activemq的基礎(chǔ)內(nèi)容，包括下載、安裝、配置以及項目整合，就算之前沒接觸過activemq的小伙伴，通過今天的內(nèi)容，我想你也應(yīng)該對activemq有了基本的認知，基本上可以確保在工作中用到activemq的時候，你可以很好地上手。當然，今天的內(nèi)容，也算是我最近一段時間學習activemq的一個簡單總結(jié)，后面還需要進一步的深入學習和探索。

最后，再給各個小伙伴一個建議，學習的時候要盡可能多看官方文檔，多留意提示信息，多橫向比較，多做延申，多積累，要慢慢培養(yǎng)自己系統(tǒng)的學習思維和習慣，提升自己知識橫向擴展能力，定期做知識的歸納整理，將知識點連成知識鏈，再將知識鏈鏈接形成知識面，這樣隨著你知識面的不斷擴展，你會發(fā)現(xiàn)你學東西的時候效率會更好，學東西會更快，而且學習質(zhì)量也更高，畢竟大腦更擅長的是關(guān)聯(lián)，而不是記憶。另外，就算大腦是個數(shù)據(jù)庫，那你也應(yīng)該知道，建立索引，我們的知識才可以更快速地被檢索，所以管理知識，其實比學習知識更重要。