對應B站視頻：

MQ入門-01.MQ課程介紹_嗶哩嗶哩_bilibili

微服務一旦拆分，必然涉及到服務之間的相互調用，目前我們服務之間調用采用的都是基于OpenFeign的調用。這種調用中，調用者發起請求后需要等待服務提供者執行業務返回結果后，才能繼續執行后面的業務。也就是說調用者在調用過程中處于阻塞狀態，因此我們稱這種調用方式為同步調用，也可以叫同步通訊。但在很多場景下，我們可能需要采用異步通訊的方式，為什么呢？

我們先來看看什么是同步通訊和異步通訊。如圖：

解讀：

• 同步通訊：就如同打視頻電話，雙方的交互都是實時的。因此同一時刻你只能跟一個人打視頻電話。

• 異步通訊：就如同發微信聊天，雙方的交互不是實時的，你不需要立刻給對方回應。因此你可以多線操作，同時跟多人聊天。

兩種方式各有優劣，打電話可以立即得到響應，但是你卻不能跟多個人同時通話。發微信可以同時與多個人收發微信，但是往往響應會有延遲。

所以，如果我們的業務需要實時得到服務提供方的響應，則應該選擇同步通訊（同步調用）。而如果我們追求更高的效率，并且不需要實時響應，則應該選擇異步通訊（異步調用）。

同步調用的方式我們已經學過了，之前的OpenFeign調用就是。但是：

• 異步調用又該如何實現？

• 哪些業務適合用異步調用來實現呢？

通過今天的學習你就能明白這些問題了。

1.初識MQ

1.1.同步調用

之前說過，我們現在基于OpenFeign的調用都屬于是同步調用，那么這種方式存在哪些問題呢？

舉個例子，我們以昨天留給大家作為作業的余額支付功能為例來分析，首先看下整個流程：

目前我們采用的是基于OpenFeign的同步調用，也就是說業務執行流程是這樣的：

• 支付服務需要先調用用戶服務完成余額扣減

• 然后支付服務自己要更新支付流水單的狀態

• 然后支付服務調用交易服務，更新業務訂單狀態為已支付

三個步驟依次執行。

這其中就存在3個問題：

第一，拓展性差

我們目前的業務相對簡單，但是隨著業務規模擴大，產品的功能也在不斷完善。

在大多數電商業務中，用戶支付成功后都會以短信或者其它方式通知用戶，告知支付成功。假如后期產品經理提出這樣新的需求，你怎么辦？是不是要在上述業務中再加入通知用戶的業務？

某些電商項目中，還會有積分或金幣的概念。假如產品經理提出需求，用戶支付成功后，給用戶以積分獎勵或者返還金幣，你怎么辦？是不是要在上述業務中再加入積分業務、返還金幣業務？

。。。

最終你的支付業務會越來越臃腫：

也就是說每次有新的需求，現有支付邏輯都要跟著變化，代碼經常變動，不符合開閉原則，拓展性不好。

第二，性能下降

由于我們采用了同步調用，調用者需要等待服務提供者執行完返回結果后，才能繼續向下執行，也就是說每次遠程調用，調用者都是阻塞等待狀態。最終整個業務的響應時長就是每次遠程調用的執行時長之和：

假如每個微服務的執行時長都是50ms，則最終整個業務的耗時可能高達300ms，性能太差了。

第三，級聯失敗

由于我們是基于OpenFeign調用交易服務、通知服務。當交易服務、通知服務出現故障時，整個事務都會回滾，交易失敗。

這其實就是同步調用的級聯失敗問題。

但是大家思考一下，我們假設用戶余額充足，扣款已經成功，此時我們應該確保支付流水單更新為已支付，確保交易成功。畢竟收到手里的錢沒道理再退回去吧。因此，這里不能因為短信通知、更新訂單狀態失敗而回滾整個事務。

綜上，同步調用的方式存在下列問題：

• 拓展性差

• 性能下降

• 級聯失敗

而要解決這些問題，我們就必須用異步調用的方式來代替同步調用。

1.2.異步調用

異步調用方式其實就是基于消息通知的方式，一般包含三個角色：

• 消息發送者：投遞消息的人，就是原來的調用方

• 消息Broker：管理、暫存、轉發消息，你可以把它理解成微信服務器

• 消息接收者：接收和處理消息的人，就是原來的服務提供方

在異步調用中，發送者不再直接同步調用接收者的業務接口，而是發送一條消息投遞給消息Broker。然后接收者根據自己的需求從消息Broker那里訂閱消息。每當發送方發送消息后，接受者都能獲取消息并處理。

這樣，發送消息的人和接收消息的人就完全解耦了。

還是以余額支付業務為例：

