039 RabbitMq及數據同步01

時間 2019-11-25

標籤 rabbitmq 數據同步欄目 RabbitMQ 简体版

原文原文鏈接

1.RabbitMqhtml

(1)問題引出git

目前咱們已經完成了商品詳情和搜索系統的開發。咱們思考一下，是否存在問題？github

商品的原始數據保存在數據庫中，增刪改查都在數據庫中完成。數據庫
搜索服務數據來源是索引庫，若是數據庫商品發生變化，索引庫數據不能及時更新。windows
商品詳情作了頁面靜態化，靜態頁面數據也不會隨着數據庫商品發生變化。瀏覽器

若是咱們在後臺修改了商品的價格，搜索頁面和商品詳情頁顯示的依然是舊的價格，這樣顯然不對。該如何解決？併發

這裏有兩種解決方案：異步

方案1：每當後臺對商品作增刪改操做，同時要修改索引庫數據及靜態頁面分佈式
方案2：搜索服務和商品頁面服務對外提供操做接口，後臺在商品增刪改後，調用接口ide

以上兩種方式都有同一個嚴重問題：就是代碼耦合，後臺服務中須要嵌入搜索和商品頁面服務，違背了微服務的獨立原則。

因此，咱們會經過另一種方式來解決這個問題：消息隊列

(2)消息隊列

<1>什麼是消息隊列

消息隊列，即MQ，Message Queue。

消息隊列是典型的：生產者、消費者模型。生產者不斷向消息隊列中生產消息，消費者不斷的從隊列中獲取消息。由於消息的生產和消費都是異步的，並且只關心消息的發送和接收，沒有業務邏輯的侵入，這樣就實現了生產者和消費者的解耦。

結合前面所說的問題：

商品服務對商品增刪改之後，無需去操做索引庫或靜態頁面，只是發送一條消息，也不關心消息被誰接收。
搜索服務和靜態頁面服務接收消息，分別去處理索引庫和靜態頁面。

若是之後有其它系統也依賴商品服務的數據，一樣監聽消息便可，商品服務無需任何代碼修改。

<2>AMQP和JMS

MQ是消息通訊的模型，並非具體實現。如今實現MQ的有兩種主流方式：AMQP、JMS。

二者間的區別和聯繫：

JMS是定義了統一的接口，來對消息操做進行統一；AMQP是經過規定協議來統一數據交互的格式
JMS限定了必須使用Java語言；AMQP只是協議，不規定實現方式，所以是跨語言的。
JMS規定了兩種消息模型；而AMQP的消息模型更加豐富

<3>常見MQ產品

ActiveMQ：基於JMS
RabbitMQ：基於AMQP協議，erlang語言開發，穩定性好
RocketMQ：基於JMS，阿里巴巴產品，目前交由Apache基金會
Kafka：分佈式消息系統，高吞吐量

<4>RabbitMQ

RabbitMQ是基於AMQP的一款消息管理系統

官網： http://www.rabbitmq.com/

官方教程：http://www.rabbitmq.com/getstarted.html

(3)下載與安裝(windows環境)

<1>下載安裝

RabbitMQ由Erlang語言開發，Erlang語言用於併發及分佈式系統的開發，在電信領域應用普遍，OTP（Open Telecom Platform）做爲Erlang語言的一部分，包含了不少基於Erlang開發的中間件及工具庫，安裝RabbitMQ須要安裝Erlang/OTP，並保持版本匹配，以下圖：

RabbitMQ的下載地址：http://www.rabbitmq.com/download.html

本項目使用Erlang/OTP 20.3版本和RabbitMQ3.7.3版本。

1）下載erlang
地址以下：
http://erlang.org/download/otp_win64_20.3.exe，以管理員方式運行此文件，安裝。

erlang安裝完成須要配置erlang環境變量： ERLANG_HOME=D:\Program Files\erl9.3 在path中添加%ERLANG_HOME%\bin;

此電腦-->鼠標右鍵「屬性」-->高級系統設置-->環境變量-->「新建」系統環境變量

變量名：ERLANG_HOME

變量值就是剛纔erlang的安裝地址，點擊肯定。

而後雙擊系統變量path

點擊「新建」，將%ERLANG_HOME%\bin加入到path中。

最後windows鍵+R鍵，輸入cmd，再輸入erl，看到版本號就說明erlang安裝成功了。

2）安裝RabbitMQ
https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.7.3,以管理員方式運行此文件，安裝。

<2>啓動
安裝成功後會自動建立RabbitMQ服務而且啓動。
1）從開始菜單啓動RabbitMQ
完成在開始菜單找到RabbitMQ的菜單：

