訊息佇列概念和使用場景及ActiveMQ詳細入門使用教程

阿新 • • 發佈：2018-12-15

一、訊息佇列

訊息即是資訊的載體。為了讓訊息傳送者和訊息接收者都能夠明白訊息所承載的資訊（訊息傳送者需要知道如何構造訊息；訊息接收者需要知道如何解析訊息），它們就需要按照一種統一的格式描述訊息，這種統一的格式稱之為訊息協議。所以，有效的訊息一定具有某一種格式；而沒有格式的訊息是沒有意義的。

而訊息從傳送者到接收者的方式也有兩種。一種我們可以稱為即時訊息通訊，也就是說訊息從一端發出後（訊息傳送者）立即就可以達到另一端（訊息接收者），這種方式的具體實現就是我們已經介紹過的RPC（當然單純的http通訊也滿足這個定義）；另一種方式稱為延遲訊息通訊，即訊息從某一端發出後，首先進入一個容器進行臨時儲存，當達到某種條件後，再由這個容器傳送給另一端。這個容器的一種具體實現就是訊息佇列。二、訊息佇列的應用場景

以下介紹訊息佇列在實際應用中常用的使用場景。非同步處理，應用解耦，流量削鋒和訊息通訊四個場景。 2.1非同步處理

場景說明：使用者註冊後，需要發註冊郵件和註冊簡訊。傳統的做法有兩種1.序列的方式；2.並行方式。

**（1）序列方式：**將註冊資訊寫入資料庫成功後，傳送註冊郵件，再發送註冊簡訊。以上三個任務全部完成後，返回給客戶端。

**（2）並行方式：**將註冊資訊寫入資料庫成功後，傳送註冊郵件的同時，傳送註冊簡訊。以上三個任務完成後，返回給客戶端。與序列的差別是，並行的方式可以提高處理的時間。

假設三個業務節點每個使用50毫秒鐘，不考慮網路等其他開銷，則序列方式的時間是150毫秒，並行的時間可能是100毫秒。

因為CPU在單位時間內處理的請求數是一定的，假設CPU1秒內吞吐量是100次。則序列方式1秒內CPU可處理的請求量是7次（1000/150）。並行方式處理的請求量是10次（1000/100）。

小結：如以上案例描述，傳統的方式系統的效能（併發量，吞吐量，響應時間）會有瓶頸。如何解決這個問題呢？

引入訊息佇列，將不是必須的業務邏輯，非同步處理。改造後的架構如下：

按照以上約定，使用者的響應時間相當於是註冊資訊寫入資料庫的時間，也就是50毫秒。註冊郵件，傳送簡訊寫入訊息佇列後，直接返回，因此寫入訊息佇列的速度很快，基本可以忽略，因此使用者的響應時間可能是50毫秒。因此架構改變後，系統的吞吐量提高到每秒20 QPS。比序列提高了3倍，比並行提高了兩倍。 2.2應用解耦

場景說明：使用者下單後，訂單系統需要通知庫存系統。傳統的做法是，訂單系統呼叫庫存系統的介面。如下圖：

傳統模式的缺點：

1）假如庫存系統無法訪問，則訂單減庫存將失敗，從而導致訂單失敗；

2）訂單系統與庫存系統耦合；

如何解決以上問題呢？引入應用訊息佇列後的方案，如下圖：

訂單系統：使用者下單後，訂單系統完成持久化處理，將訊息寫入訊息佇列，返回使用者訂單下單成功。
庫存系統：訂閱下單的訊息，採用拉/推的方式，獲取下單資訊，庫存系統根據下單資訊，進行庫存操作。
假如：在下單時庫存系統不能正常使用。也不影響正常下單，因為下單後，訂單系統寫入訊息佇列就不再關心其他的後續操作了。實現訂單系統與庫存系統的應用解耦。

2.3流量削鋒

流量削鋒也是訊息佇列中的常用場景，一般在秒殺或團搶活動中使用廣泛。

應用場景：秒殺活動，一般會因為流量過大，導致流量暴增，應用掛掉。為解決這個問題，一般需要在應用前端加入訊息佇列。

可以控制活動的人數；
可以緩解短時間內高流量壓垮應用；



使用者的請求，伺服器接收後，首先寫入訊息佇列。假如訊息佇列長度超過最大數量，則直接拋棄使用者請求或跳轉到錯誤頁面；
秒殺業務根據訊息佇列中的請求資訊，再做後續處理。

2.4日誌處理