除了扣減余額、更新支付流水單狀態以外，其它調用邏輯全部取消。而是改為發送一條消息到Broker。而相關的微服務都可以訂閱消息通知，一旦消息到達Broker，則會分發給每一個訂閱了的微服務，處理各自的業務。

假如產品經理提出了新的需求，比如要在支付成功后更新用戶積分。支付代碼完全不用變更，而僅僅是讓積分服務也訂閱消息即可：

不管后期增加了多少消息訂閱者，作為支付服務來講，執行問扣減余額、更新支付流水狀態后，發送消息即可。業務耗時僅僅是這三部分業務耗時，僅僅100ms，大大提高了業務性能。

另外，不管是交易服務、通知服務，還是積分服務，他們的業務與支付關聯度低。現在采用了異步調用，解除了耦合，他們即便執行過程中出現了故障，也不會影響到支付服務。

綜上，異步調用的優勢包括：

• 耦合度更低

• 性能更好

• 業務拓展性強

• 故障隔離，避免級聯失敗

當然，異步通信也并非完美無缺，它存在下列缺點：

• 完全依賴于Broker的可靠性、安全性和性能

• 架構復雜，后期維護和調試麻煩

1.3.技術選型

消息Broker，目前常見的實現方案就是消息隊列（MessageQueue），簡稱為MQ.

目比較常見的MQ實現：

• ActiveMQ

• RabbitMQ

• RocketMQ

• Kafka

幾種常見MQ的對比：

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社區	Rabbit	Apache	阿里	Apache
開發語言	Erlang	Java	Java	Scala&Java
協議支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定義協議	自定義協議
可用性	高	一般	高	高
單機吞吐量	一般	差	高	非常高
消息延遲	微秒級	毫秒級	毫秒級	毫秒以內
消息可靠性	高	一般	高	一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延遲：RabbitMQ、Kafka

據統計，目前國內消息隊列使用最多的還是RabbitMQ，再加上其各方面都比較均衡，穩定性也好，因此我們課堂上選擇RabbitMQ來學習。

2.RabbitMQ

RabbitMQ是基于Erlang語言開發的開源消息通信中間件，官網地址：

RabbitMQ: One broker to queue them all | RabbitMQ

接下來，我們就學習它的基本概念和基礎用法。

2.1.安裝

我們同樣基于Docker來安裝RabbitMQ，使用下面的命令即可：

docker run \-e RABBITMQ_DEFAULT_USER=itheima \-e RABBITMQ_DEFAULT_PASS=123321 \-v mq-plugins:/plugins \--name mq \--hostname mq \-p 15672:15672 \-p 5672:5672 \--network hm-net\-d \rabbitmq:3.8-management

?如果拉取鏡像困難的話，文章結尾給大家準備好了鏡像，利用docker load命令加載：

可以看到在安裝命令中有兩個映射的端口：

• 15672：RabbitMQ提供的管理控制臺的端口

• 5672：RabbitMQ的消息發送處理接口

安裝完成后，我們訪問 http://192.168.150.101:15672即可看到管理控制臺。首次訪問需要登錄，默認的用戶名和密碼在配置文件中已經指定了。

登錄后即可看到管理控制臺總覽頁面：

RabbitMQ對應的架構如圖：

其中包含幾個概念：

• publisher：生產者，也就是發送消息的一方

• consumer：消費者，也就是消費消息的一方

• queue：隊列，存儲消息。生產者投遞的消息會暫存在消息隊列中，等待消費者處理

• exchange：交換機，負責消息路由。生產者發送的消息由交換機決定投遞到哪個隊列。

• virtual host：虛擬主機，起到數據隔離的作用。每個虛擬主機相互獨立，有各自的exchange、queue

上述這些東西都可以在RabbitMQ的管理控制臺來管理，下一節我們就一起來學習控制臺的使用。

2.2.收發消息

2.2.1.交換機

我們打開Exchanges選項卡，可以看到已經存在很多交換機：

我們點擊任意交換機，即可進入交換機詳情頁面。仍然會利用控制臺中的publish message 發送一條消息：?這里是由控制臺模擬了生產者發送的消息。由于沒有消費者存在，最終消息丟失了，這樣說明交換機沒有存儲消息的能力。

2.2.2.隊列

我們打開Queues選項卡，新建一個隊列：

?

再以相同的方式，創建一個隊列，密碼為hello.queue2，最終隊列列表如下：?

此時，我們再次向amq.fanout交換機發送一條消息。會發現消息依然沒有到達隊列！！

怎么回事呢？

發送到交換機的消息，只會路由到與其綁定的隊列，因此僅僅創建隊列是不夠的，我們還需要將其與交換機綁定。

2.2.3.綁定關系

點擊Exchanges選項卡，點擊amq.fanout交換機，進入交換機詳情頁，然后點擊Bindings菜單，在表單中填寫要綁定的隊列名稱：