RabbitMQ Service-install :安裝服務
RabbitMQ Service-remove 刪除服務
RabbitMQ Service-start 啓動
RabbitMQ Service-stop 中止

2）若是沒有開始菜單則進入安裝目錄下sbin目錄手動啓動：

　　1）安裝並運行服務
　　rabbitmq-service.bat install 安裝服務 rabbitmq-service.bat stop 中止服務 rabbitmq-service.bat start 啓動服務

　　2）安裝管理插件
　　安裝rabbitMQ的管理插件，方便在瀏覽器端管理RabbitMQ
　　管理員身份運行 rabbitmq-plugins.bat enable rabbitmq_management

安裝完插件後，須要重啓rabbitmq,此電腦-->鼠標右鍵「管理」-->服務與應用程序-->服務

打開sbin目錄，雙擊rabbitmq-server.bat

等幾秒鐘看到這個界面後，訪問http://localhost:15672

初始帳號和密碼：guest/guest

<3>注意事項

安裝erlang和rabbitMQ以管理員身份運行。
當卸載從新安裝時會出現RabbitMQ服務註冊失敗，此時須要進入註冊表清理erlang搜索RabbitMQ、ErlSrv，將對應的項所有刪除。

2.五種消息模型

RabbitMQ提供了6種消息模型，可是第6種實際上是RPC，並非MQ，所以不予學習。那麼也就剩下5種。

可是其實三、四、5這三種都屬於訂閱模型，只不過進行路由的方式不一樣。

咱們經過一個demo工程來了解下RabbitMQ的工做方式：

向已存在的工程導入模塊：

導入成功後，模塊的結構：

package cn.itcast.rabbitmq.simple;

import cn.itcast.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
/**
 * 生產者
 */
public class Send {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 從鏈接中建立通道，使用通道才能完成消息相關的操做
        Channel channel = connection.createChannel();
        // 聲明（建立）隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 消息內容
        String message = "Hello World!";
        // 向指定的隊列中發送消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        
        System.out.println(" [x] Sent '" + message + "'");

        //關閉通道和鏈接
        channel.close();
        connection.close();
    }
}

(1)基本消息模型

RabbitMQ是一個消息代理：它接受和轉發消息。你能夠把它想象成一個郵局：當你把郵件放在郵箱裏時，你能夠肯定郵差先生最終會把郵件發送給你的收件人。在這個比喻中，RabbitMQ是郵政信箱，郵局和郵遞員。

RabbitMQ與郵局的主要區別是它不處理紙張，而是接受，存儲和轉發數據消息的二進制數據塊。

P（producer/ publisher）：生產者，一個發送消息的用戶應用程序。

C（consumer）：消費者，消費和接收有相似的意思，消費者是一個主要用來等待接收消息的用戶應用程序

隊列（紅色區域）：rabbitmq內部相似於郵箱的一個概念。雖然消息流經rabbitmq和你的應用程序，可是它們只能存儲在隊列中。隊列只受主機的內存和磁盤限制，實質上是一個大的消息緩衝區。許多生產者能夠發送消息到一個隊列，許多消費者能夠嘗試從一個隊列接收數據。

總之：

生產者將消息發送到隊列，消費者從隊列中獲取消息，隊列是存儲消息的緩衝區。

咱們將用Java編寫兩個程序;發送單個消息的生產者，以及接收消息並將其打印出來的消費者。咱們將詳細介紹Java API中的一些細節，這是一個消息傳遞的「Hello World」。

咱們將調用咱們的消息發佈者（發送者）Send和咱們的消息消費者（接收者）Recv。發佈者將鏈接到RabbitMQ，發送一條消息，而後退出。

<1>生產者發送消息

public class Send {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 從鏈接中建立通道，這是完成大部分API的地方。
        Channel channel = connection.createChannel();

        // 聲明（建立）隊列，必須聲明隊列纔可以發送消息，咱們能夠把消息發送到隊列中。
        // 聲明一個隊列是冪等的 - 只有當它不存在時纔會被建立
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 消息內容
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        //關閉通道和鏈接
        channel.close();
        connection.close();
    }
}

控制檯：

<2>管理工具中查看消息

進入隊列頁面，能夠看到新建了一個隊列：simple_queue

點擊隊列名稱，進入詳情頁，能夠查看消息：

在控制檯查看消息並不會將消息消費，因此消息還在。

<3>消費者獲取消息

public class Recv {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 建立通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };
        // 監聽隊列，第二個參數：是否自動進行消息確認。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

控制檯：

這個時候，隊列中的消息就沒了：

