1.RabbitMQ

1.1.搜索與商品服務(wù)的問題

目前我們已經(jīng)完成了商品詳情和搜索系統(tǒng)的開發(fā)。我們思考一下，是否存在問題？

商品的原始數(shù)據(jù)保存在數(shù)據(jù)庫中，增刪改查都在數(shù)據(jù)庫中完成。
搜索服務(wù)數(shù)據(jù)來源是索引庫，如果數(shù)據(jù)庫商品發(fā)生變化，索引庫數(shù)據(jù)不能及時更新。
商品詳情做了頁面靜態(tài)化，靜態(tài)頁面數(shù)據(jù)也不會隨著數(shù)據(jù)庫商品發(fā)生變化。

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

這里有兩種解決方案：

方案1：每當(dāng)后臺對商品做增刪改操作，同時要修改索引庫數(shù)據(jù)及靜態(tài)頁面
方案2：搜索服務(wù)和商品頁面服務(wù)對外提供操作接口，后臺在商品增刪改后，調(diào)用接口

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

所以，我們會通過另外一種方式來解決這個問題：消息隊列

其他場景

例如短信發(fā)送，右鍵發(fā)送，和所有異步但是耗時而且對結(jié)果和執(zhí)行狀態(tài)關(guān)系不大的操作

1.2.消息隊列（MQ）

1.2.1.什么是消息隊列

消息隊列，即MQ，Message Queue。

1527063872737.png

消息隊列是典型的：生產(chǎn)者、消費者模型。生產(chǎn)者不斷向消息隊列中生產(chǎn)消息，消費者不斷的從隊列中獲取消息。因為消息的生產(chǎn)和消費都是異步的，而且只關(guān)心消息的發(fā)送和接收，沒有業(yè)務(wù)邏輯的侵入，這樣就實現(xiàn)了生產(chǎn)者和消費者的解耦。

結(jié)合前面所說的問題：

商品服務(wù)對商品增刪改以后，無需去操作索引庫或靜態(tài)頁面，只是發(fā)送一條消息，也不關(guān)心消息被誰接收。
搜索服務(wù)和靜態(tài)頁面服務(wù)接收消息，分別去處理索引庫和靜態(tài)頁面。

如果以后有其它系統(tǒng)也依賴商品服務(wù)的數(shù)據(jù)，同樣監(jiān)聽消息即可，商品服務(wù)無需任何代碼修改。

1.2.2.AMQP和JMS

MQ是消息通信的模型，并不是具體實現(xiàn)?，F(xiàn)在實現(xiàn)MQ的有兩種主流方式：AMQP、JMS。

1527064480681.png

1527064606029.png

1527064487042.png

兩者間的區(qū)別和聯(lián)系：

JMS是定義了統(tǒng)一的接口，來對消息操作進行統(tǒng)一；AMQP是通過規(guī)定協(xié)議來統(tǒng)一數(shù)據(jù)交互的格式
JMS限定了必須使用Java語言；AMQP只是協(xié)議，不規(guī)定實現(xiàn)方式，因此是跨語言的。
JMS規(guī)定了兩種消息模型；而AMQP的消息模型更加豐富

1.2.3.常見MQ產(chǎn)品

1527064606029.png

ActiveMQ：基于JMS
RabbitMQ：基于AMQP協(xié)議，erlang語言開發(fā)，穩(wěn)定性好
RocketMQ：基于JMS，阿里巴巴產(chǎn)品，目前交由Apache基金會
Kafka：分布式消息系統(tǒng)，高吞吐量

1.2.4.RabbitMQ

RabbitMQ是基于AMQP的一款消息管理系統(tǒng)

官網(wǎng)： http://www.rabbitmq.com/

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

1532758972119.png

1527064762982.png

1.3.下載和安裝

1.3.1.下載

官網(wǎng)下載地址：http://www.rabbitmq.com/download.html

1532759070767.png

目前最新版本是：3.7.5

當(dāng)前使用的是：3.4.1版本

1.3.2.安裝

參考該文章

2.五種消息模型

RabbitMQ提供了6種消息模型，但是第6種其實是RPC，并不是MQ，因此不予學(xué)習(xí)。那么也就剩下5種。