相同的方式，將hello.queue2也綁定到改交換機。

最終，綁定結果如下：

?

2.2.4.發送消息

再次回到exchange頁面，找到剛剛綁定的amq.fanout，點擊進入詳情頁，再次發送一條消息：

回到Queues頁面，可以發現hello.queue中已經有一條消息了：

點擊隊列名稱，進入詳情頁，查看隊列詳情，這次我們點擊get message：

可以看到消息到達隊列了：?

這個時候如果有消費者監聽了MQ的hello.queue1或hello.queue2隊列，自然就能接收到消息了。

2.3.數據隔離

2.3.1.用戶管理

點擊Admin選項卡，首先會看到RabbitMQ控制臺的用戶管理界面：

這里的用戶都是RabbitMQ的管理或運維人員。目前只有安裝RabbitMQ時添加的itheima這個用戶。仔細觀察用戶表格中的字段，如下：

• Name：itheima，也就是用戶名

• Tags：administrator，說明itheima用戶是超級管理員，擁有所有權限

• Can access virtual host： /，可以訪問的virtual host，這里的/是默認的virtual host

對于小型企業而言，出于成本考慮，我們通常只會搭建一套MQ集群，公司內的多個不同項目同時使用。這個時候為了避免互相干擾， 我們會利用virtual host的隔離特性，將不同項目隔離。一般會做兩件事情：

• 給每個項目創建獨立的運維賬號，將管理權限分離。

• 給每個項目創建不同的virtual host，將每個項目的數據隔離。

比如，我們給黑馬商城創建一個新的用戶，命名為hmall：

你會發現此時hmall用戶沒有任何virtual host的訪問權限：

接下來我們就來授權。?

2.3.2.virtual host

我們先退出登錄：

切換到剛剛創建的hmall用戶登錄，然后點擊Virtual Hosts菜單，進入virtual host管理頁：

可以看到目前只有一個默認的virtual host，名字為 /。

我們可以給黑馬商城項目創建一個單獨的virtual host，而不是使用默認的/。

創建完成后如圖：

由于我們是登錄hmall賬戶后創建的virtual host，因此回到users菜單，你會發現當前用戶已經具備了對/hmall這個virtual host的訪問權限了：

此時，點擊頁面右上角的virtual host下拉菜單，切換virtual host為 /hmall：?然后再次查看queues選項卡，會發現之前的隊列已經看不到了：

這就是基于virtual host 的隔離效果。

3.SpringAMQP

將來我們開發業務功能的時候，肯定不會在控制臺收發消息，而是應該基于編程的方式。由于RabbitMQ采用了AMQP協議，因此它具備跨語言的特性。任何語言只要遵循AMQP協議收發消息，都可以與RabbitMQ交互。并且RabbitMQ官方也提供了各種不同語言的客戶端。

但是，RabbitMQ官方提供的Java客戶端編碼相對復雜，一般生產環境下我們更多會結合Spring來使用。而Spring的官方剛好基于RabbitMQ提供了這樣一套消息收發的模板工具：SpringAMQP。并且還基于SpringBoot對其實現了自動裝配，使用起來非常方便。

SpringAmqp的官方地址：Spring AMQP

SpringAMQP提供了三個功能：

• 自動聲明隊列、交換機及其綁定關系

• 基于注解的監聽器模式，異步接收消息

• 封裝了RabbitTemplate工具，用于發送消息

這一章我們就一起學習一下，如何利用SpringAMQP實現對RabbitMQ的消息收發。

3.1.導入Demo工程

在文章結尾給大家提供了一個Demo工程，方便我們學習SpringAMQP的使用：

將其復制到你的工作空間，然后用Idea打開，項目結構如圖：

包括三部分：

• mq-demo：父工程，管理項目依賴

• publisher：消息的發送者

• consumer：消息的消費者

在mq-demo這個父工程中，已經配置好了SpringAMQP相關的依賴：

<?xml version="1.0" encoding="UTF-8"?>
<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.demo</groupId><artifactId>mq-demo</artifactId><version>1.0-SNAPSHOT</version><modules><module>publisher</module><module>consumer</module></modules><packaging>pom</packaging><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.12</version><relativePath/></parent><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><!--AMQP依賴，包含RabbitMQ--><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>

因此，子工程中就可以直接使用SpringAMQP了。

3.2.快速入門

在之前的案例中，我們都是經過交換機發送消息到隊列，不過有時候為了測試方便，我們也可以直接向隊列發送消息，跳過交換機。

在入門案例中，我們就演示這樣的簡單模型，如圖：

也就是：

• publisher直接發送消息到隊列

• 消費者監聽并處理隊列中的消息

注意：這種模式一般測試使用，很少在生產中使用。

為了方便測試，我們現在控制臺新建一個隊列：simple.queue?

