SpringBoot集成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 發送過來的消息。
這裡有一個比較重要的概念：路由鍵 。消息到交換機的時候，交互機會轉發到對應的隊列中，那麼究竟轉發到哪個隊列，就要根據該路由鍵。
- 綁定：也就是交換機需要和隊列相綁定，這其中如上圖所示，是多對多的關系。
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);
}
結果如下：
Sender : hi, fanout msg
…
fanout Receiver B: hi, fanout msg
fanout Receiver A : hi, fanout msg
fanout Receiver C: hi, fanout msg
結果說明，綁定到fanout交換機上面的隊列都收到瞭消息.
到此這篇關於SpringBoot集成RabbitMQ和概念介紹的文章就介紹到這瞭,更多相關SpringBoot集成RabbitMQ內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- SpringBoot整合RabbitMQ實現六種工作模式的示例
- Spring Boot+RabbitMQ 通過fanout模式實現消息接收功能(支持消費者多實例部署)
- springboot整合消息隊列RabbitMQ
- SpringBoot整合RabbitMQ的5種模式實戰
- springBoot整合rabbitmq測試常用模型小結