但是其實3、4、5這三種都屬于訂閱模型，只不過進行路由的方式不同。

1527068544487.png

我們通過一個demo工程來了解下RabbitMQ的工作方式：

依賴：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.itcast.rabbitmq</groupId>
    <artifactId>itcast-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.2.RELEASE</version>
    </parent>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
    </dependencies>
</project>

我們抽取一個建立RabbitMQ連接的工具類，方便其他程序獲取連接：

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

2.1.基本消息模型

官方介紹：

1532762961149.png

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

RabbitMQ與郵局的主要區(qū)別是它不處理紙張，而是接受，存儲和轉(zhuǎn)發(fā)數(shù)據(jù)消息的二進制數(shù)據(jù)塊。

1532762975546.png

P（producer/ publisher）：生產(chǎn)者，一個發(fā)送消息的用戶應(yīng)用程序。

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

隊列（紅色區(qū)域）：rabbitmq內(nèi)部類似于郵箱的一個概念。雖然消息流經(jīng)rabbitmq和你的應(yīng)用程序，但是它們只能存儲在隊列中。隊列只受主機的內(nèi)存和磁盤限制，實質(zhì)上是一個大的消息緩沖區(qū)。許多生產(chǎn)者可以發(fā)送消息到一個隊列，許多消費者可以嘗試從一個隊列接收數(shù)據(jù)。

總之：

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

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

我們將調(diào)用我們的消息發(fā)布者（發(fā)送者）Send和我們的消息消費者（接收者）Recv。發(fā)布者將連接到RabbitMQ，發(fā)送一條消息，然后退出。

2.1.1.生產(chǎn)者發(fā)送消息

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ā)送到隊列中。
        // 聲明一個隊列是冪等的 - 只有當(dāng)它不存在時才會被創(chuàng)建
         /**              
        * 參數(shù)明細(xì)：              
        * 1、隊列名稱              
        * 2、是否持久化              
        * 3、是否獨占此隊列              
        * 4、隊列不用是否自動刪除              
        * 5、參數(shù)              
        */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

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

        //關(guān)閉通道和連接
        channel.close();
        connection.close();
    }
}

控制臺：

1532763328424.png

2.1.2.管理工具中查看消息

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

1532763817830.png

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

1532763489858.png

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

2.1.3.消費者獲取消息

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)聽，如果有消息的時候，會被自動調(diào)用
            @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)聽隊列，第二個參數(shù)：是否自動進行消息確認(rèn)。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

控制臺：

1532763733443.png

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

1532763773208.png

我們發(fā)現(xiàn)，消費者已經(jīng)獲取了消息，但是程序沒有停止，一直在監(jiān)聽隊列中是否有新的消息。一旦有新的消息進入隊列，就會立即打印.

2.1.4.消息確認(rèn)機制（ACK）

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

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

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

因此，RabbitMQ有一個ACK機制。當(dāng)消費者獲取消息后，會向RabbitMQ發(fā)送回執(zhí)ACK，告知消息已經(jīng)被接收。不過這種回執(zhí)ACK分兩種情況：

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

大家覺得哪種更好呢？

這需要看消息的重要性：

如果消息不太重要，丟失也沒有影響，那么自動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();
        // 創(chuàng)建通道
        final Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，并且處理，這個方法類似事件監(jiān)聽，如果有消息的時候，會被自動調(diào)用
            @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)聽隊列，第二個參數(shù)false，手動進行ACK
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

注意到最后一行代碼：

channel.basicConsume(QUEUE_NAME, false, consumer);

如果第二個參數(shù)為true，則會自動進行ACK；如果為false，則需要手動ACK。方法的聲明：

1532764253019.png

2.1.4.1.自動ACK存在的問題

修改消費者，添加異常，如下：

1532764600849.png

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

1532764694290.png

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

1532764717995.png

管理界面：

1532764734232.png

2.1.4.2.演示手動ACK

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

1532764831241.png

生產(chǎn)者不變，再次運行：

1532764895239.png

運行消費者

1532764957092.png

但是，查看管理界面，發(fā)現(xiàn)：

1532765013834.png

停掉消費者的程序，發(fā)現(xiàn)：

