SpringBoot如何集成RabbitMQ

這篇文章主要介紹“SpringBoot如何集成RabbitMQ”的相關(guān)知識(shí)，小編通過(guò)實(shí)際案例向大家展示操作過(guò)程，操作方法簡(jiǎn)單快捷，實(shí)用性強(qiáng)，希望這篇“SpringBoot如何集成RabbitMQ”文章能幫助大家解決問(wèn)題。

一、RabbitMQ介紹

RabbitMQ是實(shí)現(xiàn)AMQP（高級(jí)消息隊(duì)列協(xié)議）的消息中間件的一種，最初起源于金融系統(tǒng)，用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、 高可用性等方面表現(xiàn)不俗。RabbitMQ主要是為了實(shí)現(xiàn)系統(tǒng)之間的雙向解耦而實(shí)現(xiàn)的。當(dāng)生產(chǎn)者大量產(chǎn)生數(shù)據(jù)時(shí)，消費(fèi)者無(wú)法快速消費(fèi)，那么需要一個(gè)中間層。保存這個(gè)數(shù)據(jù)。

AMQP，即Advanced Message Queuing Protocol，高級(jí)消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開(kāi)放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無(wú)需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、隊(duì)列、路由（包括點(diǎn)對(duì)點(diǎn)和發(fā)布/訂閱）、可靠性、安全。

RabbitMQ是一個(gè)開(kāi)源的AMQP實(shí)現(xiàn)，服務(wù)器端用Erlang語(yǔ)言編寫(xiě)，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。

二、相關(guān)概念

通常我們談到隊(duì)列服務(wù), 會(huì)有三個(gè)概念： 發(fā)消息者、隊(duì)列、收消息者，RabbitMQ 在這個(gè)基本概念之上, 多做了一層抽象, 在發(fā)消息者和 隊(duì)列之間, 加入了交換器 (Exchange). 這樣發(fā)消息者和隊(duì)列就沒(méi)有直接聯(lián)系, 轉(zhuǎn)而變成發(fā)消息者把消息給交換器, 交換器根據(jù)調(diào)度策略再把消息再給隊(duì)列

• 左側(cè) P 代表 生產(chǎn)者，也就是往 RabbitMQ 發(fā)消息的程序。

• 中間即是 RabbitMQ，其中包括了 交換機(jī) 和 隊(duì)列。

• 右側(cè) C 代表 消費(fèi)者，也就是往 RabbitMQ 拿消息的程序。

其中比較重要概念有 4 個(gè)，分別為：虛擬主機(jī)，交換機(jī)，隊(duì)列，和綁定。

• 虛擬主機(jī)：一個(gè)虛擬主機(jī)持有一組交換機(jī)、隊(duì)列和綁定。為什么需要多個(gè)虛擬主機(jī)呢？很簡(jiǎn)單，RabbitMQ當(dāng)中，用戶只能在虛擬主機(jī)的粒度進(jìn)行權(quán)限控制。 因此，如果需要禁止A組訪問(wèn)B組的交換機(jī)/隊(duì)列/綁定，必須為A和B分別創(chuàng)建一個(gè)虛擬主機(jī)。每一個(gè)RabbitMQ服務(wù)器都有一個(gè)默認(rèn)的虛擬主機(jī)“/”。

• 交換機(jī)：Exchange 用于轉(zhuǎn)發(fā)消息，但是它不會(huì)做存儲(chǔ) ，如果沒(méi)有 Queue bind 到 Exchange 的話，它會(huì)直接丟棄掉 Producer 發(fā)送過(guò)來(lái)的消息。

這里有一個(gè)比較重要的概念：路由鍵 。消息到交換機(jī)的時(shí)候，交互機(jī)會(huì)轉(zhuǎn)發(fā)到對(duì)應(yīng)的隊(duì)列中，那么究竟轉(zhuǎn)發(fā)到哪個(gè)隊(duì)列，就要根據(jù)該路由鍵。

