rabbitmq及數據同步

1.RabbitMQ

1.1.搜索與商品服務的問題

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

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

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

這里有兩種解決方案：

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

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

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

1.2.消息隊列（MQ）

1.2.1.什么是消息隊列

消息隊列，即MQ，Message Queue。

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

結合前面所說的問題：

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

如果以后有其它系統也依賴商品服務的數據，同樣監聽消息即可，商品服務無需任何代碼修改。

1.2.2.AMQP和JMS

MQ是消息通信的模型，並不是具體實現。現在實現MQ的有兩種主流方式：AMQP、JMS。

兩者間的區別和聯系：

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

1.2.3.常見MQ產品

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

1.2.4.RabbitMQ

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

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

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

1.3.下載和安裝

1.3.1.下載

官網下載地址：http://www.rabbitmq.com/download.html

目前最新版本是：3.7.5

我們的課程中使用的是：3.4.1版本

課前資料提供了安裝包：

1.3.2.安裝

詳見課前資料中的：

2.五種消息模型

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

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

我們通過一個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(); //設置服務地址 factory.setHost("192.168.56.101"); //端口 factory.setPort(5672); //設置賬號信息，用戶名、密碼、vhost factory.setVirtualHost("/leyou"); factory.setUsername("leyou"); factory.setPassword("leyou"); // 通過工程獲取連接 Connection connection = factory.newConnection(); return connection; } }

2.1.基本消息模型

官方介紹：

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

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

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

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

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

總之：

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

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

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

2.1.1.生產者發送消息

