SpringBoot整合RabbitMQ和概念介紹

2022-05-22 19:00:03

一、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 傳送過來的訊息。

這裡有一個比較重要的概念：路由鍵。訊息到交換機的時候，互動機會轉發到對應的佇列中，那麼究竟轉發到哪個佇列，就要根據該路由鍵。

繫結：也就是交換機需要和佇列相繫結，這其中如上圖所示，是多對多的關係。

SpringBoot整合RabbitMQ非常簡單，如果只是簡單的使用設定非常少，springboot提供了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=spirng-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");
}
}

4.傳送者

rabbitTemplate是springboot 提供的預設實現
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);
}
}

5.接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver : " + hello);
}
}

6.測試

@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: spirng boot neo queue ****** 11
Receiver 2: spirng boot neo queue ****** 12
Receiver 2: spirng boot neo queue ****** 14
Receiver 1: spirng boot neo queue ****** 13
Receiver 2: spirng boot neo queue ****** 15
Receiver 1: spirng boot neo queue ****** 16
Receiver 1: spirng boot neo queue ****** 18
Receiver 2: spirng boot neo queue ****** 17
Receiver 2: spirng boot neo queue ****** 19
Receiver 1: spirng 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: spirng boot neo queue ****** 20
Receiver 2: spirng boot neo queue ****** 20
Receiver 1: spirng boot neo queue ****** 21
Receiver 2: spirng boot neo queue ****** 21
Receiver 1: spirng boot neo queue ****** 22
Receiver 2: spirng boot neo queue ****** 22
Receiver 1: spirng boot neo queue ****** 23
Receiver 2: spirng boot neo queue ****** 23
Receiver 1: spirng boot neo queue ****** 24
Receiver 2: spirng boot neo queue ****** 24
Receiver 1: spirng boot neo queue ****** 25
Receiver 2: spirng boot neo queue ****** 25

結論：和一對多一樣，接收端仍然會均勻接收到訊息.

四、高階使用

//物件的支援
//springboot以及完美的支援物件的傳送和接收，不需要格外的設定。
//傳送者
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'}

1.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監聽到訊息

2.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);
}

結果如下：