0.學習目标
了解常見的MQ産品
了解RabbitMQ的5種消息模型
會使用Spring AMQP
利用MQ實作搜尋和靜态頁的資料同步
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.訂閱模型分類
在之前的模式中，我們建立了一個工作隊列。 工作隊列背後的假設是：每個任務隻被傳遞給一個從業人員。 在這一部分，我們将做一些完全不同的事情 - 我們将會傳遞一個資訊給多個消費者。 這種模式被稱為“釋出/訂閱”。
訂閱模型示意圖：
1527086284940
解讀：
1、1個生産者，多個消費者
2、每一個消費者都有自己的一個隊列
3、生産者沒有将消息直接發送到隊列，而是發送到了交換機
4、每個隊列都要綁定到交換機
5、生産者發送的消息，經過交換機到達隊列，實作一個消息被多個消費者擷取的目的 X（Exchanges）：交換機一方面：接收生産者發送的消息。另一方面：知道如何處理消息，例如遞交給某個特别隊列、遞交給所有隊列、或是将消息丢棄。到底如何操作，取決于Exchange的類型。
Exchange類型有以下幾種：
? Fanout：廣播，将消息交給所有綁定到交換機的隊列
? Direct：定向，把消息交給符合指定routing key 的隊列
? Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列 我們這裡先學習
? Fanout：即廣播模式
Exchange（交換機）隻負責轉發消息，不具備存儲消息的能力，是以如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那麼消息會丢失！ 2.4.訂閱模型-Fanout
Fanout，也稱為廣播。
流程圖：
1527086564505
在廣播模式下，消息發送流程是這樣的：
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
可以看到，交換機已經建立出來了：
隊列也已經建立完畢：
并且隊列都已經綁定到交換機：