1532765038088.png

這是因為雖然我們設(shè)置了手動ACK，但是代碼中并沒有進行消息確認(rèn)！所以消息并未被真正消費掉。

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

修改代碼手動ACK：

1532765123282.png

執(zhí)行：

1532765151039.png

消息消費成功！

2.2.work消息模型

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

1532765197277.png

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

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

這個概念在Web應(yīng)用程序中特別有用，因為在短的HTTP請求窗口中無法處理復(fù)雜的任務(wù)。

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

? P：生產(chǎn)者：任務(wù)的發(fā)布者

? C1：消費者，領(lǐng)取任務(wù)并且完成任務(wù)，假設(shè)完成速度較快

? C2：消費者2：領(lǐng)取任務(wù)并完成任務(wù)，假設(shè)完成速度慢

面試題：避免消息堆積？

1）采用workqueue，多個消費者監(jiān)聽同一隊列。

2）接收到消息以后，而是通過線程池，異步消費。

2.2.1.生產(chǎn)者

生產(chǎ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ā)布任務(wù)
        for (int i = 0; i < 50; i++) {
            // 消息內(nèi)容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

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

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

2.2.2.消費者1

1527085386747.png

2.2.3.消費者2

1527085448377.png

與消費者1基本類似，就是沒有設(shè)置消費耗時時間。

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

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

1527085826462.png

可以發(fā)現(xiàn)，兩個消費者各自消費了25條消息，而且各不相同，這就實現(xiàn)了任務(wù)的分發(fā)。

2.2.4.能者多勞

剛才的實現(xiàn)有問題嗎？

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

現(xiàn)在的狀態(tài)屬于是把任務(wù)平均分配，正確的做法應(yīng)該是消費越快的人，消費的越多。

怎么實現(xiàn)呢？

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

1532765689904.png

再次測試：

1527086159534.png

2.3.訂閱模型分類

在之前的模式中，我們創(chuàng)建了一個工作隊列。工作隊列背后的假設(shè)是：每個任務(wù)只被傳遞給一個工作人員。在這一部分，我們將做一些完全不同的事情 - 我們將會傳遞一個信息給多個消費者。這種模式被稱為“發(fā)布/訂閱”。

解讀：

1、1個生產(chǎn)者，多個消費者

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

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

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

5、生產(chǎn)者發(fā)送的消息，經(jīng)過交換機到達隊列，實現(xiàn)一個消息被多個消費者獲取的目的

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

Exchange類型有以下幾種：

? Fanout：廣播，將消息交給所有綁定到交換機的隊列

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

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

我們這里先學(xué)習(xí)

? Fanout：即廣播模式

Exchange（交換機）只負(fù)責(zé)轉(zhuǎn)發(fā)消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規(guī)則的隊列，那么消息會丟失！

2.4.訂閱模型-Fanout

Fanout，也稱為廣播。

流程圖：

1527086284940.png

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

1）可以有多個消費者
2）每個消費者有自己的queue（隊列）
3）每個隊列都要綁定到Exchange（交換機）
4） 生產(chǎn)者發(fā)送的消息，只能發(fā)送到交換機，交換機來決定要發(fā)給哪個隊列，生產(chǎn)者無法決定。
5）交換機把消息發(fā)送給綁定過的所有隊列
6）隊列的消費者都能拿到消息。實現(xiàn)一條消息被多個消費者消費

2.4.1.生產(chǎn)者

兩個變化：

1）聲明Exchange，不再聲明Queue
2）發(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
        /**              
        * 參數(shù)明細(xì)
            * 1、交換機名稱              
            * 2、交換機類型，fanout、topic、direct、headers     
            BuiltinExchangeType.TOPIC         
        */ 
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        
        // 消息內(nèi)容
        String message = "Hello everyone";
        // 發(fā)布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生產(chǎn)者] Sent '" + message + "'");

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

2.4.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) {
            // 獲取消息，并且處理，這個方法類似事件監(jiān)聽，如果有消息的時候，會被自動調(diào)用
            @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.4.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) {
            // 獲取消息，并且處理，這個方法類似事件監(jiān)聽，如果有消息的時候，會被自動調(diào)用
            @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);
    }
}

