消息解耦 - 簡書

在cn.demo.rabbitmq.util包下我們抽取一個建立RabbitMQ連接的工具類若专，方便其他程序獲取連接：

把這里的服務地址改為自己安裝RabbitMQ的地址陡叠，vhost掰烟，用戶名以及密碼都改為自己開始設置的

public class ConnectionUtil {
    /**
     * 建立與RabbitMQ的連接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception {
        //定義連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //設置服務地址
        factory.setHost("172.16.145.141");
        //端口
        factory.setPort(5672);
        //設置賬號信息痒钝，用戶名痴昧、密碼、vhost
        factory.setVirtualHost("/leyou");
        factory.setUsername("leyou");
        factory.setPassword("123456");
        // 通過工程獲取連接
        Connection connection = factory.newConnection();
        return connection;
    }
}

基本消息模型

官方介紹：

[圖片上傳失敗...(image-de14d1-1592834746729)]

RabbitMQ是一個消息代理：它接受和轉發(fā)消息巨朦。你可以把它想象成一個郵局：當你把郵件放在郵箱里時米丘，你可以確定郵差先生最終會把郵件發(fā)送給你的收件人。在這個比喻中糊啡，RabbitMQ是郵政信箱拄查，郵局和郵遞員。

RabbitMQ與郵局的主要區(qū)別是它不處理紙張棚蓄，而是接受靶累，存儲和轉發(fā)數據消息的二進制數據塊腺毫。

[圖片上傳失敗...(image-55bfe2-1592834746729)]

P（producer/ publisher）：生產者癣疟，一個發(fā)送消息的用戶應用程序挣柬。

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

隊列（紅色區(qū)域）：rabbitmq內部類似于郵箱的一個概念邪蛔。雖然消息流經rabbitmq和你的應用程序，但是它們只能存儲在隊列中扎狱。隊列只受主機的內存和磁盤限制侧到，實質上是一個大的消息緩沖區(qū)。許多生產者可以發(fā)送消息到一個隊列淤击，許多消費者可以嘗試從一個隊列接收數據匠抗。

總之：

生產者將消息發(fā)送到隊列，消費者從隊列中獲取消息污抬，隊列是存儲消息的緩沖區(qū)汞贸。

我們將用Java編寫兩個程序;發(fā)送單個消息的生產者，以及接收消息并將其打印出來的消費者印机。我們將詳細介紹Java API中的一些細節(jié)矢腻，這是一個消息傳遞的“Hello World”。

我們將調用我們的消息發(fā)布者（發(fā)送者）Send和我們的消息消費者（接收者）Recv射赛。發(fā)布者將連接到RabbitMQ多柑，發(fā)送一條消息，然后退出楣责。

生產者發(fā)送消息

運行cn.demo.rabbitmq.simple.Send類的main方法

public class Send {

    private final static String QUEUE_NAME = "simple_queue";

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

        // 聲明（創(chuàng)建）隊列秆麸，必須聲明隊列才能夠發(fā)送消息初嘹，我們可以把消息發(fā)送到隊列中。
        // 聲明一個隊列是冪等的 - 只有當它不存在時才會被創(chuàng)建
        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();
    }
}

控制臺：

1532763328424

管理工具中查看消息

進入隊列頁面蛔屹，可以看到新建了一個隊列：simple_queue

[圖片上傳失敗...(image-d8fdf-1592834746729)]

點擊隊列名稱削樊，進入詳情頁，可以查看消息：

[圖片上傳失敗...(image-ec5b3a-1592834746729)]

在控制臺查看消息并不會將消息消費兔毒，所以消息還在漫贞。

消費者獲取消息

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

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 創(chuàng)建通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，并且處理育叁，這個方法類似事件監(jiān)聽迅脐，如果有消息的時候，會被自動調用
            @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 + "!");
            }
        };
        // 監(jiān)聽隊列豪嗽，第二個參數：是否自動進行消息確認谴蔑。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

控制臺：

1532763733443

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

[圖片上傳失敗...(image-45fd38-1592834746729)]

我們發(fā)現，消費者已經獲取了消息隐锭，但是程序沒有停止窃躲，一直在監(jiān)聽隊列中是否有新的消息。一旦有新的消息進入隊列钦睡，就會立即打印蒂窒。不信那你就在運行一次服務send嘍

消息確認機制（ACK）

通過剛才的案例可以看出，消息一旦被消費者接收荞怒，隊列中的消息就會被刪除洒琢。

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

如果消費者領取消息后褐桌，還沒執(zhí)行操作就掛掉了呢衰抑？或者拋出了異常？消息消費失敗荧嵌，但是RabbitMQ無從得知呛踊，這樣消息就丟失了！

因此完丽，RabbitMQ有一個ACK機制恋技。當消費者獲取消息后，會向RabbitMQ發(fā)送回執(zhí)ACK逻族，告知消息已經被接收蜻底。不過這種回執(zhí)ACK分兩種情況：

自動ACK：消息一旦被接收，消費者自動發(fā)送ACK
手動ACK：消息接收后聘鳞，不會發(fā)送ACK薄辅，需要手動調用

大家覺得哪種更好呢？

這需要看消息的重要性：

如果消息不太重要抠璃，丟失也沒有影響站楚，那么自動ACK會比較方便
如果消息非常重要，不容丟失搏嗡。那么最好在消費完成后手動ACK窿春，否則接收消息后就自動ACK，RabbitMQ就會把消息從隊列中刪除采盒。如果此時消費者宕機旧乞，那么消息就丟失了。

我們之前的測試都是自動ACK的磅氨，如果要手動ACK尺栖，需要改動我們的代碼：

cn.demo.rabbitmq.simple.Recv2

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

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 創(chuàng)建通道
        final Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，并且處理烦租，這個方法類似事件監(jiān)聽延赌，如果有消息的時候除盏，會被自動調用
            @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);
            }
        };
        // 監(jiān)聽隊列，第二個參數false挫以，手動進行ACK
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

注意到最后一行代碼：

channel.basicConsume(QUEUE_NAME, false, consumer);

如果第二個參數為true者蠕，則會自動進行ACK；如果為false屡贺，則需要手動ACK蠢棱。方法的聲明：

[圖片上傳失敗...(image-57db4f-1592834746729)]

自動ACK存在的問題

修改消費者槽奕，添加異常禀忆，如下：

1532764600849

生產者不做任何修改隘弊，直接運行，消息發(fā)送成功：

1532764694290

運行消費者量没，程序拋出異常。但是消息依然被消費：

1532764717995

管理界面：

[圖片上傳失敗...(image-bda982-1592834746729)]

演示手動ACK

修改消費者突想，把自動改成手動（去掉之前制造的異常）

1532764831241

生產者不變殴蹄，再次運行：

[圖片上傳失敗...(image-bb00f3-1592834746729)]

運行消費者

[圖片上傳失敗...(image-badfe3-1592834746729)]

但是，查看管理界面猾担，發(fā)現：

1532765013834

停掉消費者的程序袭灯，發(fā)現：

[圖片上傳失敗...(image-cc8c3f-1592834746729)]

這是因為雖然我們設置了手動ACK，但是代碼中并沒有進行消息確認绑嘹！所以消息并未被真正消費掉稽荧。

當我們關掉這個消費者，消息的狀態(tài)再次稱為Ready

修改代碼手動ACK：

1532765123282

執(zhí)行：

1532765151039

消息消費成功工腋！

work消息模型

cn.demo.rabbitmq.work包下測試類

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

[圖片上傳失敗...(image-b793e4-1592834746729)]

在第一篇教程中姨丈，我們編寫了一個程序，從一個命名隊列中發(fā)送并接受消息擅腰。在這里蟋恬，我們將創(chuàng)建一個工作隊列，在多個工作者之間分配耗時任務趁冈。

工作隊列歼争，又稱任務隊列。主要思想就是避免執(zhí)行資源密集型任務時渗勘，必須等待它執(zhí)行完成沐绒。相反我們稍后完成任務，我們將任務封裝為消息并將其發(fā)送到隊列呀邢。在后臺運行的工作進程將獲取任務并最終執(zhí)行作業(yè)洒沦。當你運行許多消費者時，任務將在他們之間共享价淌，但是一個消息只能被一個消費者獲取申眼。

這個概念在Web應用程序中特別有用瞒津，因為在短的HTTP請求窗口中無法處理復雜的任務。

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

P：生產者：任務的發(fā)布者

C1：消費者括尸，領取任務并且完成任務巷蚪，假設完成速度較快

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

面試題：避免消息堆積濒翻？

采用workqueue屁柏，多個消費者監(jiān)聽同一隊列。
接收到消息以后有送，而是通過線程池淌喻，異步消費。

生產者

生產者與案例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);
        // 循環(huán)發(fā)布任務
        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();
    }
}

不過這里我們是循環(huán)發(fā)送50條消息雀摘。

消費者1

[圖片上傳失敗...(image-8264e3-1592834746729)]

消費者2

1527085448377

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

這里是模擬有些消費者快阵赠，有些比較慢涯塔。

接下來，兩個消費者一同啟動清蚀，然后發(fā)送50條消息：

[圖片上傳失敗...(image-a53eec-1592834746729)]

可以發(fā)現匕荸，兩個消費者各自消費了25條消息，而且各不相同枷邪，這就實現了任務的分發(fā)榛搔。

能者多勞

剛才的實現有問題嗎？

消費者1比消費者2的效率要低齿风，一次任務的耗時較長
然而兩人最終消費的消息數量是一樣的
消費者2大量時間處于空閑狀態(tài)药薯，消費者1一直忙碌

現在的狀態(tài)屬于是把任務平均分配，正確的做法應該是消費越快的人救斑，消費的越多童本。

怎么實現呢？

我們可以使用basicQos方法和prefetchCount = 1設置脸候。這告訴RabbitMQ一次不要向工作人員發(fā)送多于一條消息穷娱。或者換句話說，不要向工作人員發(fā)送新消息运沦，直到它處理并確認了前一個消息泵额。相反，它會將其分派給不是仍然忙碌的下一個工作人員携添。

1532765689904

再次測試：

[圖片上傳失敗...(image-b75611-1592834746729)]

訂閱模型分類

在之前的模式中嫁盲，我們創(chuàng)建了一個工作隊列。工作隊列背后的假設是：每個任務只被傳遞給一個工作人員烈掠。

在這一部分羞秤，我們將做一些完全不同的事情 - 我們將會傳遞一個信息給多個消費者缸托。這種模式被稱為“發(fā)布/訂閱”。

訂閱模型示意圖：

1527086284940

解讀：

1瘾蛋、1個生產者俐镐，多個消費者

2、每一個消費者都有自己的一個隊列

3哺哼、生產者沒有將消息直接發(fā)送到隊列佩抹，而是發(fā)送到了交換機

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

5取董、生產者發(fā)送的消息棍苹，經過交換機到達隊列，實現一個消息被多個消費者獲取的目的

X（Exchanges）：交換機甲葬，一方面：接收生產者發(fā)送的消息廊勃。另一方面：知道如何處理消息，例如遞交給某個特別隊列经窖、遞交給所有隊列、或是將消息丟棄梭灿。到底如何操作画侣，取決于Exchange的類型。

Exchange類型有以下幾種：

Fanout：廣播堡妒，將消息交給所有綁定到交換機的隊列
Direct：定向配乱，把消息交給符合指定routing key 的隊列
Topic：通配符，把消息交給符合routing pattern（路由模式）的隊列

我們這里先學習

Fanout：即廣播模式

Exchange（交換機）只負責轉發(fā)消息皮迟，不具備存儲消息的能力搬泥，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規(guī)則的隊列伏尼，那么消息會丟失忿檩！

訂閱模型-Fanout

Fanout，也稱為廣播爆阶。

流程圖：

1527086564505

在廣播模式下燥透，消息發(fā)送流程是這樣的：

可以有多個消費者
每個消費者有自己的queue（隊列）
每個隊列都要綁定到Exchange（交換機）
生產者發(fā)送的消息，只能發(fā)送到交換機辨图，交換機來決定要發(fā)給哪個隊列班套，生產者無法決定。
交換機把消息發(fā)送給綁定過的所有隊列
隊列的消費者都能拿到消息故河。實現一條消息被多個消費者消費

生產者

兩個變化：

聲明Exchange吱韭，不再聲明Queue
發(fā)送消息到Exchange，不再發(fā)送到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";
        // 發(fā)布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生產者] Sent '" + message + "'");

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

消費者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) {
            // 獲取消息理盆，并且處理痘煤，這個方法類似事件監(jiān)聽，如果有消息的時候熏挎，會被自動調用
            @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 + "!");
            }
        };
        // 監(jiān)聽隊列速勇，自動返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

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

消費者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) {
            // 獲取消息，并且處理坎拐，這個方法類似事件監(jiān)聽烦磁，如果有消息的時候，會被自動調用
            @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 + "!");
            }
        };
        // 監(jiān)聽隊列哼勇，手動返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

可以看出兩個消費者都有自己的隊列都伪，而綁定到的是同一個交換機

測試

首先運行消息生產者，要先聲明交換機积担，即使這一條消息會被丟棄

然后運行兩個消費者陨晶，然后發(fā)送1條消息：

1532766264386

1532766291204

訂閱模型-Direct

有選擇性的接收消息

在訂閱模式中，生產者發(fā)布消息帝璧，所有消費者都可以獲取所有消息先誉。

在路由模式中，我們將添加一個功能 - 我們將只能訂閱一部分消息的烁。例如褐耳，我們只能將重要的錯誤消息引導到日志文件（以節(jié)省磁盤空間），同時仍然能夠在控制臺上打印所有日志消息渴庆。

但是铃芦，在某些場景下，我們希望不同的消息被不同的隊列消費襟雷。這時就要用到Direct類型的Exchange刃滓。

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

消息的發(fā)送方在向Exchange發(fā)送消息時，也必須指定消息的routing key叙赚。

1532766437787

P：生產者老客，向Exchange發(fā)送消息，發(fā)送消息時震叮，會指定一個routing key胧砰。

X：Exchange（交換機），接收生產者的消息苇瓣，然后把消息遞交給與routing key完全匹配的隊列

C1：消費者尉间，其所在隊列指定了需要routing key 為 error 的消息

C2：消費者，其所在隊列指定了需要routing key 為 info、error哲嘲、warning 的消息

生產者

此處我們模擬商品的增刪改贪薪，發(fā)送消息的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";
        // 發(fā)送消息囱怕，并且指定routing key 為：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服務：] Sent '" + message + "'");

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

消費者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) {
            // 獲取消息典格，并且處理，這個方法類似事件監(jiān)聽台丛，如果有消息的時候耍缴，會被自動調用
            @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 + "!");
            }
        };
        // 監(jiān)聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消費者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) {
            // 獲取消息硅蹦，并且處理，這個方法類似事件監(jiān)聽闷煤，如果有消息的時候童芹，會被自動調用
            @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 + "!");
            }
        };
        // 監(jiān)聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

測試

我們分別發(fā)送增鲤拿、刪假褪、改的RoutingKey，發(fā)現結果：

1527088296131

訂閱模型-Topic

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

Routingkey 一般都是有一個或多個單詞組成窒升，多個單詞之間以”.”分割缀遍，例如： item.insert

通配符規(guī)則：

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

`*`：匹配不多不少恰好1個詞

舉例：

`audit.#`：能夠匹配`audit.irs.corporate` 或者 `audit.irs`

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

1532766712166

在這個例子中，我們將發(fā)送所有描述動物的消息饱须。消息將使用由三個字（兩個點）組成的routing key發(fā)送域醇。路由關鍵字中的第一個單詞將描述速度，第二個顏色和第三個種類：速度.顏色.種類。

我們創(chuàng)建了三個綁定：Q1綁定了* .orange.*譬挚，Q2綁定了*.*.rabbit和lazy.＃锅铅。

Q1匹配所有的橙色動物。

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

練習盐须，生產者發(fā)送如下消息，會進入那個隊列：

quick.orange.rabbit Q1 Q2

lazy.orange.elephant Q1 Q2

quick.orange.fox Q1

lazy.pink.rabbit Q2

quick.brown.fox

quick.orange.male.rabbit

orange

生產者

使用topic類型的Exchange漆腌，發(fā)送消息的routing key有3種： item.isnert贼邓、item.update、item.delete：

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

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

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

消費者1

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

public class Recv {
    private final static String QUEUE_NAME = "topic_exchange_queue_1";
    private final static String EXCHANGE_NAME = "topic_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, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，并且處理灌旧，這個方法類似事件監(jiān)聽绑咱，如果有消息的時候，會被自動調用
            @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 + "!");
            }
        };
        // 監(jiān)聽隊列枢泰，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消費者2

我們此處假設消費者2接收所有類型的消息：新增商品描融，更新商品和刪除商品。

/**
 * 消費者2
 */
public class Recv2 {
    private final static String QUEUE_NAME = "topic_exchange_queue_2";
    private final static String EXCHANGE_NAME = "topic_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, "item.*");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息年叮，并且處理，這個方法類似事件監(jiān)聽玻募，如果有消息的時候只损，會被自動調用
            @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 + "!");
            }
        };
        // 監(jiān)聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