日誌處理是指將訊息佇列用在日誌處理中，比如Kafka的應用，解決大量日誌傳輸的問題。架構簡化如下：

日誌採集客戶端，負責日誌資料採集，定時寫受寫入Kafka佇列；
Kafka訊息佇列，負責日誌資料的接收，儲存和轉發；
日誌處理應用：訂閱並消費kafka佇列中的日誌資料；

以下是新浪kafka日誌處理應用案例：轉自（新浪技術分享：我們如何扛下32億條實時日誌的分析處理）

(1)Kafka：接收使用者日誌的訊息佇列。

(2)Logstash：做日誌解析，統一成JSON輸出給Elasticsearch。

(3)Elasticsearch：實時日誌分析服務的核心技術，一個schemaless，實時的資料儲存服務，通過index組織資料，兼具強大的搜尋和統計功能。

(4)Kibana：基於Elasticsearch的資料視覺化元件，超強的資料視覺化能力是眾多公司選擇ELK stack的重要原因。 2.5訊息通訊

訊息通訊是指，訊息佇列一般都內建了高效的通訊機制，因此也可以用在純的訊息通訊。比如實現點對點訊息佇列，或者聊天室等。

點對點通訊：

客戶端A和客戶端B使用同一佇列，進行訊息通訊。

聊天室通訊：

客戶端A，客戶端B，客戶端N訂閱同一主題，進行訊息釋出和接收。實現類似聊天室效果。

以上實際是訊息佇列的兩種訊息模式，點對點或釋出訂閱模式。

三、訊息模式 1. 點對點模式和釋出訂閱模式：是否可以重複消費

P2P模式：

P2P模式包含三個角色：訊息佇列（Queue），傳送者(Sender)，接收者(Receiver)。每個訊息都被髮送到一個特定的佇列，接收者從佇列中獲取訊息。佇列保留著訊息，直到他們被消費或超時。 P2P的特點

每個訊息只有一個消費者（Consumer）(即一旦被消費，訊息就不再在訊息佇列中)
傳送者和接收者之間在時間上沒有依賴性，也就是說當傳送者傳送了訊息之後，不管接收者有沒有正在執行，它不會影響到訊息被髮送到佇列
接收者在成功接收訊息之後需向佇列應答成功

如果希望傳送的每個訊息都會被成功處理的話，那麼需要P2P模式。、

Pub/sub模式：

包含三個角色：主題（Topic），釋出者（Publisher），訂閱者（Subscriber）。多個釋出者將訊息傳送到Topic,系統將這些訊息傳遞給多個訂閱者。 Pub/Sub的特點

每個訊息可以有多個消費者
釋出者和訂閱者之間有時間上的依賴性。針對某個主題（Topic）的訂閱者，它必須建立一個訂閱者之後，才能消費釋出者的訊息。
為了消費訊息，訂閱者必須保持執行的狀態。

為了緩和這樣嚴格的時間相關性，JMS允許訂閱者建立一個可持久化的訂閱。這樣，即使訂閱者沒有被啟用（執行），它也能接收到釋出者的訊息。如果希望傳送的訊息可以不被做任何處理、或者只被一個訊息者處理、或者可以被多個消費者處理的話，那麼可以採用Pub/Sub模型。 2. 推模式和拉模式：訊息的更新者

推(push)模式是一種基於C/S機制、由伺服器主動將資訊送到客戶器的技術。

在push模式應用中，伺服器把資訊送給客戶器之前，並沒有明顯的客戶請求。push事務由伺服器發起。push模式可以讓資訊主動、快速地尋找使用者/客戶器，資訊的主動性和實時性比較好。但精確性較差，可能推送的資訊並不一定滿足客戶的需求。
推送模式不能保證能把資訊送到客戶器，因為推模式採用了廣播機制，如果客戶器正好聯網並且和伺服器在同一個頻道上，推送模式才是有效的。
push模式無法跟蹤狀態，採用了開環控制模式，沒有使用者反饋資訊。在實際應用中，由客戶器向伺服器傳送一個申請，並把自己的地址（如IP、port）告知伺服器，然後伺服器就源源不斷地把資訊推送到指定地址。在多媒體資訊廣播中也採用了推模式。

拉（pull）模式與推模式相反，是由客戶器主動發起的事務。

伺服器把自己所擁有的資訊放在指定地址（如IP、port），客戶器向指定地址傳送請求，把自己需要的資源“拉”回來。不僅可以準確獲取自己需要的資源，還可以及時把客戶端的狀態反饋給伺服器。