2.4.4.測試

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

1532766264386.png

1532766291204.png

2.5.訂閱模型-Direct

有選擇性的接收消息

在訂閱模式中，生產(chǎn)者發(fā)布消息，所有消費者都可以獲取所有消息。

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

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

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

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

1532766437787.png

P：生產(chǎn)者，向Exchange發(fā)送消息，發(fā)送消息時，會指定一個routing key。

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

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

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

2.5.1.生產(chǎn)者

此處我們模擬商品的增刪改，發(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");
        // 消息內(nèi)容
        String message = "商品新增了， id = 1001";
        // 發(fā)送消息，并且指定routing key 為：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服務(wù)：] Sent '" + message + "'");

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

2.5.2.消費者1

我們此處假設(shè)消費者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。假設(shè)此處需要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，并且處理，這個方法類似事件監(jiān)聽，如果有消息的時候，會被自動調(diào)用
            @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.5.3.消費者2

我們此處假設(shè)消費者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)聽，如果有消息的時候，會被自動調(diào)用
            @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);
    }
}

2.5.4.測試

我們分別發(fā)送增、刪、改的RoutingKey，發(fā)現(xiàn)結(jié)果：

1527088296131.png

2.6.訂閱模型-Topic

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

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

通配符規(guī)則：

? #：匹配一個或多個詞(注意不是一個字母，是完整的一個單詞)

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

舉例：

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

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

1532766712166.png

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

我們創(chuàng)建了三個綁定：Q1綁定了綁定鍵“* .orange.”，Q2綁定了“.*.rabbit”和“l(fā)azy.?！薄?/p>

Q1匹配所有的橙色動物。

Q2匹配關(guān)于兔子以及懶惰動物的消息。

練習(xí)，生產(chǎn)者發(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 à

2.6.1.生產(chǎn)者

使用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");
        // 消息內(nèi)容
        String message = "新增商品 : id = 1001";
        // 發(fā)送消息，并且指定routing key 為：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
        System.out.println(" [商品服務(wù)：] Sent '" + message + "'");

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

2.6.2.消費者1

我們此處假設(shè)消費者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)聽，如果有消息的時候，會被自動調(diào)用
            @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.6.3.消費者2

我們此處假設(shè)消費者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)聽，如果有消息的時候，會被自動調(diào)用
            @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);
    }
}

RPC

TIM截圖20190505130354.png

RPC即客戶端遠程調(diào)用服務(wù)端的方法，使用MQ可以實現(xiàn)RPC的異步調(diào)用，基于Direct交換機實現(xiàn)，流程如下：

1、客戶端即是生產(chǎn)者就是消費者，向RPC請求隊列發(fā)送RPC調(diào)用消息，同時監(jiān)聽RPC響應(yīng)隊列。

2、服務(wù)端監(jiān)聽RPC請求隊列的消息，收到消息后執(zhí)行服務(wù)端的方法，得到方法返回的結(jié)果

3、服務(wù)端將RPC方法的結(jié)果發(fā)送到RPC響應(yīng)隊列

4、客戶端（RPC調(diào)用方）監(jiān)聽RPC響應(yīng)隊列，接收到RPC調(diào)用結(jié)果。

2.7.持久化

如何避免消息丟失？

1）消費者的ACK機制。可以防止消費者丟失消息。

2）但是，如果在消費者消費之前，MQ就宕機了，消息就沒了。

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

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

2.7.1.交換機持久化

1532766951432.png

2.7.2.隊列持久化

1532766981230.png

2.7.3.消息持久化

1532767057491.png

3.Spring AMQP

3.1.簡介

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

1532767136007.png

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

1532767171063.png

注意這里一段描述：

1532767227821.png

? Spring-amqp是對AMQP協(xié)議的抽象實現(xiàn)，而spring-rabbit 是對協(xié)議的具體實現(xiàn)，也是目前的唯一實現(xiàn)。底層使用的就是RabbitMQ。

2.2.依賴和配置

添加AMQP的啟動器：

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

在application.yml中添加RabbitMQ地址：

spring:
  rabbitmq:
    host: 192.168.56.101
    username: zq
    password: zq
    virtual-host: /zq