咱們發現，消費者已經獲取了消息，可是程序沒有中止，一直在監聽隊列中是否有新的消息。一旦有新的消息進入隊列，就會當即打印.

<4>消息確認機制（ACK）

經過剛纔的案例能夠看出，消息一旦被消費者接收，隊列中的消息就會被刪除。

那麼問題來了：RabbitMQ怎麼知道消息被接收了呢？

若是消費者領取消息後，還沒執行操做就掛掉了呢？或者拋出了異常？消息消費失敗，可是RabbitMQ無從得知，這樣消息就丟失了！

所以，RabbitMQ有一個ACK機制。當消費者獲取消息後，會向RabbitMQ發送回執ACK，告知消息已經被接收。不過這種回執ACK分兩種狀況：

自動ACK：消息一旦被接收，消費者自動發送ACK
手動ACK：消息接收後，不會發送ACK，須要手動調用

你們以爲哪一種更好呢？

這須要看消息的重要性：

若是消息不過重要，丟失也沒有影響，那麼自動ACK會比較方便
若是消息很是重要，不容丟失。那麼最好在消費完成後手動ACK，不然接收消息後就自動ACK，RabbitMQ就會把消息從隊列中刪除。若是此時消費者宕機，那麼消息就丟失了。

咱們以前的測試都是自動ACK的，若是要手動ACK，須要改動咱們的代碼：

public class Recv2 {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 建立通道
        final Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
                // 手動進行ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 監聽隊列，第二個參數false，手動進行ACK
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

注意到最後一行代碼：

channel.basicConsume(QUEUE_NAME, false, consumer);

若是第二個參數爲true，則會自動進行ACK；若是爲false，則須要手動ACK。

(2)work消息模型

工做隊列或者競爭消費者模式

工做隊列，又稱任務隊列。主要思想就是避免執行資源密集型任務時，必須等待它執行完成。相反咱們稍後完成任務，咱們將任務封裝爲消息並將其發送到隊列。在後臺運行的工做進程將獲取任務並最終執行做業。當你運行許多消費者時，任務將在他們之間共享，可是一個消息只能被一個消費者獲取。

這個概念在Web應用程序中特別有用，由於在短的HTTP請求窗口中沒法處理複雜的任務。

接下來咱們來模擬這個流程：

P：生產者：任務的發佈者

C1：消費者，領取任務而且完成任務，假設完成速度較快

C2：消費者2：領取任務並完成任務，假設完成速度慢

<1>生產者

生產者與案例1中的幾乎同樣：

public class Send {
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循環發佈任務
        for (int i = 0; i < 50; i++) {
            // 消息內容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

            Thread.sleep(i * 2);
        }
        // 關閉通道和鏈接
        channel.close();
        connection.close();
    }
}

<2>消費者1

與消費者1基本相似，就是沒有設置消費耗時時間。

這裏是模擬有些消費者快，有些比較慢。

咱們可使用basicQos方法和prefetchCount = 1設置。這告訴RabbitMQ一次不要向工做人員發送多於一條消息。或者換句話說，不要向工做人員發送新消息，直到它處理並確認了前一個消息。相反，它會將其分派給不是仍然忙碌的下一個工做人員。

接下來，兩個消費者一同啓動，而後發送50條消息：

能者多勞。

(3)訂閱模型分類

在以前的模式中，咱們建立了一個工做隊列。工做隊列背後的假設是：每一個任務只被傳遞給一個工做人員。在這一部分，咱們將作一些徹底不一樣的事情 - 咱們將會傳遞一個信息給多個消費者。這種模式被稱爲「發佈/訂閱」。

訂閱模型示意圖：

解讀：

　　一、1個生產者，多個消費者

　　二、每個消費者都有本身的一個隊列

　　三、生產者沒有將消息直接發送到隊列，而是發送到了交換機

　　四、每一個隊列都要綁定到交換機

　　五、生產者發送的消息，通過交換機到達隊列，實現一個消息被多個消費者獲取的目的

　　X（Exchanges）：交換機一方面：接收生產者發送的消息。另外一方面：知道如何處理消息，例如遞交給某個特別隊列、遞交給全部隊列、或是將消息丟棄。到底如何操做，取決於Exchange的類型。

Exchange類型有如下幾種：

Fanout：廣播，將消息交給全部綁定到交換機的隊列

Direct：定向，把消息交給符合指定routing key 的隊列 

Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列

Exchange（交換機）只負責轉發消息，不具有存儲消息的能力，所以若是沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那麼消息會丟失！

(4)訂閱模型-Fanout

Fanout，也稱爲廣播。

流程圖：