添加成功：?接下來，我們就可以利用Java代碼收發消息了。

3.2.1.消息發送

首先配置MQ地址，在publisher服務的application.yml中添加配置：

spring:rabbitmq:host: 192.168.22.88 # 你的虛擬機IPport: 5672 # 端口virtual-host: /hmall # 虛擬主機username: hmall # 用戶名password: 123 # 密碼

然后在publisher服務中編寫測試類SpringAmqpTest，并利用RabbitTemplate實現消息發送：

package com.itheima.publisher.amqp;import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSimpleQueue() {// 隊列名稱String queueName = "simple.queue";// 消息String message = "hello, spring amqp!";// 發送消息rabbitTemplate.convertAndSend(queueName, message);}
}

打開控制臺，可以看到消息已經發送到隊列中：

接下來，我們再來實現消息接收。

3.2.2.消息接收

首先配置MQ地址，在consumer服務的application.yml中添加配置：

spring:rabbitmq:host: 192.168.22.88 # 你的虛擬機IPport: 5672 # 端口virtual-host: /hmall # 虛擬主機username: hmall # 用戶名password: 123 # 密碼

然后在consumer服務的com.itheima.consumer.listener包中新建一個類SpringRabbitListener，代碼如下：

package com.itheima.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class SpringRabbitListener {// 利用RabbitListener來聲明要監聽的隊列信息// 將來一旦監聽的隊列中有了消息，就會推送給當前服務，調用當前方法，處理消息。// 可以看到方法體中接收的就是消息體的內容@RabbitListener(queues = "simple.queue")public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println("spring 消費者接收到消息：【" + msg + "】");}
}

3.2.3.測試

啟動consumer服務，然后在publisher服務中運行測試代碼，發送MQ消息。最終consumer收到消息：

3.3.WorkQueues模型

Work queues，任務模型。簡單來說就是讓多個消費者綁定到一個隊列，共同消費隊列中的消息。

當消息處理比較耗時的時候，可能生產消息的速度會遠遠大于消息的消費速度。長此以往，消息就會堆積越來越多，無法及時處理。

此時就可以使用work 模型，多個消費者共同處理消息處理，消息處理的速度就能大大提高了。

接下來，我們就來模擬這樣的場景。

首先，我們在控制臺創建一個新的隊列，命名為work.queue：

3.3.1.消息發送

這次我們循環發送，模擬大量消息堆積現象。

在publisher服務中的SpringAmqpTest類中添加一個測試方法：

/*** workQueue* 向隊列中不停發送消息，模擬消息堆積。*/
@Test
public void testWorkQueue() throws InterruptedException {// 隊列名稱String queueName = "work.queue";// 消息String message = "hello, message_";for (int i = 0; i < 50; i++) {// 發送消息，每20毫秒發送一次，相當于每秒發送50條消息rabbitTemplate.convertAndSend(queueName, message + i);Thread.sleep(20);}
}

3.3.2.消息接收

要模擬多個消費者綁定同一個隊列，我們在consumer服務的SpringRabbitListener中添加2個新的方法：

