1.RabbitMQ

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

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

• 商品的原始數(shù)據(jù)保存在數(shù)據(jù)庫(kù)中，增刪改查都在數(shù)據(jù)庫(kù)中完成纬朝。
• 搜索服務(wù)數(shù)據(jù)來(lái)源是索引庫(kù)收叶，如果數(shù)據(jù)庫(kù)商品發(fā)生變化，索引庫(kù)數(shù)據(jù)不能及時(shí)更新共苛。
• 商品詳情做了頁(yè)面靜態(tài)化判没，靜態(tài)頁(yè)面數(shù)據(jù)也不會(huì)隨著數(shù)據(jù)庫(kù)商品發(fā)生變化。

如果我們?cè)诤笈_(tái)修改了商品的價(jià)格隅茎，搜索頁(yè)面和商品詳情頁(yè)顯示的依然是舊的價(jià)格澄峰，這樣顯然不對(duì)。該如何解決辟犀？

這里有兩種解決方案：

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

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

所以，我們會(huì)通過(guò)另外一種方式來(lái)解決這個(gè)問(wèn)題：消息隊(duì)列

其他場(chǎng)景

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

1.2.消息隊(duì)列（MQ）

1.2.1.什么是消息隊(duì)列

消息隊(duì)列，即MQ，Message Queue娶聘。

1527063872737.png

消息隊(duì)列是典型的：生產(chǎn)者闻镶、消費(fèi)者模型。生產(chǎn)者不斷向消息隊(duì)列中生產(chǎn)消息丸升，消費(fèi)者不斷的從隊(duì)列中獲取消息铆农。因?yàn)橄⒌纳a(chǎn)和消費(fèi)都是異步的，而且只關(guān)心消息的發(fā)送和接收狡耻，沒有業(yè)務(wù)邏輯的侵入墩剖，這樣就實(shí)現(xiàn)了生產(chǎn)者和消費(fèi)者的解耦。

結(jié)合前面所說(shuō)的問(wèn)題：

• 商品服務(wù)對(duì)商品增刪改以后夷狰，無(wú)需去操作索引庫(kù)或靜態(tài)頁(yè)面岭皂，只是發(fā)送一條消息，也不關(guān)心消息被誰(shuí)接收沼头。
• 搜索服務(wù)和靜態(tài)頁(yè)面服務(wù)接收消息爷绘，分別去處理索引庫(kù)和靜態(tài)頁(yè)面。

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

1.2.2.AMQP和JMS

MQ是消息通信的模型，并不是具體實(shí)現(xiàn)”凶眩現(xiàn)在實(shí)現(xiàn)MQ的有兩種主流方式：AMQP、JMS毡庆。

1527064480681.png

1527064606029.png

1527064487042.png

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

• JMS是定義了統(tǒng)一的接口坑赡，來(lái)對(duì)消息操作進(jìn)行統(tǒng)一；AMQP是通過(guò)規(guī)定協(xié)議來(lái)統(tǒng)一數(shù)據(jù)交互的格式
• JMS限定了必須使用Java語(yǔ)言么抗；AMQP只是協(xié)議毅否，不規(guī)定實(shí)現(xiàn)方式，因此是跨語(yǔ)言的蝇刀。
• JMS規(guī)定了兩種消息模型螟加；而AMQP的消息模型更加豐富

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

1527064606029.png

• ActiveMQ：基于JMS
• RabbitMQ：基于AMQP協(xié)議，erlang語(yǔ)言開發(fā)吞琐，穩(wěn)定性好
• RocketMQ：基于JMS捆探，阿里巴巴產(chǎn)品，目前交由Apache基金會(huì)
• 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種其實(shí)是RPC，并不是MQ奴烙，因此不予學(xué)習(xí)助被。那么也就剩下5種剖张。

但是其實(shí)3、4揩环、5這三種都屬于訂閱模型搔弄，只不過(guò)進(jìn)行路由的方式不同。

1527068544487.png

我們通過(guò)一個(gè)demo工程來(lái)了解下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>

我們抽取一個(gè)建立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è)置賬號(hào)信息顾犹，用戶名、密碼吨枉、vhost
        factory.setVirtualHost("/zq");
        factory.setUsername("zq");
        factory.setPassword("zq");
        // 通過(guò)工程獲取連接
        Connection connection = factory.newConnection();
        return connection;
    }
}

2.1.基本消息模型

官方介紹：