在廣播模式下，消息發送流程是這樣的：

1）能夠有多個消費者
2）每一個消費者有本身的queue（隊列）
3）每一個隊列都要綁定到Exchange（交換機）
4） 生產者發送的消息，只能發送到交換機，交換機來決定要發給哪一個隊列，生產者沒法決定。
5）交換機把消息發送給綁定過的全部隊列
6）隊列的消費者都能拿到消息。實現一條消息被多個消費者消費

<1>生產者

兩個變化：

1）聲明Exchange，再也不聲明Queue
2）發送消息到Exchange，再也不發送到Queue

public class Send {

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        
        // 聲明exchange，指定類型爲fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        
        // 消息內容
        String message = "Hello everyone";
        // 發佈消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生產者] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

<2>消費者1

public class Recv {
    private final static String QUEUE_NAME = "fanout_exchange_queue_1";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 綁定隊列到交換機
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者1] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

要注意代碼中：隊列須要和交換機綁定

<3>消費者2

public class Recv2 {
    private final static String QUEUE_NAME = "fanout_exchange_queue_2";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 綁定隊列到交換機
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者2] received : " + msg + "!");
            }
        };
        // 監聽隊列，手動返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

<4>測試

咱們運行兩個消費者，而後發送1條消息：

(5)訂閱模型-Direct

有選擇性的接收消息

在訂閱模式中，生產者發佈消息，全部消費者均可以獲取全部消息。

在路由模式中，咱們將添加一個功能 - 咱們將只能訂閱一部分消息。例如，咱們只能將重要的錯誤消息引導到日誌文件（以節省磁盤空間），同時仍然可以在控制檯上打印全部日誌消息。

可是，在某些場景下，咱們但願不一樣的消息被不一樣的隊列消費。這時就要用到Direct類型的Exchange。

在Direct模型下，隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）

消息的發送方在向Exchange發送消息時，也必須指定消息的routing key。

P：生產者，向Exchange發送消息，發送消息時，會指定一個routing key。

X：Exchange（交換機），接收生產者的消息，而後把消息遞交給與routing key徹底匹配的隊列

C1：消費者，其所在隊列指定了須要routing key 爲 error 的消息

C2：消費者，其所在隊列指定了須要routing key 爲 info、error、warning 的消息

<1>生產者

此處咱們模擬商品的增刪改，發送消息的RoutingKey分別是：insert、update、delete

public class Send {
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明exchange，指定類型爲direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息內容
        String message = "商品新增了， id = 1001";
        // 發送消息，而且指定routing key 爲：insert ,表明新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服務：] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

<2>消費者1

咱們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品。

public class Recv {
    private final static String QUEUE_NAME = "direct_exchange_queue_1";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定須要訂閱的routing key。假設此處須要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者1] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

<2>消費者2

咱們此處假設消費者2接收全部類型的消息：新增商品，更新商品和刪除商品

public class Recv2 {
    private final static String QUEUE_NAME = "direct_exchange_queue_2";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 獲取到鏈接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定須要訂閱的routing key。訂閱 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者2] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

<4>測試

咱們分別發送增、刪、改的RoutingKey，發現結果：

(6)訂閱模型-Topic

Topic類型的Exchange與Direct相比，都是能夠根據RoutingKey把消息路由到不一樣的隊列。只不過Topic類型Exchange可讓隊列在綁定Routing key 的時候使用通配符！

Routingkey 通常都是有一個或多個單詞組成，多個單詞之間以」.」分割，例如： item.insert

通配符規則：

`#`：匹配一個或多個詞

`*`：匹配很少很多剛好1個詞

舉例：

`audit.#`：可以匹配`audit.irs.corporate` 或者 `audit.irs`

`audit.*`：只能匹配`audit.irs`

在這個例子中，咱們將發送全部描述動物的消息。消息將使用由三個字（兩個點）組成的routing key發送。路由關鍵字中的第一個單詞將描述速度，第二個顏色和第三個種類：「<speed>.<color>.<species>」。

咱們建立了三個綁定：Q1綁定了綁定鍵「* .orange.」，Q2綁定了「.*.rabbit」和「lazy.＃」。

Q1匹配全部的橙色動物。

Q2匹配關於兔子以及懶惰動物的消息。

(7)持久化

如何避免消息丟失？

1）消費者的ACK機制。能夠防止消費者丟失消息。

2）可是，若是在消費者消費以前，MQ就宕機了，消息就沒了。

是能夠將消息進行持久化呢？

要將消息持久化，前提是：隊列、Exchange都持久化

<1>交換機持久化

<2>隊列持久化

<3>消息持久化

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。