持久化

如何避免消息丟失七咧？

消費者的ACK機制跃惫。可以防止消費者丟失消息艾栋。
但是爆存，如果在消費者消費之前，MQ就宕機了裹粤，消息就沒了终蒂。

是可以將消息進行持久化呢蜂林？

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

交換機持久化

1532766951432

隊列持久化

1532766981230

消息持久化

[圖片上傳失敗...(image-6c9c4a-1592834746729)]

Spring AMQP

簡介

Sprin有很多不同的項目噪叙，其中就有對AMQP的支持：

[圖片上傳失敗...(image-429459-1592834746729)]

Spring AMQP的頁面：http://spring.io/projects/spring-amqp

[圖片上傳失敗...(image-aa9185-1592834746729)]

注意這里一段描述：

1532767227821

 Spring-amqp是對AMQP協議的抽象實現，而spring-rabbit 是對協議的具體實現霉翔，也是目前的唯一實現睁蕾。底層使用的就是RabbitMQ。

依賴和配置

添加AMQP的啟動器：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在application.yml中添加RabbitMQ地址：

spring:
  rabbitmq:
    host: 172.16.145.141
    username: leyou
    password: 123456
    virtual-host: /leyou

監(jiān)聽者

在SpringAmqp中债朵，對消息的消費者進行了封裝和抽象子眶，一個普通的JavaBean中的普通方法，只要通過簡單的注解序芦，就可以成為一個消費者臭杰。