public class Send { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] argv) throws Exception { // 獲取到連接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 從連接中創建通道，這是完成大部分API的地方。 Channel channel = connection.createChannel(); // 聲明（創建）隊列，必須聲明隊列才能夠發送消息，我們可以把消息發送到隊列中。 // 聲明一個隊列是冪等的 - 只有當它不存在時才會被創建 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 消息內容 String message = "Hello World!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); //關閉通道和連接 channel.close(); connection.close(); } }

控制台：

2.1.2.管理工具中查看消息

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

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

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

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(); // 創建通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定義隊列的消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [x] received : " + msg + "!"); } }; // 監聽隊列，第二個參數：是否自動進行消息確認。 channel.basicConsume(QUEUE_NAME, true, consumer); } }

控制台：

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

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

2.1.4.消息確認機制（ACK）

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

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

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

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

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

大家覺得哪種更好呢？

這需要看消息的重要性：

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

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

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

注意到最后一行代碼：

channel.basicConsume(QUEUE_NAME, false, consumer);

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

2.1.4.1.自動ACK存在的問題

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

生產者不做任何修改，直接運行，消息發送成功：

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

管理界面：

2.1.4.2.演示手動ACK

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

生產者不變，再次運行：

運行消費者

但是，查看管理界面，發現：

停掉消費者的程序，發現：

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

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

修改代碼手動ACK：

執行：

消息消費成功！

2.2.work消息模型

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

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

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

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

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

​ P：生產者：任務的發布者

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

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

面試題：避免消息堆積？

1） 采用workqueue，多個消費者監聽同一隊列。

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

2.2.1.生產者

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

public class Send { private final static String QUEUE_NAME = "test_work_queue"; public static void main(String[] argv) throws Exception { // 獲取到連接 Connection connection = ConnectionUtil.getConnection(); // 獲取通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 循環發布任務 for (int i = 0; i < 50; i++) { // 消息內容 String message = "task .. " + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); Thread.sleep(i * 2); } // 關閉通道和連接 channel.close(); connection.close(); } }

不過這里我們是循環發送50條消息。

2.2.2.消費者1

2.2.3.消費者2

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

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

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

可以發現，兩個消費者各自消費了25條消息，而且各不相同，這就實現了任務的分發。

2.2.4.能者多勞

剛才的實現有問題嗎？

• 消費者1比消費者2的效率要低，一次任務的耗時較長
• 然而兩人最終消費的消息數量是一樣的
• 消費者2大量時間處於空閑狀態，消費者1一直忙碌

現在的狀態屬於是把任務平均分配，正確的做法應該是消費越快的人，消費的越多。

怎么實現呢？

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

再次測試：

2.3.訂閱模型分類

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

訂閱模型示意圖：

解讀：

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

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

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

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

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

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

Exchange類型有以下幾種：

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

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

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

我們這里先學習

​ Fanout：即廣播模式

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

2.4.訂閱模型-Fanout

Fanout，也稱為廣播。

流程圖：

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

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

2.4.1.生產者

兩個變化：

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

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

2.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) { // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者1] received : " + msg + "!"); } }; // 監聽隊列，自動返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }

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

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) { // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者2] received : " + msg + "!"); } }; // 監聽隊列，手動返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }

2.4.4.測試

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

2.5.訂閱模型-Direct

有選擇性的接收消息

在訂閱模式中，生產者發布消息，所有消費者都可以獲取所有消息。

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

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

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

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

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

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

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

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

2.5.1.生產者

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

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

2.5.2.消費者1

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

public class Recv { private final static String QUEUE_NAME = "direct_exchange_queue_1"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 獲取到連接 Connection connection = ConnectionUtil.getConnection(); // 獲取通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 綁定隊列到交換機，同時指定需要訂閱的routing key。假設此處需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定義隊列的消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者1] received : " + msg + "!"); } }; // 監聽隊列，自動ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }

2.5.3.消費者2

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

public class Recv2 { private final static String QUEUE_NAME = "direct_exchange_queue_2"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 獲取到連接 Connection connection = ConnectionUtil.getConnection(); // 獲取通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定義隊列的消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者2] received : " + msg + "!"); } }; // 監聽隊列，自動ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }

2.5.4.測試

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

2.6.訂閱模型-Topic

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

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

通配符規則：

​ #：匹配一個或多個詞

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

舉例：

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

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

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

我們創建了三個綁定：Q1綁定了綁定鍵“* .orange.”，Q2綁定了“.*.rabbit”和“lazy.＃”。

Q1匹配所有的橙色動物。

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

練習，生產者發送如下消息，會進入那個隊列：

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.生產者

使用topic類型的Exchange，發送消息的routing key有3種： item.isnert、item.update、item.delete：

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

2.6.2.消費者1

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

public class Recv { private final static String QUEUE_NAME = "topic_exchange_queue_1"; private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 獲取到連接 Connection connection = ConnectionUtil.getConnection(); // 獲取通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 綁定隊列到交換機，同時指定需要訂閱的routing key。需要 update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete"); // 定義隊列的消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者1] received : " + msg + "!"); } }; // 監聽隊列，自動ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }

2.6.3.消費者2

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

/**
 * 消費者2  */ public class Recv2 { private final static String QUEUE_NAME = "topic_exchange_queue_2"; private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 獲取到連接 Connection connection = ConnectionUtil.getConnection(); // 獲取通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*"); // 定義隊列的消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者2] received : " + msg + "!"); } }; // 監聽隊列，自動ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }

2.7.持久化

如何避免消息丟失？

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

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

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

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

2.7.1.交換機持久化

2.7.2.隊列持久化

2.7.3.消息持久化

3.Spring AMQP

3.1.簡介

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

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

注意這里一段描述：

​ Spring-amqp是對AMQP協議的抽象實現，而spring-rabbit 是對協議的具體實現，也是目前的唯一實現。底層使用的就是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: leyou password: leyou virtual-host: /leyou

2.3.監聽者

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

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

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

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

2.4.AmqpTemplate

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

Spring為AMQP提供了統一的消息處理模板：AmqpTemplate，非常方便的發送消息，其發送方法：

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

• 指定交換機、RoutingKey和消息體
• 指定消息
• 指定RoutingKey和消息，會向默認的交換機發送消息

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

運行后查看日志：

3.項目改造

接下來，我們就改造項目，實現搜索服務、商品靜態頁的數據同步。

3.1.思路分析

發送方：商品微服務

• 什么時候發？

  當商品服務對商品進行寫操作：增、刪、改的時候，需要發送一條消息，通知其它服務。

• 發送什么內容？

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

接收方：搜索微服務、靜態頁微服務

接收消息后如何處理？

• 搜索微服務：
  • 增/改：添加新的數據到索引庫
  • 刪：刪除索引庫數據
• 靜態頁微服務：
  • 增：創建新的靜態頁
  • 刪：刪除原來的靜態頁
  • 改：創建新的靜態頁並刪除原來的

3.2.商品服務發送消息

我們先在商品微服務leyou-item-service中實現發送消息。

3.2.1.引入依賴

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

3.2.2.配置文件

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

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

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

3.2.3.改造GoodsService

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

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

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

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

然后在新增的時候調用：

修改的時候調用：

3.3.搜索服務接收消息

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

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

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

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: leyou password: leyou virtual-host: /leyou

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

3.3.3.編寫監聽器

代碼：

@Component
public class GoodsListener { @Autowired private SearchService searchService; /**  * 處理insert和update的消息  *  * @param id  * @throws Exception  */ @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "leyou.create.index.queue", durable = "true"), exchange = @Exchange( value = "leyou.item.exchange", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC), key = {"item.insert", "item.update"})) public void listenCreate(Long id) throws Exception { if (id == null) { return; } // 創建或更新索引 this.searchService.createIndex(id); } /**  * 處理delete的消息  *  * @param id  */ @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "leyou.delete.index.queue", durable = "true"), exchange = @Exchange( value = "leyou.item.exchange", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC), key = "item.delete")) public void listenDelete(Long id) { if (id == null) { return; } // 刪除索引 this.searchService.deleteIndex(id); } }

3.3.4.編寫創建和刪除索引方法

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

public void createIndex(Long id) throws IOException { Spu spu = this.goodsClient.querySpuById(id); // 構建商品 Goods goods = this.buildGoods(spu); // 保存數據到索引庫 this.goodsRepository.save(goods); } public void deleteIndex(Long id) { this.goodsRepository.deleteById(id); }

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

3.4.靜態頁服務接收消息

商品靜態頁服務接收到消息后的處理：

• 增：創建新的靜態頁
• 刪：刪除原來的靜態頁
• 改：創建新的靜態頁並刪除原來的

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

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: leyou password: leyou virtual-host: /leyou

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

3.4.3.編寫監聽器

代碼：

@Component
public class GoodsListener { @Autowired private GoodsHtmlService goodsHtmlService; @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "leyou.create.web.queue", durable = "true"), exchange = @Exchange( value = "leyou.item.exchange", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC), key = {"item.insert", "item.update"})) public void listenCreate(Long id) throws Exception { if (id == null) { return; } // 創建頁面 goodsHtmlService.createHtml(id); } @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "leyou.delete.web.queue", durable = "true"), exchange = @Exchange( value = "leyou.item.exchange", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC), key = "item.delete")) public void listenDelete(Long id) { if (id == null) { return; } // 創建頁面 goodsHtmlService.deleteHtml(id); } }

3.4.4.添加刪除頁面方法

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

3.5.測試

3.5.1.查看RabbitMQ控制台

重新啟動項目，並且登錄RabbitMQ管理界面：http://192.168.56.101:15672

可以看到，交換機已經創建出來了：

隊列也已經創建完畢：

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

3.5.2.修改數據試一試

在后台修改商品數據的價格，分別在搜索及商品詳情頁查看是否統一。