@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {System.out.println("消費者1接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(20);
}@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {System.err.println("消費者2........接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(200);
}

注意到這兩消費者，都設置了Thead.sleep，模擬任務耗時：

• 消費者1 sleep了20毫秒，相當于每秒鐘處理50個消息

• 消費者2 sleep了200毫秒，相當于每秒處理5個消息

3.3.3.測試

啟動ConsumerApplication后，在執行publisher服務中剛剛編寫的發送測試方法testWorkQueue。

最終結果如下：

消費者2........接收到消息：【hello, message_0】13:27:28.854272300
消費者1接收到消息：【hello, message_1】13:27:28.866843
消費者1接收到消息：【hello, message_3】13:27:28.910834700
消費者1接收到消息：【hello, message_5】13:27:28.955471200
消費者1接收到消息：【hello, message_7】13:27:28.999427900
消費者1接收到消息：【hello, message_9】13:27:29.052203100
消費者2........接收到消息：【hello, message_2】13:27:29.057720700
消費者1接收到消息：【hello, message_11】13:27:29.094622500
消費者1接收到消息：【hello, message_13】13:27:29.134576400
消費者1接收到消息：【hello, message_15】13:27:29.178579
消費者1接收到消息：【hello, message_17】13:27:29.225458300
消費者2........接收到消息：【hello, message_4】13:27:29.258751700
消費者1接收到消息：【hello, message_19】13:27:29.268782600
消費者1接收到消息：【hello, message_21】13:27:29.310326
消費者1接收到消息：【hello, message_23】13:27:29.355328200
消費者1接收到消息：【hello, message_25】13:27:29.404193400
消費者1接收到消息：【hello, message_27】13:27:29.451027300
消費者2........接收到消息：【hello, message_6】13:27:29.462536500
消費者1接收到消息：【hello, message_29】13:27:29.489696700
消費者1接收到消息：【hello, message_31】13:27:29.536279400
消費者1接收到消息：【hello, message_33】13:27:29.579879800
消費者1接收到消息：【hello, message_35】13:27:29.623404100
消費者2........接收到消息：【hello, message_8】13:27:29.665144700
消費者1接收到消息：【hello, message_37】13:27:29.670203800
消費者1接收到消息：【hello, message_39】13:27:29.716523400
消費者1接收到消息：【hello, message_41】13:27:29.762430400
消費者1接收到消息：【hello, message_43】13:27:29.807719700
消費者1接收到消息：【hello, message_45】13:27:29.851229900
消費者2........接收到消息：【hello, message_10】13:27:29.868609700
消費者1接收到消息：【hello, message_47】13:27:29.900501200
消費者1接收到消息：【hello, message_49】13:27:29.943365800
消費者2........接收到消息：【hello, message_12】13:27:30.068846800
消費者2........接收到消息：【hello, message_14】13:27:30.271521800
消費者2........接收到消息：【hello, message_16】13:27:30.471542
消費者2........接收到消息：【hello, message_18】13:27:30.676032400
消費者2........接收到消息：【hello, message_20】13:27:30.878225100
消費者2........接收到消息：【hello, message_22】13:27:31.081266400
消費者2........接收到消息：【hello, message_24】13:27:31.284766100
消費者2........接收到消息：【hello, message_26】13:27:31.487893
消費者2........接收到消息：【hello, message_28】13:27:31.689919200
消費者2........接收到消息：【hello, message_30】13:27:31.892238800
消費者2........接收到消息：【hello, message_32】13:27:32.094483300
消費者2........接收到消息：【hello, message_34】13:27:32.295227700
消費者2........接收到消息：【hello, message_36】13:27:32.498640300
消費者2........接收到消息：【hello, message_38】13:27:32.702505100
消費者2........接收到消息：【hello, message_40】13:27:32.904806600
消費者2........接收到消息：【hello, message_42】13:27:33.107721900
消費者2........接收到消息：【hello, message_44】13:27:33.310408400
消費者2........接收到消息：【hello, message_46】13:27:33.511858400
消費者2........接收到消息：【hello, message_48】13:27:33.712508900

可以看到消費者1和消費者2竟然每人消費了25條消息：

• 消費者1很快完成了自己的25條消息

• 消費者2卻在緩慢的處理自己的25條消息。

也就是說消息是平均分配給每個消費者，并沒有考慮到消費者的處理能力。導致1個消費者空閑，另一個消費者忙的不可開交。沒有充分利用每一個消費者的能力，最終消息處理的耗時遠遠超過了1秒。這樣顯然是有問題的。

3.3.4.能者多勞

在spring中有一個簡單的配置，可以解決這個問題。我們修改consumer服務的application.yml文件，添加配置：

spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能獲取一條消息，處理完成才能獲取下一個消息

再次測試，發現結果如下：

消費者2........接收到消息：【hello, message_0】13:30:19.798834700
消費者1接收到消息：【hello, message_1】13:30:19.817134500
消費者1接收到消息：【hello, message_2】13:30:19.841788
消費者1接收到消息：【hello, message_3】13:30:19.865491
消費者1接收到消息：【hello, message_4】13:30:19.887660700
消費者1接收到消息：【hello, message_5】13:30:19.911784100
消費者1接收到消息：【hello, message_6】13:30:19.935059300
消費者1接收到消息：【hello, message_7】13:30:19.959279200
消費者1接收到消息：【hello, message_8】13:30:19.981466700
消費者2........接收到消息：【hello, message_9】13:30:20.002149200
消費者1接收到消息：【hello, message_10】13:30:20.016352100
消費者1接收到消息：【hello, message_11】13:30:20.039829200
消費者1接收到消息：【hello, message_12】13:30:20.062955
消費者1接收到消息：【hello, message_13】13:30:20.085630800
消費者1接收到消息：【hello, message_14】13:30:20.107801800
消費者1接收到消息：【hello, message_15】13:30:20.134063400
消費者1接收到消息：【hello, message_16】13:30:20.155275300
消費者1接收到消息：【hello, message_17】13:30:20.179395100
消費者1接收到消息：【hello, message_18】13:30:20.200618500
消費者2........接收到消息：【hello, message_19】13:30:20.218727900
消費者1接收到消息：【hello, message_20】13:30:20.243433900
消費者1接收到消息：【hello, message_21】13:30:20.266193400
消費者1接收到消息：【hello, message_22】13:30:20.290405
消費者1接收到消息：【hello, message_23】13:30:20.311534800
消費者1接收到消息：【hello, message_24】13:30:20.336404100
消費者1接收到消息：【hello, message_25】13:30:20.361637200
消費者1接收到消息：【hello, message_26】13:30:20.385411
消費者1接收到消息：【hello, message_27】13:30:20.407505400
消費者2........接收到消息：【hello, message_28】13:30:20.423395300
消費者1接收到消息：【hello, message_29】13:30:20.439001
消費者1接收到消息：【hello, message_30】13:30:20.462126600
消費者1接收到消息：【hello, message_31】13:30:20.485764200
消費者1接收到消息：【hello, message_32】13:30:20.509009700
消費者1接收到消息：【hello, message_33】13:30:20.536239400
消費者1接收到消息：【hello, message_34】13:30:20.559640
消費者1接收到消息：【hello, message_35】13:30:20.583492600
消費者1接收到消息：【hello, message_36】13:30:20.605689700
消費者2........接收到消息：【hello, message_37】13:30:20.628424
消費者1接收到消息：【hello, message_38】13:30:20.643664
消費者1接收到消息：【hello, message_39】13:30:20.668690600
消費者1接收到消息：【hello, message_40】13:30:20.693530500
消費者1接收到消息：【hello, message_41】13:30:20.719437900
消費者1接收到消息：【hello, message_42】13:30:20.743191800
消費者1接收到消息：【hello, message_43】13:30:20.768960500
消費者1接收到消息：【hello, message_44】13:30:20.792175200
消費者1接收到消息：【hello, message_45】13:30:20.817870100
消費者2........接收到消息：【hello, message_46】13:30:20.831696900
消費者1接收到消息：【hello, message_47】13:30:20.840242
消費者1接收到消息：【hello, message_48】13:30:20.863963500
消費者1接收到消息：【hello, message_49】13:30:20.885220800

可以發現，由于消費者1處理速度較快，所以處理了更多的消息；消費者2處理速度較慢，只處理了6條消息。而最終總的執行耗時也在1秒左右，大大提升。

正所謂能者多勞，這樣充分利用了每一個消費者的處理能力，可以有效避免消息積壓問題。

3.3.5.總結

Work模型的使用：

• 多個消費者綁定到一個隊列，同一條消息只會被一個消費者處理

• 通過設置prefetch來控制消費者預取的消息數量

• 3.4.交換機類型

  在之前的兩個測試案例中，都沒有交換機，生產者直接發送消息到隊列。而一旦引入交換機，消息發送的模式會有很大變化：

可以看到，在訂閱模型中，多了一個exchange角色，而且過程略有變化：

• Publisher：生產者，不再發送消息到隊列中，而是發給交換機

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

• Queue：消息隊列也與以前一樣，接收消息、緩存消息。不過隊列一定要與交換機綁定。

• Consumer：消費者，與以前一樣，訂閱隊列，沒有變化

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

交換機的類型有四種：

• Fanout：廣播，將消息交給所有綁定到交換機的隊列。我們最早在控制臺使用的正是Fanout交換機

• Direct：訂閱，基于RoutingKey（路由key）發送給訂閱了消息的隊列

• Topic：通配符訂閱，與Direct類似，只不過RoutingKey可以使用通配符

• Headers：頭匹配，基于MQ的消息頭匹配，用的較少。

課堂中，我們講解前面的三種交換機模式。

3.5.Fanout交換機

Fanout，英文翻譯是扇出，我覺得在MQ中叫廣播更合適。

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

• 1） 可以有多個隊列

• 2） 每個隊列都要綁定到Exchange（交換機）

• 3） 生產者發送的消息，只能發送到交換機

• 4） 交換機把消息發送給綁定過的所有隊列

• 5） 訂閱隊列的消費者都能拿到消息

我們的計劃是這樣的：

• 創建一個名為 hmall.fanout的交換機，類型是Fanout

• 創建兩個隊列fanout.queue1和fanout.queue2，綁定到交換機hmall.fanout

3.5.1.聲明隊列和交換機

在控制臺創建隊列fanout.queue1:

在創建一個隊列fanout.queue2：?然后再創建一個交換機：

?然后綁定兩個隊列到交換機：

?

3.5.2.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

@Test
public void testFanoutExchange() {// 交換機名稱String exchangeName = "hmall.fanout";// 消息String message = "hello, everyone!";rabbitTemplate.convertAndSend(exchangeName, "", message);
}

3.5.3.消息接收

在consumer服務的SpringRabbitListener中添加兩個方法，作為消費者：

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {System.out.println("消費者1接收到Fanout消息：【" + msg + "】");
}@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {System.out.println("消費者2接收到Fanout消息：【" + msg + "】");
}

3.5.4.總結

交換機的作用是什么？

• 接收publisher發送的消息

• 將消息按照規則路由到與之綁定的隊列

• 不能緩存消息，路由失敗，消息丟失

• FanoutExchange的會將消息路由到每個綁定的隊列

3.6.Direct交換機

在Fanout模式中，一條消息，會被所有訂閱的隊列都消費。但是，在某些場景下，我們希望不同的消息被不同的隊列消費。這時就要用到Direct類型的Exchange。

在Direct模型下：

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

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

• Exchange不再把消息交給每一個綁定的隊列，而是根據消息的Routing Key進行判斷，只有隊列的Routingkey與消息的 Routing key完全一致，才會接收到消息

案例需求如圖：

1. 聲明一個名為hmall.direct的交換機

2. 聲明隊列direct.queue1，綁定hmall.direct，bindingKey為blud和red

3. 聲明隊列direct.queue2，綁定hmall.direct，bindingKey為yellow和red

4. 在consumer服務中，編寫兩個消費者方法，分別監聽direct.queue1和direct.queue2

5. 在publisher中編寫測試方法，向hmall.direct發送消息

3.6.1.聲明隊列和交換機

首先在控制臺聲明兩個隊列direct.queue1和direct.queue2，這里不再展示過程：

然后聲明一個direct類型的交換機，命名為hmall.direct:

然后使用red和blue作為key，綁定direct.queue1到hmall.direct：?

同理，使用red和yellow作為key，綁定direct.queue2到hmall.direct，步驟略，最終結果：?

3.6.2.消息接收

在consumer服務的SpringRabbitListener中添加方法：

@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg) {System.out.println("消費者1接收到direct.queue1的消息：【" + msg + "】");
}@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg) {System.out.println("消費者2接收到direct.queue2的消息：【" + msg + "】");
}

