ActiveMQ

1、什么是消息中間件

消息中間件利用高效可靠的消息傳遞機制進行平台無關的數據交流，並基於數據通信來進行分布式系統的集成。通過提供消息傳遞和消息排隊模型，它可以在分布式環境下擴展進程間的通信。對於消息中間件，常見的角色大致也就有Producer（生產者）、Consumer（消費者）

執行過程長，不需要返回結果的服務可以用消息中間件。

常見的消息中間件產品:

（1）ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現。我們在本次課程中介紹 ActiveMQ的使用。中小企業使用較多。

（2）RabbitMQ

AMQP協議的領導實現，支持多種場景。淘寶的MySQL集群內部有使用它進行通訊，OpenStack開源雲平台的通信組件，最先在金融行業得到運用。適合大型企業。

（3）ZeroMQ

史上最快的消息隊列系統

（4）Kafka

Apache下的一個子項目 。特點：高吞吐，在一台普通的服務器上既可以達到10W/s的吞吐速率；完全的分布式系統。適合處理海量數據。

 

2、 JMS簡介

2.1 什么是JMS

JMS（Java Messaging Service）是Java平台上有關面向消息中間件的技術規范，它便於消息系統中的Java應用程序進行消息交換,並且通過提供標准的產生、發送、接收消息的接口簡化企業應用的開發。

       JMS本身只定義了一系列的接口規范，是一種與廠商無關的 API，用來訪問消息收發系統。它類似於 JDBC(java Database Connectivity)：這里，JDBC 是可以用來訪問許多不同關系數據庫的 API，而 JMS 則提供同樣與廠商無關的訪問方法，以訪問消息收發服務。許多廠商目前都支持 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ，這只是幾個例子。 JMS 使您能夠通過消息收發服務（有時稱為消息中介程序或路由器）從一個 JMS 客戶機向另一個 JML 客戶機發送消息。消息是 JMS 中的一種類型對象，由兩部分組成：報頭和消息主體。報頭由路由信息以及有關該消息的元數據組成。消息主體則攜帶着應用程序的數據或有效負載。

JMS 定義了五種不同的消息正文格式，以及調用的消息類型，允許你發送並接收以一

些不同形式的數據，提供現有消息格式的一些級別的兼容性。

· TextMessage--一個字符串對象

· MapMessage--一套名稱-值對

· ObjectMessage--一個序列化的 Java 對象

· BytesMessage--一個字節的數據流

· StreamMessage -- Java 原始值的數據流

2.2  JMS消息傳遞類型

對於消息的傳遞有兩種類型：

一種是點對點的，即一個生產者和一個消費者一一對應；

 

 

另一種是發布/ 訂閱模式，即一個生產者產生消息並進行發送后，可以由多個消費者進

行接收。

 

 

3、ActiveMQ下載與安裝

其實這是一個解壓即用的項目

3.1 下載

官方網站下載：http://activemq.apache.org/

3.2安裝（Linux）

（1）將apache-activemq-5.12.0-bin.tar.gz 上傳至服務器

（2）解壓此文件

tar  zxvf  apache-activemq-5.12.0-bin.tar.gz

（3）為apache-activemq-5.12.0目錄賦權

chmod 777 apache-activemq-5.12.0

（4）進入apache-activemq-5.12.0\bin目錄

（5）賦與執行權限

chmod 755 activemq 
--------------------------------------  知識點小貼士    --------------------------
linux 命令chmod 755的意思
chmod是Linux下設置文件權限的命令，后面的數字表示不同用戶或用戶組的權限。
一般是三個數字：
第一個數字表示文件所有者的權限
第二個數字表示與文件所有者同屬一個用戶組的其他用戶的權限
第三個數字表示其它用戶組的權限。
      權限分為三種：讀（r=4），寫（w=2），執行（x=1） 。 綜合起來還有可讀可執行（rx=5=4+1）、可讀可寫（rw=6=4+2）、可讀可寫可執行(rwx=7=4+2+1)。
      所以，chmod 755 設置用戶的權限為： 
1.文件所有者可讀可寫可執行                                   --7
2.與文件所有者同屬一個用戶組的其他用戶可讀可執行 --5 
3.其它用戶組可讀可執行              

3.3 啟動

 

 ./activemq start

 

出現下列提示表示成功！

 

windows系統啟動 需要進入/bin 目錄下 選擇 win64 或 win32 里面的 activemq.bat 啟動。當然Windows下載的安裝包不一樣，請看官網。

 

假設服務器地址為192.168.25.135 ，打開瀏覽器輸入地址

http://192.168.25.135:8161/ 即可進入ActiveMQ管理頁面

 

 

點擊進入管理頁面

 

 

輸入用戶名和密碼  均為 admin

 

 

進入主界面

 

 

點對點消息列表Qucues：

 

 發布/訂閱消息列表Topics:

 

列表各列信息含義如下：

Number Of Pending Messages  ：等待消費的消息 這個是當前未出隊列的數量。

Number Of Consumers  ：消費者 這個是消費者端的消費者數量

