Spring Boot之RabbitMQ
RabbitMQ 即一個訊息佇列，主要是用來實現應用程式的 非同步和解耦 ，同時也能起到訊息緩衝，訊息分發的作用。
訊息中介軟體在網際網路公司的使用中越來越多，剛才還看到新聞阿里將 RocketMQ 捐獻給了 Apache ，當然了今天的主角還是講 RabbitMQ 。 訊息中介軟體最主要的作用是解耦 ，中介軟體最標準的用法是生產者生產訊息傳送到佇列，消費者從佇列中拿取訊息並處理，生產者不用關心是誰來消費，消費者不用關心誰在生產訊息，從而達到解耦的目的。在分散式的系統中，訊息佇列也會被用在很多其它的方面，比如： 分散式事務的支援，RPC 的呼叫等等。
以前一直使用的是 ActiveMQ ，在實際的生產使用中也出現了一些小問題，在網路查閱了很多的資料後，決定嘗試使用 RabbitMQ 來替換 ActiveMQ，RabbitMQ 的高可用性、高效能、靈活性等 一些特點吸引了我們，查閱了一些資料整理出此文。
RabbitMQ 介紹
RabbitMQ是實現 AMQP（高階訊息佇列協議） 的訊息中介軟體的一種，最初起源於金融系統，用於在分散式系統中儲存轉發訊息，在 易用性、擴充套件性、高可用性 等方面表現不俗。 RabbitMQ 主要是為了 實現系統之間的雙向解耦而實現的 。當生產者大量產生資料時，消費者無法快速消費，那麼需要一箇中間層。儲存這個資料。
AMQP，即 Advanced Message Queuing Protocol， 高階訊息佇列協議 ，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。訊息中介軟體 主要用於元件之間的解耦 ，訊息的傳送者無需知道訊息使用者的存在，反之亦然。AMQP 的主要特徵是 面向訊息、佇列、路由（包括點對點和釋出/訂閱）、可靠性、安全。
RabbitMQ是一個開源的 AMQP 實現，伺服器端用Erlang語言編寫，支援多種客戶端，如： Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支援 AJAX。用於在 分散式系統中儲存轉發訊息 ，在 易用性、擴充套件性、高可用性 等方面表現不俗。
相關概念
通常我們談到佇列服務, 會有三個概念： 發訊息者、佇列、收訊息者 ，RabbitMQ 在這個基本概念之上, 多做了一層抽象, 在發訊息者和佇列之間, 加入了交換器 (Exchange). 這樣發訊息者和佇列就沒有直接聯絡, 轉而變成發訊息者把訊息給交換器, 交換器根據排程策略再把訊息再給佇列。
- 左側 P 代表 生產者，也就是往 RabbitMQ 發訊息的程式。
- 中間即是 RabbitMQ， 其中包括了交換機和佇列。
- 右側 C 代表 消費者，也就是往 RabbitMQ 拿訊息的程式。
那麼，其中比較重要的概念有 4 個，分別為：虛擬主機，交換機，佇列，和繫結。
- 虛擬主機 ：一個虛擬主機持有一組交換機、佇列和繫結。為什麼需要多個虛擬主機呢？很簡單， RabbitMQ 當中， 使用者只能在虛擬主機的粒度進行許可權控制。 因此，如果需要禁止A組訪問B組的交換機/佇列/繫結，必須為A和B分別建立一個虛擬主機。每一個 RabbitMQ 伺服器都有一個預設的虛擬主機“/”。
- 交換機 ： Exchange 用於轉發訊息，但是它不會做儲存 ，如果沒有 Queue bind 到 Exchange 的話，它會直接丟棄掉 Producer 傳送過來的訊息。 這裡有一個比較重要的概念： 路由鍵 。訊息到交換機的時候，互動機會轉發到對應的佇列中，那麼究竟轉發到哪個佇列，就要根據該路由鍵。
- 繫結 ：也就是交換機需要和佇列相繫結，這其中如上圖所示，是多對多的關係。
交換機(Exchange)
交換機的功能主要是接收訊息並且轉發到繫結的佇列，交換機不儲存訊息，在啟用ack模式後，交換機找不到佇列會返回錯誤。交換機有四種類型： Direct, topic, Headers and Fanout
- Direct ：direct 型別的行為是”先匹配, 再投送”. 即在繫結時設定一個 routing_key , 訊息的 routing_key 匹配時, 才會被交換器投送到繫結的佇列中去.
- Topic ：按規則轉發訊息（最靈活）
- Headers ：設定 header attribute 引數型別的交換機
- Fanout ：轉發訊息到所有繫結佇列
Direct Exchange
Direct Exchange 是 RabbitMQ 預設的交換機模式，也是最簡單的模式，根據key全文匹配去尋找佇列。
第一個 X - Q1 就有一個 binding key，名字為 orange； X - Q2 就有 2 個 binding key，名字為 black 和 green。 當訊息中的 路由鍵 和 這個 binding key 對應上的時候，那麼就知道了該訊息去到哪一個佇列中。
Ps：為什麼 X 到 Q2 要有 black，green，2個 binding key呢，一個不就行了嗎？ - 這個主要是因為可能又有 Q3，而Q3只接受 black 的資訊，而Q2不僅接受black 的資訊，還接受 green 的資訊。
Topic Exchange
Topic Exchange 轉發訊息主要是根據萬用字元。在這種交換機下，佇列和交換機的繫結會定義一種路由模式，那麼，萬用字元就要在這種路由模式和路由鍵之間匹配後交換機才能轉發訊息。
在這種交換機模式下：
. *
具體程式碼傳送的時候還是一樣，第一個引數表示交換機，第二個引數表示 routing key，第三個引數即訊息。如下：
rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this isRabbitMQ!");
topic 和 direct 類似, 只是匹配上支援了”模式”, 在”點分”的 routing_key 形式中, 可以使用兩個萬用字元:
* #
Headers Exchange
headers 也是根據規則匹配, 相較於 direct 和 topic 固定地使用 routing_key , headers 則是一個自定義匹配規則的型別. 在佇列與交換器繫結時, 會設定一組鍵值對規則, 訊息中也包括一組鍵值對( headers 屬性), 當這些鍵值對有一對, 或全部匹配時, 訊息被投送到對應佇列.
Fanout Exchange
Fanout Exchange 訊息廣播的模式，不管路由鍵或者是路由模式， 會把訊息發給繫結給它的全部佇列 ，如果配置了 routing_key 會被忽略。
Spring Boot 整合 RabbitMQ
Spring Boot 整合 RabbitMQ 非常簡單，如果只是簡單的使用配置非常少，Spring Boot 提供了 spring-boot-starter-amqp 專案對訊息各種支援。
簡單使用
1、配置 Pom 包，主要是新增 spring-boot-starter-amqp 的支援
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
2、配置檔案
配置 RabbitMQ 的安裝地址、埠以及賬戶資訊
spring.application.name=Spring-boot-rabbitmq spring.rabbitmq.host=192.168.0.86 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=123456
3、佇列配置
@Configuration
public class RabbitConfig {
@Bean
public Queue Queue() {
return new Queue("hello");
}
}
3、傳送者
rabbitTemplate 是 Spring Boot 提供的預設實現
@component
public class HelloSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String context = "hello " + new Date();
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("hello", context);
}
}
4、接收者
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver: " + hello);
}
}
5、測試
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {
@Autowired
private HelloSender helloSender;
@Test
public void hello() throws Exception {
helloSender.send();
}
}
注意，傳送者和接收者的 queue name 必須一致，不然不能接收
多對多使用
一個傳送者，N 個接收者或者 N 個傳送者和 N 個接收者會出現什麼情況呢？
一對多傳送
對上面的程式碼進行了小改造，接收端註冊了兩個 Receiver,Receiver1 和 Receiver2，傳送端加入引數計數，接收端列印接收到的引數，下面是測試程式碼，傳送一百條訊息，來觀察兩個接收端的執行效果
@Test
public void oneToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
}
}
結果如下：
Receiver 1: Spring boot neo queue ****** 11 Receiver 2: Spring boot neo queue ****** 12 Receiver 2: Spring boot neo queue ****** 14 Receiver 1: Spring boot neo queue ****** 13 Receiver 2: Spring boot neo queue ****** 15 Receiver 1: Spring boot neo queue ****** 16 Receiver 1: Spring boot neo queue ****** 18 Receiver 2: Spring boot neo queue ****** 17 Receiver 2: Spring boot neo queue ****** 19 Receiver 1: Spring boot neo queue ****** 20
根據返回結果得到以下結論
一個傳送者，N個接受者,經過測試會均勻的將訊息傳送到N個接收者中
多對多傳送
複製了一份傳送者，加入標記，在一百個迴圈中相互交替傳送
@Test
public void manyToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
neoSender2.send(i);
}
}
結果如下：
Receiver 1: Spring boot neo queue ****** 20 Receiver 2: Spring boot neo queue ****** 20 Receiver 1: Spring boot neo queue ****** 21 Receiver 2: Spring boot neo queue ****** 21 Receiver 1: Spring boot neo queue ****** 22 Receiver 2: Spring boot neo queue ****** 22 Receiver 1: Spring boot neo queue ****** 23 Receiver 2: Spring boot neo queue ****** 23 Receiver 1: Spring boot neo queue ****** 24 Receiver 2: Spring boot neo queue ****** 24 Receiver 1: Spring boot neo queue ****** 25 Receiver 2: Spring boot neo queue ****** 25
結論：和一對多一樣，接收端仍然會均勻接收到訊息
高階使用
物件的支援
Spring Boot 以及完美的支援物件的傳送和接收，不需要格外的配置。
//傳送者
public void send(User user) {
System.out.println("Sender object: " + user.toString());
this.rabbitTemplate.convertAndSend("object", user);
}
...
//接收者
@RabbitHandler
public void process(User user) {
System.out.println("Receiver object : " + user);
}
結果如下：
Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}
Topic Exchange
topic 是 RabbitMQ 中最靈活的一種方式，可以根據 routing_key 自由的繫結不同的佇列
首先對 topic 規則配置，這裡使用兩個佇列來測試
@Configuration
public class TopicRabbitConfig {
final static String message = "topic.message";
final static String messages = "topic.messages";
@Bean
public Queue queueMessage() {
return new Queue(TopicRabbitConfig.message);
}
@Bean
public Queue queueMessages() {
return new Queue(TopicRabbitConfig.messages);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("exchange");
}
@Bean
Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}
@Bean
Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}
}
使用 queueMessages 同時匹配兩個佇列，queueMessage 只匹配 “topic.message” 佇列
public void send1() {
String context = "hi, i am message 1";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}
public void send2() {
String context = "hi, i am messages 2";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
}
傳送send1會匹配到topic.#和topic.message 兩個Receiver都可以收到訊息，傳送send2只有topic.#可以匹配所有隻有Receiver2監聽到訊息
Fanout Exchange
Fanout 就是我們熟悉的廣播模式或者訂閱模式，給 Fanout 交換機發送訊息，綁定了這個交換機的所有佇列都收到這個訊息。
Fanout 相關配置
@Configuration
public class FanoutRabbitConfig {
@Bean
public Queue AMessage() {
return new Queue("fanout.A");
}
@Bean
public Queue BMessage() {
return new Queue("fanout.B");
}
@Bean
public Queue CMessage() {
return new Queue("fanout.C");
}
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
@Bean
Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
return BindingBuilder.bind(AMessage).to(fanoutExchange);
}
@Bean
Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(BMessage).to(fanoutExchange);
}
@Bean
Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(CMessage).to(fanoutExchange);
}
}
這裡使用了 A、B、C 三個佇列繫結到 Fanout 交換機上面，傳送端的 routing_key 寫任何字元都會被忽略：
public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}
結果如下：
Sender : hi, fanout msg ... fanout Receiver B: hi, fanout msg fanout Receiver A: hi, fanout msg fanout Receiver C: hi, fanout msg
結果說明，繫結到 fanout 交換機上面的佇列都收到了訊息
喜歡這篇文章的朋友可以點個喜歡，也可以關注一下我的個人專題：Java成長之路
針對於上面所涉及到的知識點我總結出了有1到5年開發經驗的程式設計師在面試中涉及到的絕大部分架構面試題及答案做成了文件和架構視訊資料免費分享給大家（包括Dubbo、Redis、Netty、zookeeper、Spring cloud、分散式、高併發等架構技術資料），希望能幫助到您面試前的複習且找到一個好的工作，也節省大家在網上搜索資料的時間來學習，也可以關注我一下以後會有更多幹貨分享。
資料獲取方式： QQ群搜尋“708-701-457” 即可免費領取