ActiveMQ介紹

 MQ是訊息中介軟體，是一種在分散式系統中應用程式藉以傳遞訊息的媒介，常用的有ActiveMQ，RabbitMQ，kafka。ActiveMQ是Apache下的開源專案，完全支援JMS1.1和J2EE1.4規範的JMS Provider實現。

特點： 1、支援多種語言編寫客戶端 2、對spring的支援，很容易和spring整合 3、支援多種傳輸協議：TCP,SSL,NIO,UDP等 4、支援AJAX 訊息形式： 1、點對點（queue） 2、一對多（topic）

ActiveMQ安裝 在這裡插入圖片描述 ActiveMQ測試

編寫一個測試類對ActiveMQ進行測試，首先得向pom檔案中新增ActiveMQ相關的jar包：

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

queue的傳送程式碼如下:

    public void testMQProducerQueue() throws Exception{
        //1、建立工廠連線物件，需要制定ip和埠號
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
        //2、使用連線工廠建立一個連線物件
        Connection connection = connectionFactory.createConnection();
        //3、開啟連線
        connection.start();
        //4、使用連線物件建立會話（session）物件
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用會話物件建立目標物件，包含queue和topic（一對一和一對多）
        Queue queue = session.createQueue("test-queue");
        //6、使用會話物件建立生產者物件
        MessageProducer producer = session.createProducer(queue);
        //7、使用會話物件建立一個訊息物件
        TextMessage textMessage = session.createTextMessage("hello!test-queue");
        //8、傳送訊息
        producer.send(textMessage);
        //9、關閉資源
        producer.close();
        session.close();
        connection.close();
    }