@Component
public class Listener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.test.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.test.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}

@Componet：類上的注解，注冊到Spring容器
@RabbitListener：方法上的注解谚中，聲明這個方法是一個消費者方法渴杆，需要指定下面的屬性：
- bindings：指定綁定關系，可以有多個宪塔。值是@QueueBinding的數組磁奖。@QueueBinding包含下面屬性：
  - value：這個消費者關聯的隊列。值是@Queue某筐，代表一個隊列
  - exchange：隊列所綁定的交換機比搭，值是@Exchange類型
  - key：隊列和交換機綁定的RoutingKey

類似listen這樣的方法在一個類中可以寫多個，就代表多個消費者南誊。

AmqpTemplate

Spring最擅長的事情就是封裝身诺，把他人的框架進行封裝和整合。

Spring為AMQP提供了統(tǒng)一的消息處理模板：AmqpTemplate抄囚，非常方便的發(fā)送消息戚长，其發(fā)送方法：

1527090258083

紅框圈起來的是比較常用的3個方法，分別是：

指定交換機怠苔、RoutingKey和消息體
指定消息
指定RoutingKey和消息，會向默認的交換機發(fā)送消息

測試代碼

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg);
        // 等待10秒后再結束
        Thread.sleep(10000);
    }
}

運行后查看日志：