3.6.3.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

@Test
public void testSendDirectExchange() {// 交換機名稱String exchangeName = "direct";// 消息String message = "紅色警報！日本亂排核廢水，導致海洋生物變異，驚現哥斯拉！";// 發送消息rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

由于使用的red這個key，所以兩個消費者都收到了消息：

我們再切換為blue這個key：

@Test
public void testSendDirectExchange() {// 交換機名稱String exchangeName = "hmall.direct";// 消息String message = "最新報道，哥斯拉是居民自治巨型氣球，虛驚一場！";// 發送消息rabbitTemplate.convertAndSend(exchangeName, "blue", message);
}

?你會發現，只有消費者1收到了消息：

3.6.4.總結

描述下Direct交換機與Fanout交換機的差異？

• Fanout交換機將消息路由給每一個與之綁定的隊列

• Direct交換機根據RoutingKey判斷路由給哪個隊列

• 如果多個隊列具有相同的RoutingKey，則與Fanout功能類似

3.7.Topic交換機

3.7.1.說明

Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。

只不過Topic類型Exchange可以讓隊列在綁定BindingKey 的時候使用通配符！

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

通配符規則：

• #：匹配一個或多個詞

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

舉例：

• item.#：能夠匹配item.spu.insert 或者 item.spu

• item.*：只能匹配item.spu

圖示：

假如此時publisher發送的消息使用的RoutingKey共有四種：

• china.news 代表有中國的新聞消息；

• china.weather 代表中國的天氣消息；

• japan.news 則代表日本新聞

• japan.weather 代表日本的天氣消息；

解釋：

• topic.queue1：綁定的是china.# ，凡是以 china.開頭的routing key 都會被匹配到，包括：

  • china.news

  • china.weather

• topic.queue2：綁定的是#.news ，凡是以 .news結尾的 routing key 都會被匹配。包括:

  • china.news

  • japan.news

接下來，我們就按照上圖所示，來演示一下Topic交換機的用法。

首先，在控制臺按照圖示例子創建隊列、交換機，并利用通配符綁定隊列和交換機。此處步驟略。最終結果如下：

3.7.2.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

/*** topicExchange*/
@Test
public void testSendTopicExchange() {// 交換機名稱String exchangeName = "hmall.topic";// 消息String message = "喜報！孫悟空大戰哥斯拉，勝!";// 發送消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

?

3.7.3.消息接收

在consumer服務的SpringRabbitListener中添加方法：

@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg){System.out.println("消費者1接收到topic.queue1的消息：【" + msg + "】");
}@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String msg){System.out.println("消費者2接收到topic.queue2的消息：【" + msg + "】");
}