• 綁定：也就是交換機(jī)需要和隊(duì)列相綁定，這其中如上圖所示，是多對(duì)多的關(guān)系。

SpringBoot集成RabbitMQ非常簡(jiǎn)單，如果只是簡(jiǎn)單的使用配置非常少，springboot提供了spring-boot-starter-amqp項(xiàng)目對(duì)消息各種支持。

三、簡(jiǎn)單使用

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.隊(duì)列配置

@Configuration
public class RabbitConfig {
@Bean
public Queue Queue() {
return new Queue("hello");
}
}

4.發(fā)送者

rabbitTemplate是springboot 提供的默認(rèn)實(shí)現(xiàn)
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.測(cè)試

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {
@Autowired
private HelloSender helloSender;
@Test
public void hello() throws Exception {
helloSender.send();
}
}

注意：發(fā)送者和接收者的queue name必須一致，不然不能接收

多對(duì)多使用

一個(gè)發(fā)送者，N個(gè)接收者或者N個(gè)發(fā)送者和N個(gè)接收者會(huì)出現(xiàn)什么情況呢？

一對(duì)多發(fā)送

  對(duì)上面的代碼進(jìn)行了小改造,接收端注冊(cè)了兩個(gè)Receiver,Receiver1和Receiver2，發(fā)送端加入?yún)?shù)計(jì)數(shù)，接收端打印接收到的參數(shù)，下面是測(cè)試代碼，發(fā)送一百條消息，來(lái)觀察兩個(gè)接收端的執(zhí)行效果.

@Test
public void oneToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
}
}

結(jié)果如下：

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

根據(jù)返回結(jié)果得到以下結(jié)論

一個(gè)發(fā)送者，N個(gè)接受者,經(jīng)過(guò)測(cè)試會(huì)均勻的將消息發(fā)送到N個(gè)接收者中

多對(duì)多發(fā)送

復(fù)制了一份發(fā)送者，加入標(biāo)記，在一百個(gè)循環(huán)中相互交替發(fā)送

@Test
public void manyToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
neoSender2.send(i);
}
}

結(jié)果如下：

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

結(jié)論：和一對(duì)多一樣，接收端仍然會(huì)均勻接收到消息.

四、高級(jí)使用

//對(duì)象的支持
//springboot以及完美的支持對(duì)象的發(fā)送和接收，不需要格外的配置。
//發(fā)送者
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);
}

結(jié)果如下：

Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}

1.Topic Exchange

topic 是RabbitMQ中最靈活的一種方式，可以根據(jù)routing_key自由的綁定不同的隊(duì)列

首先對(duì)topic規(guī)則配置，這里使用兩個(gè)隊(duì)列來(lái)測(cè)試

@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同時(shí)匹配兩個(gè)隊(duì)列，queueMessage只匹配"topic.message"隊(duì)列

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

發(fā)送send1會(huì)匹配到topic.#和topic.message 兩個(gè)Receiver都可以收到消息，發(fā)送send2只有topic.#可以匹配所有只有Receiver2監(jiān)聽(tīng)到消息

2.Fanout Exchange

Fanout 就是我們熟悉的廣播模式或者訂閱模式，給Fanout交換機(jī)發(fā)送消息，綁定了這個(gè)交換機(jī)的所有隊(duì)列都收到這個(gè)消息。

Fanout 相關(guān)配置：

@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三個(gè)隊(duì)列綁定到Fanout交換機(jī)上面，發(fā)送端的routing_key寫(xiě)任何字符都會(huì)被忽略：

public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}

結(jié)果如下：

Sender : hi, fanout msg
...
fanout Receiver B: hi, fanout msg
fanout Receiver A : hi, fanout msg
fanout Receiver C: hi, fanout msg

結(jié)果說(shuō)明，綁定到fanout交換機(jī)上面的隊(duì)列都收到了消息.

關(guān)于“SpringBoot如何集成RabbitMQ”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)，可以關(guān)注億速云行業(yè)資訊頻道，小編每天都會(huì)為大家更新不同的知識(shí)點(diǎn)。