[圖片上傳失敗...(image-8b3094-1592834746729)]

項目改造

接下來仪糖，我們就改造項目柑司，實現搜索服務、商品靜態(tài)頁的數據同步锅劝。

思路分析

發(fā)送方：商品微服務

什么時候發(fā)攒驰？

當商品服務對商品進行寫操作：增、刪故爵、改的時候玻粪，需要發(fā)送一條消息，通知其它服務。
發(fā)送什么內容劲室？

對商品的增刪改時其它服務可能需要新的商品數據伦仍，但是如果消息內容中包含全部商品信息，數據量太大很洋，而且并不是每個服務都需要全部的信息充蓝。因此我們只發(fā)送商品id，其它服務可以根據id查詢自己需要的信息喉磁。

接收方：搜索微服務谓苟、靜態(tài)頁微服務

接收消息后如何處理？

搜索微服務：
- 增/改：添加新的數據到索引庫
- 刪：刪除索引庫數據
靜態(tài)頁微服務：
- 增/改：創(chuàng)建新的靜態(tài)頁
- 刪：刪除原來的靜態(tài)頁

÷我們先在商品微服務leyou-item-service中實現發(fā)送消息协怒。

引入依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置文件

我們在application.yml中添加一些有關RabbitMQ的配置：

spring:
  rabbitmq:
    host: 172.16.145.141
    username: leyou
    password: 123456
    virtual-host: /leyou
    template:
      exchange: leyou.item.exchange
    publisher-confirms: true