3.7.4.總結

描述下Direct交換機與Topic交換機的差異？

• Topic交換機接收的消息RoutingKey必須是多個單詞，以 . 分割

• Topic交換機與隊列綁定時的bindingKey可以指定通配符

• #：代表0個或多個詞

• *：代表1個詞

3.8.聲明隊列和交換機

在之前我們都是基于RabbitMQ控制臺來創建隊列、交換機。但是在實際開發時，隊列和交換機是程序員定義的，將來項目上線，又要交給運維去創建。那么程序員就需要把程序中運行的所有隊列和交換機都寫下來，交給運維。在這個過程中是很容易出現錯誤的。

因此推薦的做法是由程序啟動時檢查隊列和交換機是否存在，如果不存在自動創建。

3.8.1.基本API

SpringAMQP提供了一個Queue類，用來創建隊列：

SpringAMQP還提供了一個Exchange接口，來表示所有不同類型的交換機：?我們可以自己創建隊列和交換機，不過SpringAMQP還提供了ExchangeBuilder來簡化這個過程：

?而在綁定隊列和交換機時，則需要使用BindingBuilder來創建Binding對象：

3.8.2.fanout示例

在consumer中創建一個類，聲明隊列和交換機：

package com.itheima.consumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class FanoutConfig {/*** 聲明交換機* @return Fanout類型交換機*/@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("hmall.fanout");}/*** 第1個隊列*/@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2個隊列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