Messages Enqueued  ：進入隊列的消息  進入隊列的總數量,包括出隊列的。

Messages Dequeued  ：出了隊列的消息  可以理解為是消費這消費掉的數量。

Messages Enqueued = Messages Dequeued  + Number Of Pending Messages 

 

4、 JMS入門小Demo

4.1 點對點模式

點對點的模式主要建立在一個隊列上面，當連接一個列隊的時候，發送端不需要知道接收端是否正在接收，可以直接向ActiveMQ發送消息，發送的消息，將會先進入隊列中，如果有接收端在監聽，則會發向接收端，如果沒有接收端接收，則會保存在activemq服務器，直到接收端接收消息，點對點的消息模式可以有多個發送端，多個接收端，但是一條消息，只會被一個接收端給接收到，哪個接收端先連上ActiveMQ，則會先接收到，而后來的接收端則接收不到那條消息。

消息生產者

（1）創建工程jmsDemo ，引入依賴

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

 

（2）創建類QueueProducer  main方法代碼如下：

 　　//1.創建連接工廠
    ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");

    //2.獲取連接
    Connection connection = connectionFactory.createConnection();

    //3.啟動連接
    connection.start();

    //4.獲取session  (參數1：是否啟動事務,參數2：消息確認模式)
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);       

    //5.創建隊列對象
    Queue queue = session.createQueue("test-queue");

    //6.創建消息生產者
    MessageProducer producer = session.createProducer(queue);

    //7.創建消息
    TextMessage textMessage = session.createTextMessage("歡迎來到神奇的品優購世界");

    //8.發送消息
    producer.send(textMessage);

    //9.關閉資源
    producer.close();
    session.close();
    connection.close();

上述代碼中第4步創建session  的兩個參數：

第1個參數 是否使用事務

第2個參數 消息的確認模式

• AUTO_ACKNOWLEDGE = 1    自動確認
• CLIENT_ACKNOWLEDGE = 2    客戶端手動確認   
• DUPS_OK_ACKNOWLEDGE = 3    自動批量確認
• SESSION_TRANSACTED = 0    事務提交並確認

運行后通過ActiveMQ管理界面查詢

 

消息消費者

創建類QueueConsumer ，main方法代碼如下：

 　　//1.創建連接工廠
    ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");

    //2.獲取連接
    Connection connection = connectionFactory.createConnection();

    //3.啟動連接
    connection.start();

    //4.獲取session  (參數1：是否啟動事務,參數2：消息確認模式)
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    //5.創建隊列對象
    Queue queue = session.createQueue("test-queue");

    //6.創建消息消費
    MessageConsumer consumer = session.createConsumer(queue);

    //7.監聽消息
    consumer.setMessageListener(new MessageListener() {
        publicvoid onMessage(Message message) {
            TextMessage textMessage=(TextMessage)message;
            try {
                System.out.println("接收到消息："+textMessage.getText());
            } catch (JMSException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    });

    //8.等待鍵盤輸入
    System.in.read();  

    //9.關閉資源
    consumer.close();
    session.close();
    connection.close();

 執行后看到控制台輸出

運行測試

同時開啟2個以上的消費者，再次運行生產者，觀察每個消費者控制台的輸出，會發現只有一個消費者會接收到消息。

 

4.2  發布/訂閱模式

消息生產者

創建類TopicProducer ，main方法代碼如下：

　　//1.創建連接工廠

    ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");

    //2.獲取連接

    Connection connection = connectionFactory.createConnection();

    //3.啟動連接

    connection.start();

    //4.獲取session  (參數1：是否啟動事務,參數2：消息確認模式)

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    //5.創建主題對象

    Topic topic = session.createTopic("test-topic");

    //6.創建消息生產者

    MessageProducer producer = session.createProducer(topic);

    //7.創建消息

    TextMessage textMessage = session.createTextMessage("歡迎來到神奇的品優購世界");

    //8.發送消息

    producer.send(textMessage);

    //9.關閉資源

    producer.close();

    session.close();

  connection.close();

 運行效果如下：

 

消息消費者

創建類TopicConsumer ,main方法代碼如下：

//1.創建連接工廠
    ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");
    //2.獲取連接
    Connection connection = connectionFactory.createConnection();
    //3.啟動連接
    connection.start();
    //4.獲取session  (參數1：是否啟動事務,參數2：消息確認模式)
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    //5.創建主題對象
    //Queue queue = session.createQueue("test-queue");
    Topic topic = session.createTopic("test-topic");
    //6.創建消息消費
    MessageConsumer consumer = session.createConsumer(topic);
    
    //7.監聽消息
    consumer.setMessageListener(new MessageListener() {
        publicvoid onMessage(Message message) {
            TextMessage textMessage=(TextMessage)message;
            try {
                System.out.println("接收到消息："+textMessage.getText());
            } catch (JMSException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    });
    //8.等待鍵盤輸入
    System.in.read();
    //9.關閉資源
    consumer.close();
    session.close();
    connection.close();    

運行測試

同時開啟2個以上的消費者，再次運行生產者，觀察每個消費者控制台的輸出，會發現每個消費者會接收到消息。