2.3.監(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：指定綁定關(guān)系，可以有多個。值是@QueueBinding的數(shù)組。@QueueBinding包含下面屬性：
  - value：這個消費者關(guān)聯(lián)的隊列。值是@Queue，代表一個隊列
  - exchange：隊列所綁定的交換機，值是@Exchange類型
  - key：隊列和交換機綁定的RoutingKey

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

2.4.AmqpTemplate

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

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

1527090258083.png

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

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

2.5.測試代碼

@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秒后再結(jié)束
        Thread.sleep(10000);
    }
}

運行后查看日志：

1532767726274.png

3.應(yīng)用場景(此處是頁面靜態(tài)化)

3.1.靜態(tài)頁面

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

什么時候發(fā)？

當(dāng)商品服務(wù)對商品進行寫操作：增、刪、改的時候，需要發(fā)送一條消息，通知其它服務(wù)。
發(fā)送什么內(nèi)容？

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

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

接收消息后如何處理？

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

3.2.商品服務(wù)發(fā)送消息

我們先在商品微服務(wù)zq-item-service中實現(xiàn)發(fā)送消息。

3.2.1.引入依賴

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

3.2.2.配置文件

我們在application.yml中添加一些有關(guān)RabbitMQ的配置：

spring:
  rabbitmq:
    host: 192.168.56.101
    username: zq
    password: zq
    virtual-host: /zq
    template:
      exchange: zq.item.exchange
    publisher-confirms: true

template：有關(guān)AmqpTemplate的配置
- retry：失敗重試
  - enabled：開啟失敗重試
  - initial-interval：第一次重試的間隔時長
  - max-interval：最長重試間隔，超過這個間隔將不再重試
  - multiplier：下次重試間隔的倍數(shù)，此處是2即下次重試間隔是上次的2倍
- exchange：缺省的交換機名稱，此處配置后，發(fā)送消息如果不指定交換機就會使用這個
publisher-confirms：生產(chǎn)者確認(rèn)機制，確保消息會正確發(fā)送，如果發(fā)送失敗會有錯誤回執(zhí)，從而觸發(fā)重試

3.2.3.改造GoodsService

在GoodsService中封裝一個發(fā)送消息到mq的方法：

private void sendMessage(Long id, String type){
    // 發(fā)送消息
    try {
        this.amqpTemplate.convertAndSend("item." + type, id);
    } catch (Exception e) {
        logger.error("{}商品消息發(fā)送異常，商品id：{}", type, id, e);
    }
}

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

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

然后在新增的時候調(diào)用：

1532768930797.png

修改的時候調(diào)用：

1532769005960.png

3.3.搜索服務(wù)接收消息

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

增：添加新的數(shù)據(jù)到索引庫
刪：刪除索引庫數(shù)據(jù)
改：修改索引庫數(shù)據(jù)

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

3.3.1.引入依賴

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

3.3.2.添加配置

spring:
  rabbitmq:
    host: 192.168.56.101
    username: zq
    password: zq
    virtual-host: /zq

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

3.3.3.編寫監(jiān)聽器

1532769181819.png

代碼：

@Component
public class GoodsListener {

    @Autowired
    private SearchService searchService;

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

    /**
     * 處理delete的消息
     *
     * @param id
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "zq.delete.index.queue", durable = "true"),
            exchange = @Exchange(
                    value = "zq.item.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC),
            key = "item.delete"))
    public void listenDelete(Long id) {
        if (id == null) {
            return;
        }
        // 刪除索引
        this.searchService.deleteIndex(id);
    }
}

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

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

public void createIndex(Long id) throws IOException {

    Spu spu = this.goodsClient.querySpuById(id);
    // 構(gòu)建商品
    Goods goods = this.buildGoods(spu);

    // 保存數(shù)據(jù)到索引庫
    this.goodsRepository.save(goods);
}

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

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

3.4.靜態(tài)頁服務(wù)接收消息

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

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

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

3.4.1.引入依賴

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

3.4.2.添加配置

spring:
  rabbitmq:
    host: 192.168.56.101
    username: zq
    password: zq
    virtual-host: /zq