template：有關AmqpTemplate的配置
- exchange：缺省的交換機名稱涝焙，此處配置后，發(fā)送消息如果不指定交換機就會使用這個
publisher-confirms：生產者確認機制孕暇，確保消息會正確發(fā)送仑撞，如果發(fā)送失敗會有錯誤回執(zhí)，從而觸發(fā)重試

改造GoodsService

在GoodsService中封裝一個發(fā)送消息到mq的方法：（需要注入AmqpTemplate模板）

private void sendMessage(Long id, String type) {
    try {
        amqpTemplate.convertAndSend("item." + type, id);
    } catch (AmqpException e) {
        LOGGER.error("發(fā)送消息失敗芭商，消息類型：{}派草，商品id：{}", "item." + type, id);
    }
}

這里沒有指定交換機，因此默認發(fā)送到了配置中的：leyou.item.exchange

注意：這里要把所有異常都try起來铛楣，不能讓消息的發(fā)送影響到正常的業(yè)務邏輯

然后在新增的時候調用：

[圖片上傳失敗...(image-c7a11c-1592834746730)]

修改的時候調用：

[圖片上傳失敗...(image-d09827-1592834746730)]

搜索服務接收消息

搜索服務接收到消息后要做的事情：

增：添加新的數據到索引庫
刪：刪除索引庫數據
改：修改索引庫數據