3.8.2.direct示例

direct模式由于要綁定多個KEY，會非常麻煩，每一個Key都要編寫一個binding：

package com.itheima.consumer.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DirectConfig {/*** 聲明交換機* @return Direct類型交換機*/@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange("hmall.direct").build();}/*** 第1個隊列*/@Beanpublic Queue directQueue1(){return new Queue("direct.queue1");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("red");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");}/*** 第2個隊列*/@Beanpublic Queue directQueue2(){return new Queue("direct.queue2");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("red");}/*** 綁定隊列和交換機*/@Beanpublic Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");}
}

3.8.4.基于注解聲明

基于@Bean的方式聲明隊列和交換機比較麻煩，Spring還提供了基于注解方式來聲明。

例如，我們同樣聲明Direct模式的交換機和隊列：

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消費者1接收到direct.queue1的消息：【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){System.out.println("消費者2接收到direct.queue2的消息：【" + msg + "】");
}

是不是簡單多了。

再試試Topic模式：

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),key = "china.#"
))
public void listenTopicQueue1(String msg){System.out.println("消費者1接收到topic.queue1的消息：【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),key = "#.news"
))
public void listenTopicQueue2(String msg){System.out.println("消費者2接收到topic.queue2的消息：【" + msg + "】");
}

3.9.消息轉換器

Spring的消息發送代碼接收的消息體是一個Object：

而在數據傳輸時，它會把你發送的消息序列化為字節發送給MQ，接收消息的時候，還會把字節反序列化為Java對象。

只不過，默認情況下Spring采用的序列化方式是JDK序列化。眾所周知，JDK序列化存在下列問題：

• 數據體積過大

• 有安全漏洞

• 可讀性差

我們來測試一下。

3.9.1.測試默認轉換器

1）創建測試隊列

首先，我們在consumer服務中聲明一個新的配置類：

利用@Bean的方式創建一個隊列，

具體代碼：

package com.itheima.consumer.config;import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MessageConfig {@Beanpublic Queue objectQueue() {return new Queue("object.queue");}
}

注意，這里我們先不要給這個隊列添加消費者，我們要查看消息體的格式。

重啟consumer服務以后，該隊列就會被自動創建出來了：

2）發送消息

我們在publisher模塊的SpringAmqpTest中新增一個消息發送的代碼，發送一個Map對象：

@Test
public void testSendMap() throws InterruptedException {// 準備消息Map<String,Object> msg = new HashMap<>();msg.put("name", "柳巖");msg.put("age", 21);// 發送消息rabbitTemplate.convertAndSend("object.queue", msg);
}

?發送消息后查看控制臺：

可以看到消息格式非常不友好。

3.9.2.配置JSON轉換器

顯然，JDK序列化方式并不合適。我們希望消息體的體積更小、可讀性更高，因此可以使用JSON方式來做序列化和反序列化。

在publisher和consumer兩個服務中都引入依賴：

<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>

注意，如果項目中引入了spring-boot-starter-web依賴，則無需再次引入Jackson依賴。

配置消息轉換器，在publisher和consumer兩個服務的啟動類中添加一個Bean即可：

@Bean
public MessageConverter messageConverter(){// 1.定義消息轉換器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自動創建消息id，用于識別不同消息，也可以在業務中基于ID判斷是否是重復消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;
}

消息轉換器中添加的messageId可以便于我們將來做冪等性判斷。

此時，我們到MQ控制臺刪除object.queue中的舊的消息。然后再次執行剛才的消息發送的代碼，到MQ的控制臺查看消息結構：

3.9.3.消費者接收Object

我們在consumer服務中定義一個新的消費者，publisher是用Map發送，那么消費者也一定要用Map接收，格式如下：

@RabbitListener(queues = "object.queue")
public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {System.out.println("消費者接收到object.queue消息：【" + msg + "】");
}