1532762961149.png

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

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

1532762975546.png

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

C（consumer）：消費(fèi)者书在，消費(fèi)和接收有類似的意思灰伟，消費(fèi)者是一個(gè)主要用來(lái)等待接收消息的用戶應(yīng)用程序

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

總之：

生產(chǎn)者將消息發(fā)送到隊(duì)列甚垦，消費(fèi)者從隊(duì)列中獲取消息茶鹃，隊(duì)列是存儲(chǔ)消息的緩沖區(qū)。

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

我們將調(diào)用我們的消息發(fā)布者（發(fā)送者）Send和我們的消息消費(fèi)者（接收者）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)建）隊(duì)列肤无，必須聲明隊(duì)列才能夠發(fā)送消息，我們可以把消息發(fā)送到隊(duì)列中骇钦。
        // 聲明一個(gè)隊(duì)列是冪等的 - 只有當(dāng)它不存在時(shí)才會(huì)被創(chuàng)建
         /**              
        * 參數(shù)明細(xì)：              
        * 1宛渐、隊(duì)列名稱              
        * 2、是否持久化              
        * 3眯搭、是否獨(dú)占此隊(duì)列              
        * 4窥翩、隊(duì)列不用是否自動(dòng)刪除              
        * 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();
    }
}

控制臺(tái)：

1532763328424.png

2.1.2.管理工具中查看消息

進(jìn)入隊(duì)列頁(yè)面鳞仙，可以看到新建了一個(gè)隊(duì)列：simple_queue

1532763817830.png

點(diǎn)擊隊(duì)列名稱寇蚊，進(jìn)入詳情頁(yè)，可以查看消息：

1532763489858.png

在控制臺(tái)查看消息并不會(huì)將消息消費(fèi)棍好，所以消息還在仗岸。

2.1.3.消費(fèi)者獲取消息

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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，并且處理借笙，這個(gè)方法類似事件監(jiān)聽扒怖，如果有消息的時(shí)候，會(huì)被自動(dòng)調(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)聽隊(duì)列业稼，第二個(gè)參數(shù)：是否自動(dòng)進(jìn)行消息確認(rèn)盗痒。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

控制臺(tái)：

1532763733443.png

這個(gè)時(shí)候，隊(duì)列中的消息就沒了：

1532763773208.png

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

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

通過(guò)剛才的案例可以看出，消息一旦被消費(fèi)者接收跨嘉，隊(duì)列中的消息就會(huì)被刪除川慌。

那么問(wèn)題來(lái)了：RabbitMQ怎么知道消息被接收了呢？

如果消費(fèi)者領(lǐng)取消息后祠乃，還沒執(zhí)行操作就掛掉了呢梦重？或者拋出了異常？消息消費(fèi)失敗亮瓷，但是RabbitMQ無(wú)從得知琴拧，這樣消息就丟失了！

因此嘱支，RabbitMQ有一個(gè)ACK機(jī)制蚓胸。當(dāng)消費(fèi)者獲取消息后挣饥，會(huì)向RabbitMQ發(fā)送回執(zhí)ACK，告知消息已經(jīng)被接收沛膳。不過(guò)這種回執(zhí)ACK分兩種情況：

• 自動(dòng)ACK：消息一旦被接收扔枫，消費(fèi)者自動(dòng)發(fā)送ACK
• 手動(dòng)ACK：消息接收后，不會(huì)發(fā)送ACK锹安，需要手動(dòng)調(diào)用

大家覺得哪種更好呢短荐？

這需要看消息的重要性：

• 如果消息不太重要，丟失也沒有影響叹哭，那么自動(dòng)ACK會(huì)比較方便
• 如果消息非常重要忍宋，不容丟失。那么最好在消費(fèi)完成后手動(dòng)ACK风罩，否則接收消息后就自動(dòng)ACK糠排，RabbitMQ就會(huì)把消息從隊(duì)列中刪除。如果此時(shí)消費(fèi)者宕機(jī)泊交，那么消息就丟失了乳讥。

我們之前的測(cè)試都是自動(dòng)ACK的，如果要手動(dòng)ACK廓俭，需要改動(dòng)我們的代碼：

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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息云石，并且處理，這個(gè)方法類似事件監(jiān)聽研乒，如果有消息的時(shí)候汹忠，會(huì)被自動(dòng)調(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 + "!");
                // 手動(dòng)進(jìn)行ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 監(jiān)聽隊(duì)列，第二個(gè)參數(shù)false雹熬，手動(dòng)進(jìn)行ACK
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

注意到最后一行代碼：

channel.basicConsume(QUEUE_NAME, false, consumer);

如果第二個(gè)參數(shù)為true宽菜，則會(huì)自動(dòng)進(jìn)行ACK；如果為false竿报，則需要手動(dòng)ACK铅乡。方法的聲明：

1532764253019.png

2.1.4.1.自動(dòng)ACK存在的問(wèn)題

修改消費(fèi)者，添加異常烈菌，如下：

1532764600849.png

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

1532764694290.png

運(yùn)行消費(fèi)者芽世，程序拋出異常挚赊。但是消息依然被消費(fèi)：

1532764717995.png

管理界面：

1532764734232.png

2.1.4.2.演示手動(dòng)ACK

修改消費(fèi)者，把自動(dòng)改成手動(dòng)（去掉之前制造的異常）

1532764831241.png

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

1532764895239.png

運(yùn)行消費(fèi)者

1532764957092.png

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

1532765013834.png

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

1532765038088.png

這是因?yàn)殡m然我們?cè)O(shè)置了手動(dòng)ACK蔑鹦，但是代碼中并沒有進(jìn)行消息確認(rèn)夺克！所以消息并未被真正消費(fèi)掉。

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

修改代碼手動(dòng)ACK：

1532765123282.png

執(zhí)行：

1532765151039.png

消息消費(fèi)成功懊直！

2.2.work消息模型

工作隊(duì)列或者競(jìng)爭(zhēng)消費(fèi)者模式

1532765197277.png

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

工作隊(duì)列盼铁，又稱任務(wù)隊(duì)列粗蔚。主要思想就是避免執(zhí)行資源密集型任務(wù)時(shí)，必須等待它執(zhí)行完成饶火。相反我們稍后完成任務(wù)鹏控，我們將任務(wù)封裝為消息并將其發(fā)送到隊(duì)列。 在后臺(tái)運(yùn)行的工作進(jìn)程將獲取任務(wù)并最終執(zhí)行作業(yè)肤寝。當(dāng)你運(yùn)行許多工人時(shí)当辐，任務(wù)將在他們之間共享，但是一個(gè)消息只能被一個(gè)消費(fèi)者獲取鲤看。

這個(gè)概念在Web應(yīng)用程序中特別有用缘揪，因?yàn)樵诙痰腍TTP請(qǐng)求窗口中無(wú)法處理復(fù)雜的任務(wù)。

接下來(lái)我們來(lái)模擬這個(gè)流程：

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

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

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

面試題：避免消息堆積慷吊？

1） 采用workqueue袖裕，多個(gè)消費(fèi)者監(jiān)聽同一隊(duì)列。

2）接收到消息以后溉瓶，而是通過(guò)線程池急鳄，異步消費(fèi)。

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();
        // 聲明隊(duì)列
        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();
    }
}

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

2.2.2.消費(fèi)者1

1527085386747.png

2.2.3.消費(fèi)者2

1527085448377.png

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

這里是模擬有些消費(fèi)者快胞锰，有些比較慢灾锯。

接下來(lái)，兩個(gè)消費(fèi)者一同啟動(dòng)嗅榕，然后發(fā)送50條消息：

1527085826462.png

可以發(fā)現(xiàn)顺饮，兩個(gè)消費(fèi)者各自消費(fèi)了25條消息吵聪，而且各不相同，這就實(shí)現(xiàn)了任務(wù)的分發(fā)兼雄。

2.2.4.能者多勞

剛才的實(shí)現(xiàn)有問(wèn)題嗎吟逝？

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

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

怎么實(shí)現(xiàn)呢囱井？

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

1532765689904.png

再次測(cè)試：

1527086159534.png

2.3.訂閱模型分類

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

解讀：

1枫攀、1個(gè)生產(chǎn)者括饶，多個(gè)消費(fèi)者

2、每一個(gè)消費(fèi)者都有自己的一個(gè)隊(duì)列

3来涨、生產(chǎn)者沒有將消息直接發(fā)送到隊(duì)列图焰，而是發(fā)送到了交換機(jī)

4、每個(gè)隊(duì)列都要綁定到交換機(jī)

5蹦掐、生產(chǎn)者發(fā)送的消息技羔，經(jīng)過(guò)交換機(jī)到達(dá)隊(duì)列，實(shí)現(xiàn)一個(gè)消息被多個(gè)消費(fèi)者獲取的目的

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

Exchange類型有以下幾種：

? Fanout：廣播标沪，將消息交給所有綁定到交換機(jī)的隊(duì)列

? Direct：定向榄攀，把消息交給符合指定routing key 的隊(duì)列

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

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

? Fanout：即廣播模式

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

2.4.訂閱模型-Fanout

Fanout，也稱為廣播趁曼。

流程圖：

1527086284940.png

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

• 1） 可以有多個(gè)消費(fèi)者
• 2） 每個(gè)消費(fèi)者有自己的queue（隊(duì)列）
• 3） 每個(gè)隊(duì)列都要綁定到Exchange（交換機(jī)）
• 4） 生產(chǎn)者發(fā)送的消息，只能發(fā)送到交換機(jī)彰阴，交換機(jī)來(lái)決定要發(fā)給哪個(gè)隊(duì)列，生產(chǎn)者無(wú)法決定拍冠。
• 5） 交換機(jī)把消息發(fā)送給綁定過(guò)的所有隊(duì)列
• 6） 隊(duì)列的消費(fèi)者都能拿到消息尿这。實(shí)現(xiàn)一條消息被多個(gè)消費(fèi)者消費(fèi)

2.4.1.生產(chǎn)者

兩個(gè)變化：

• 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、交換機(jī)名稱              
            * 2晃财、交換機(jī)類型叨橱，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.消費(fèi)者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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

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

        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，并且處理钢猛，這個(gè)方法類似事件監(jiān)聽伙菜，如果有消息的時(shí)候，會(huì)被自動(dòng)調(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(" [消費(fèi)者1] received : " + msg + "!");
            }
        };
        // 監(jiān)聽隊(duì)列命迈，自動(dòng)返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

要注意代碼中：隊(duì)列需要和交換機(jī)綁定

2.4.3.消費(fèi)者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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 綁定隊(duì)列到交換機(jī)
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        
        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息贩绕，并且處理，這個(gè)方法類似事件監(jiān)聽壶愤，如果有消息的時(shí)候淑倾，會(huì)被自動(dòng)調(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(" [消費(fèi)者2] received : " + msg + "!");
            }
        };
        // 監(jiān)聽隊(duì)列，手動(dòng)返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

2.4.4.測(cè)試

我們運(yùn)行兩個(gè)消費(fèi)者征椒，然后發(fā)送1條消息：

1532766264386.png

1532766291204.png

2.5.訂閱模型-Direct

有選擇性的接收消息

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

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

但是，在某些場(chǎng)景下溉躲，我們希望不同的消息被不同的隊(duì)列消費(fèi)榜田。這時(shí)就要用到Direct類型的Exchange。

在Direct模型下锻梳，隊(duì)列與交換機(jī)的綁定箭券，不能是任意綁定了，而是要指定一個(gè)RoutingKey（路由key）

消息的發(fā)送方在向Exchange發(fā)送消息時(shí)疑枯，也必須指定消息的routing key辩块。

1532766437787.png

P：生產(chǎn)者，向Exchange發(fā)送消息荆永，發(fā)送消息時(shí)废亭，會(huì)指定一個(gè)routing key。

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

C1：消費(fèi)者，其所在隊(duì)列指定了需要routing key 為 error 的消息

C2：消費(fèi)者骂删，其所在隊(duì)列指定了需要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.消費(fèi)者1

我們此處假設(shè)消費(fèi)者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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊(duì)列到交換機(jī)凭迹，同時(shí)指定需要訂閱的routing key。假設(shè)此處需要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息苦囱，并且處理嗅绸，這個(gè)方法類似事件監(jiān)聽，如果有消息的時(shí)候撕彤，會(huì)被自動(dòng)調(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(" [消費(fèi)者1] received : " + msg + "!");
            }
        };
        // 監(jiān)聽隊(duì)列鱼鸠，自動(dòng)ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

2.5.3.消費(fèi)者2

我們此處假設(shè)消費(fèi)者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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊(duì)列到交換機(jī)蚀狰，同時(shí)指定需要訂閱的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");

        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息跛溉，并且處理，這個(gè)方法類似事件監(jiān)聽扮授，如果有消息的時(shí)候芳室，會(huì)被自動(dòng)調(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(" [消費(fèi)者2] received : " + msg + "!");
            }
        };
        // 監(jiān)聽隊(duì)列，自動(dòng)ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

2.5.4.測(cè)試

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

1527088296131.png

2.6.訂閱模型-Topic

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

Routingkey 一般都是有一個(gè)或多個(gè)單詞組成乏梁，多個(gè)單詞之間以”.”分割雹拄，例如： item.insert

通配符規(guī)則：

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

? *：匹配不多不少恰好1個(gè)詞

舉例：

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

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

1532766712166.png

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

我們創(chuàng)建了三個(gè)綁定：Q1綁定了綁定鍵“* .orange.”翩肌，Q2綁定了“.*.rabbit”和“l(fā)azy.Ｄ０担”。

Q1匹配所有的橙色動(dòng)物念祭。

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

練習(xí)，生產(chǎn)者發(fā)送如下消息粱坤，會(huì)進(jìn)入那個(gè)隊(duì)列：

quick.orange.rabbit à Q1 Q2

lazy.orange.elephant à Q1 Q2

quick.orange.fox à Q1

lazy.pink.rabbit à Q2

quick.brown.fox à 不匹配任意隊(duì)列隶糕，被丟棄

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.消費(fèi)者1

我們此處假設(shè)消費(fèi)者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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊(duì)列到交換機(jī)，同時(shí)指定需要訂閱的routing key耕陷。需要 update名秀、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息兰珍，并且處理锉矢，這個(gè)方法類似事件監(jiān)聽梯嗽，如果有消息的時(shí)候，會(huì)被自動(dòng)調(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(" [消費(fèi)者1] received : " + msg + "!");
            }
        };
        // 監(jiān)聽隊(duì)列沽损，自動(dòng)ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

2.6.3.消費(fèi)者2

我們此處假設(shè)消費(fèi)者2接收所有類型的消息：新增商品灯节，更新商品和刪除商品。

/**
 * 消費(fèi)者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();
        // 聲明隊(duì)列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊(duì)列到交換機(jī)缠俺，同時(shí)指定需要訂閱的routing key显晶。訂閱 insert、update壹士、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");

        // 定義隊(duì)列的消費(fèi)者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息磷雇，并且處理，這個(gè)方法類似事件監(jiān)聽躏救，如果有消息的時(shí)候唯笙，會(huì)被自動(dòng)調(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(" [消費(fèi)者2] received : " + msg + "!");
            }
        };
        // 監(jiān)聽隊(duì)列，自動(dòng)ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

RPC

TIM截圖20190505130354.png

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

1少办、客戶端即是生產(chǎn)者就是消費(fèi)者苞慢，向RPC請(qǐng)求隊(duì)列發(fā)送RPC調(diào)用消息，同時(shí)監(jiān)聽RPC響應(yīng)隊(duì)列英妓。

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

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

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

2.7.持久化

如何避免消息丟失？

1） 消費(fèi)者的ACK機(jī)制敷燎≡蒹荩可以防止消費(fèi)者丟失消息。

2） 但是硬贯，如果在消費(fèi)者消費(fèi)之前乖杠，MQ就宕機(jī)了，消息就沒了澄成。

是可以將消息進(jìn)行持久化呢胧洒？

要將消息持久化畏吓，前提是：隊(duì)列、Exchange都持久化

2.7.1.交換機(jī)持久化

1532766951432.png

2.7.2.隊(duì)列持久化

1532766981230.png

2.7.3.消息持久化

1532767057491.png

3.Spring AMQP

3.1.簡(jiǎn)介

Sprin有很多不同的項(xiàng)目卫漫，其中就有對(duì)AMQP的支持：

1532767136007.png

Spring AMQP的頁(yè)面：http://spring.io/projects/spring-amqp

1532767171063.png

注意這里一段描述：

1532767227821.png

? Spring-amqp是對(duì)AMQP協(xié)議的抽象實(shí)現(xiàn)菲饼，而spring-rabbit 是對(duì)協(xié)議的具體實(shí)現(xiàn)，也是目前的唯一實(shí)現(xiàn)列赎。底層使用的就是RabbitMQ宏悦。

2.2.依賴和配置

添加AMQP的啟動(dòng)器：

<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中，對(duì)消息的消費(fèi)者進(jìn)行了封裝和抽象包吝，一個(gè)普通的JavaBean中的普通方法饼煞，只要通過(guò)簡(jiǎn)單的注解，就可以成為一個(gè)消費(fèi)者诗越。

@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：類上的注解砖瞧，注冊(cè)到Spring容器
• @RabbitListener：方法上的注解，聲明這個(gè)方法是一個(gè)消費(fèi)者方法嚷狞，需要指定下面的屬性：
  • bindings：指定綁定關(guān)系块促，可以有多個(gè)。值是@QueueBinding的數(shù)組床未。@QueueBinding包含下面屬性：
    • value：這個(gè)消費(fèi)者關(guān)聯(lián)的隊(duì)列竭翠。值是@Queue，代表一個(gè)隊(duì)列
    • exchange：隊(duì)列所綁定的交換機(jī)薇搁，值是@Exchange類型
    • key：隊(duì)列和交換機(jī)綁定的RoutingKey

類似listen這樣的方法在一個(gè)類中可以寫多個(gè)斋扰，就代表多個(gè)消費(fèi)者。

2.4.AmqpTemplate

Spring最擅長(zhǎng)的事情就是封裝啃洋，把他人的框架進(jìn)行封裝和整合褥实。

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

1527090258083.png

紅框圈起來(lái)的是比較常用的3個(gè)方法，分別是：

• 指定交換機(jī)哥艇、RoutingKey和消息體
• 指定消息
• 指定RoutingKey和消息绝编，會(huì)向默認(rèn)的交換機(jī)發(fā)送消息

2.5.測(cè)試代碼

@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);
    }
}

運(yùn)行后查看日志：

1532767726274.png

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

3.1.靜態(tài)頁(yè)面

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

• 什么時(shí)候發(fā)？

  當(dāng)商品服務(wù)對(duì)商品進(jìn)行寫操作：增貌踏、刪十饥、改的時(shí)候，需要發(fā)送一條消息祖乳，通知其它服務(wù)逗堵。

• 發(fā)送什么內(nèi)容？

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

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

接收消息后如何處理纸泡？

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

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

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

3.2.1.引入依賴

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

3.2.2.配置文件

我們?cè)赼pplication.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：第一次重試的間隔時(shí)長(zhǎng)
    • max-interval：最長(zhǎng)重試間隔，超過(guò)這個(gè)間隔將不再重試
    • multiplier：下次重試間隔的倍數(shù)女揭，此處是2即下次重試間隔是上次的2倍
  • exchange：缺省的交換機(jī)名稱蚤假，此處配置后，發(fā)送消息如果不指定交換機(jī)就會(huì)使用這個(gè)
• publisher-confirms：生產(chǎn)者確認(rèn)機(jī)制吧兔，確保消息會(huì)正確發(fā)送磷仰，如果發(fā)送失敗會(huì)有錯(cuò)誤回執(zhí)，從而觸發(fā)重試

3.2.3.改造GoodsService

在GoodsService中封裝一個(gè)發(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);
    }
}

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

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

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

1532768930797.png

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

1532769005960.png

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

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

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

因?yàn)樗饕龓?kù)的新增和修改方法是合二為一的欧穴，因此我們可以將這兩類消息一同處理民逼，刪除另外處理。

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)建和刪除索引方法

這里因?yàn)橐獎(jiǎng)?chuàng)建和刪除索引，我們需要在SearchService中拓展兩個(gè)方法调缨，創(chuàng)建和刪除索引：

public void createIndex(Long id) throws IOException {

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

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

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

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

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

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

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

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

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

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

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

1532769581964.png

代碼：

@Component
public class GoodsListener {

    @Autowired
    private GoodsHtmlService goodsHtmlService;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "zq.create.web.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)建頁(yè)面
        goodsHtmlService.createHtml(id);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "zq.delete.web.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;
        }
        // 創(chuàng)建頁(yè)面
        goodsHtmlService.deleteHtml(id);
    }
}

3.4.4.添加刪除頁(yè)面方法

public void deleteHtml(Long id) {
    File file = new File("C:\\project\\nginx-1.14.0\\html\\item\\", id + ".html");
    file.deleteOnExit();
}

3.5.測(cè)試

3.5.1.查看RabbitMQ控制臺(tái)

重新啟動(dòng)項(xiàng)目，并且登錄RabbitMQ管理界面：http://192.168.56.101:15672

可以看到立莉，交換機(jī)已經(jīng)創(chuàng)建出來(lái)了：

1532772446520.png

隊(duì)列也已經(jīng)創(chuàng)建完畢：

1532772471931.png

并且隊(duì)列都已經(jīng)綁定到交換機(jī)：

1532772521932.png