接收程式碼：

    public void TestMQConsumerQueue() throws Exception{
        //1、建立工廠連線物件，需要制定ip和埠號
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
        //2、使用連線工廠建立一個連線物件
        Connection connection = connectionFactory.createConnection();
        //3、開啟連線
        connection.start();
        //4、使用連線物件建立會話（session）物件
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用會話物件建立目標物件，包含queue和topic（一對一和一對多）
        Queue queue = session.createQueue("test-queue");
        //6、使用會話物件建立生產者物件
        MessageConsumer consumer = session.createConsumer(queue);
        //7、向consumer物件中設定一個messageListener物件，用來接收訊息
        consumer.setMessageListener(new MessageListener() {
 
            @Override
            public void onMessage(Message message) {
                // TODO Auto-generated method stub
                if(message instanceof TextMessage){
                    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();
    }

接著對topic進行測試，傳送程式碼如下：

public void TestTopicProducer() throws Exception{
    //1、建立工廠連線物件，需要制定ip和埠號
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
    //2、使用連線工廠建立一個連線物件
    Connection connection = connectionFactory.createConnection();
    //3、開啟連線
    connection.start();
    //4、使用連線物件建立會話（session）物件
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    //5、使用會話物件建立目標物件，包含queue和topic（一對一和一對多）
    Topic topic = session.createTopic("test-topic");
    //6、使用會話物件建立生產者物件
    MessageProducer producer = session.createProducer(topic);
    //7、使用會話物件建立一個訊息物件
    TextMessage textMessage = session.createTextMessage("hello!test-topic");
    //8、傳送訊息
    producer.send(textMessage);
    //9、關閉資源
    producer.close();
    session.close();
    connection.close();
}

接收程式碼：

    public void TestTopicConsumer() throws Exception{
        //1、建立工廠連線物件，需要制定ip和埠號
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
        //2、使用連線工廠建立一個連線物件
        Connection connection = connectionFactory.createConnection();
        //3、開啟連線
        connection.start();
        //4、使用連線物件建立會話（session）物件
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用會話物件建立目標物件，包含queue和topic（一對一和一對多）
        Topic topic = session.createTopic("test-topic");
        //6、使用會話物件建立生產者物件
        MessageConsumer consumer = session.createConsumer(topic);
        //7、向consumer物件中設定一個messageListener物件，用來接收訊息
        consumer.setMessageListener(new MessageListener() {
 
            @Override
            public void onMessage(Message message) {
                // TODO Auto-generated method stub
                if(message instanceof TextMessage){
                    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();
    }

ActiveMQ整合spring及專案中運用

activeMQ與spring看一整合到一起使用，除了新增ActiveMQ相關的jar包外，還需要新增spring的jar包：

    <dependency>  
        <groupId>org.springframework</groupId>  
        <artifactId>spring-context</artifactId>  
    </dependency>

然後編寫applicationContext-activemq.xml檔案，程式碼如下：




<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"  
    xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"  
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd  
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd  
    http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd  
    http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd  
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">  
 
    <!-- 配置能夠產生connection的connectionfactory，由JMS對應的服務廠商提供 -->
    <bean id="tagertConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg name="brokerURL" value="tcp://192.168.156.44:61616"/>
    </bean>
    <!-- 配置spring管理真正connectionfactory的connectionfactory，相當於spring對connectionfactory的一層封裝 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="tagertConnectionFactory"/>
    </bean>
    <!-- 配置生產者 -->
    <!-- Spring使用JMS工具類，可以用來發送和接收訊息 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <!-- 這裡是配置的spring用來管理connectionfactory的connectionfactory -->
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <!-- 配置destination -->
    <!-- 佇列目的地 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="spring-queue"/>
    </bean>
    <!-- 話題目的地 -->
    <bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="item-add-topic"/>
    </bean>
</beans>

然後在我們淘淘商城中，商品新增到資料庫的時候，對應也要新增資料到我們的solr索引中，所以生產者應該在插入資料後建立：在這裡插入圖片描述當然，在xml檔案中配置好的jmstemplate和destination也要注入進來：

    @Autowired
    private JmsTemplate jmsTemplate;
    @Resource(name="itemAddTopic")
    private Destination destination;

然後消費者應該寫在我們的搜尋工程中，首先新增spring和activeMQ的jar包，然後配置xml檔案，再編寫一個監聽器，當接收到訊息時，就講資料存入索引庫，xml檔案程式碼如下：

<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"  
    xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"  
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd  
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd  
    http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd  
    http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd  
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">  
 
    <!-- 配置能夠產生connection的connectionfactory，由JMS對應的服務廠商提供 -->
    <bean id="tagertConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg name="brokerURL" value="tcp://192.168.156.44:61616"/>
    </bean>
    <!-- 配置spring管理真正connectionfactory的connectionfactory，相當於spring對connectionfactory的一層封裝 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="tagertConnectionFactory"/>
    </bean>
    <!-- 配置destination -->
    <!-- 佇列目的地 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="spring-queue"/>
    </bean>
    <!-- 話題目的地 -->
    <bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="item-add-topic"/>
    </bean>
    <!-- 配置監聽器 -->
    <bean id="myListener" class="com.taotao.search.listener.MyListener"/>
    <bean id="itemAddListener" class="com.taotao.search.listener.ItemAddListener"/>
    <!-- 系統監聽器 -->
<!--    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="queueDestination"/>
        <property name="messageListener" ref="myListener"/>
    </bean> -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="itemAddTopic"/>
        <property name="messageListener" ref="itemAddListener"/>
    </bean>
</beans>

接收訊息程式碼：在這裡插入圖片描述

訊息佇列概念和使用場景及ActiveMQ詳細入門使用教程

訊息佇列概念和使用場景及ActiveMQ詳細入門使用教程

MQ入門總結（一）訊息佇列概念和使用場景

PHP訊息佇列實現及應用：訊息佇列概念介紹

訊息佇列的應用場景和常見的訊息佇列之間的比較

ActiveMQ基本詳解與總結& 訊息佇列-推/拉模式學習 & ActiveMQ及JMS學習

數據結構11: 棧(Stack)的概念和應用及C語言實現

MQ 訊息佇列的常用場景

RabbitMQ訊息佇列（一）: Detailed Introduction 詳細介紹

RabbitMQ訊息佇列生產者和消費者

RabbitMQ實戰教程(十二):訊息佇列的應用場景

java呼叫rabbitmq訊息佇列傳送和接收訊息例項

Java中mqtt訊息佇列傳送和訂閱訊息

UI介面快速展示優化-巧用訊息佇列MessageQueue中IdleHandler及SyncBarrier

訊息佇列mq的原理及實現方法

linux中訊息佇列kfifo和訊號量sem_t的用法

SurfaceView的概念和優點及使用

Winddows 訊息佇列原理和程式設計例項

訊息佇列的使用場景

訊息佇列的應用場景

面試官：說出八種訊息佇列的應用場景。啊？八種？

訊息佇列概念和使用場景 及ActiveMQ詳細入門使用教程

相關推薦

訊息佇列概念和使用場景及ActiveMQ詳細入門使用教程