因為索引庫的新增和修改方法是合二為一的近迁，因此我們可以將這兩類消息一同處理，刪除另外處理簸州。

引入依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

添加配置

spring:
  rabbitmq:
    host: 172.16.145.141
    virtual-host: /leyou
    username: leyou
    password: 123456

這里只是接收消息而不發(fā)送鉴竭，所以不用配置template相關內容。

編寫監(jiān)聽器

1532769181819

代碼：

@Component
public class GoodsListener {

    @Autowired
    private SearchService searchService;

    /**
     * 處理insert和update消息
     *
     * @param id
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "LEYOU.CREATE.INDEX.QUEUE", durable = "true"),
                    exchange = @Exchange(value = "LEYOU.ITEM.EXCHANGE", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC),
                    key = {"item.insert", "item.update"}
            )
    )
    public void listenCreate(Long id) throws IOException {
        if (id == null) {
            return;
        }
        // 創(chuàng)建或更新索引
        this.searchService.createIndex(id);
    }

    /**
     * 處理delete消息
     *
     * @param id
     * @throws IOException
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "LEYOU.DELETE.INDEX.QUEUE", durable = "true"),
                    exchange = @Exchange(value = "LEYOU.ITEM.EXCHANGE", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC),
                    key = {"item.delete"}
            )
    )
    public void listenDelete(Long id) throws IOException {
        if (id == null) {
            return;
        }
        // 創(chuàng)建或更新索引
        this.searchService.deleteIndex(id);
    }
}

編寫創(chuàng)建和刪除索引方法

這里因為要創(chuàng)建和刪除索引岸浑，我們需要在SearchService中拓展兩個方法搏存，創(chuàng)建和刪除索引：

@Override
public void createIndex(Long id) throws IOException {
    //查詢spu
    Spu spu = this.goodsClient.querySpuById(id);
    Goods goods = this.buildGoods(spu);
    //保存
    this.goodsRepository.save(goods);
}

@Override
public void deleteIndex(Long id) {
    this.goodsRepository.deleteById(id);
}

創(chuàng)建索引的方法可以從之前導入數據的測試類中拷貝和改造。

靜態(tài)頁服務接收消息

商品靜態(tài)頁服務接收到消息后的處理：

增：創(chuàng)建新的靜態(tài)頁
刪：刪除原來的靜態(tài)頁
改：創(chuàng)建新的靜態(tài)頁并覆蓋原來的

不過矢洲，我們編寫的創(chuàng)建靜態(tài)頁的方法也具備覆蓋以前頁面的功能璧眠，因此：增和改的消息可以放在一個方法中處理，刪除消息放在另一個方法處理读虏。

引入依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

添加配置

spring:
  rabbitmq:
    host: 172.16.145.141
    username: leyou
    password: 123456
    virtual-host: /leyou

這里只是接收消息而不發(fā)送责静，所以不用配置template相關內容。

編寫監(jiān)聽器

[圖片上傳失敗...(image-95a974-1592834746730)]

代碼：

@Component
public class GoodsListener {

    @Autowired
    private GoodsHtmlService goodsHtmlService;

    @Autowired
    private GoodsService goodsService;

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "LEYOU.CREATE.WEB.QUEUE", durable = "true"),
                    exchange = @Exchange(value = "LEYOU.ITEM.EXCHANGE", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC),
                    key = {"item.insert", "item.update"}
            )
    )
    public void listenCreate(Long id) throws Exception {
        if (id == null) {
            return;
        }

        //獲取數據
        Map<String, Object> data = goodsService.loadData(id);
        // 創(chuàng)建頁面
        goodsHtmlService.createHtml(data);
    }

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "LEYOU.DELETE.WEB.QUEUE", durable = "true"),
                    exchange = @Exchange(value = "LEYOU.ITEM.EXCHANGE", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC),
                    key = {"item.delete"}
            )
    )
    public void listenDelete(Long id) {
        if (id == null) {
            return;
        }
        // 刪除頁面
        goodsHtmlService.deleteHtml(id);
    }
}

添加刪除頁面方法

@Override
public void deleteHtml(Long id) {
    File file = new File("/home/cloudlandboy/Project/leyou/html/item/", id + ".html");
    file.deleteOnExit();
}

測試

重新啟動項目盖桥，并且登錄RabbitMQ管理界面

可以看到灾螃，交換機已經創(chuàng)建出來了：

1576379783664

隊列也已經創(chuàng)建完畢：

1576379822464

并且隊列都已經綁定到交換機：

[圖片上傳失敗...(image-cc8ce4-1592834746730)]

在后臺修改商品數據的價格，分別在搜索及商品詳情頁查看是否統(tǒng)一揩徊。