SpringBoot集成RabbitMQ-阿里云开发者社区

SpringBoot集成RabbitMQ

2022-06-13 309

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： SpringBoot集成RabbitMQ

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

1.RabbitMQ介绍

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

2.相关概念

通常我们谈到队列服务, 会有三个概念：发消息者、队列、收消息者，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项目对消息各种支持。

3.简单使用

1.配置pom包

主要是添加spring-boot-starter-amqp的支持

org.springframework.boot

spring-boot-starter-amqp

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.队列配置

@ConfigurationpublicclassRabbitConfig {
@BeanpublicQueueQueue() {
returnnewQueue("hello");
    }
}

4.发送者

rabbitTemplate是springboot提供的默认实现

publicclassHelloSender {
@AutowiredprivateAmqpTemplaterabbitTemplate;
publicvoidsend() {
Stringcontext="hello "+newDate();
System.out.println("Sender : "+context);
this.rabbitTemplate.convertAndSend("hello", context);
    }
}

5.接收者

@Component@RabbitListener(queues="hello")
publicclassHelloReceiver {
@RabbitHandlerpublicvoidprocess(Stringhello) {
System.out.println("Receiver  : "+hello);
    }
}

6.测试

@RunWith(SpringRunner.class)
@SpringBootTestpublicclassRabbitMqHelloTest {
@AutowiredprivateHelloSenderhelloSender;
@Testpublicvoidhello() throwsException {
helloSender.send();
    }
}

注意，发送者和接收者的queue name必须一致，不然不能接收

多对多使用

一个发送者，N个接收者或者N个发送者和N个接收者会出现什么情况呢？

一对多发送

对上面的代码进行了小改造,接收端注册了两个Receiver,Receiver1和Receiver2，发送端加入参数计数，接收端打印接收到的参数，下面是测试代码，发送一百条消息，来观察两个接收端的执行效果.

@TestpublicvoidoneToMany() throwsException {
for (inti=0;i<100;i++){
neoSender.send(i);
    }
}
结果如下：Receiver1: spirngbootneoqueue******11Receiver2: spirngbootneoqueue******12Receiver2: spirngbootneoqueue******14Receiver1: spirngbootneoqueue******13Receiver2: spirngbootneoqueue******15Receiver1: spirngbootneoqueue******16Receiver1: spirngbootneoqueue******18Receiver2: spirngbootneoqueue******17Receiver2: spirngbootneoqueue******19Receiver1: spirngbootneoqueue******20

根据返回结果得到以下结论

一个发送者，N个接受者,经过测试会均匀的将消息发送到N个接收者中

多对多发送

复制了一份发送者，加入标记，在一百个循环中相互交替发送

@TestpublicvoidmanyToMany() throwsException {
for (inti=0;i<100;i++){
neoSender.send(i);
neoSender2.send(i);
        }
}
结果如下：Receiver1: spirngbootneoqueue******20Receiver2: spirngbootneoqueue******20Receiver1: spirngbootneoqueue******21Receiver2: spirngbootneoqueue******21Receiver1: spirngbootneoqueue******22Receiver2: spirngbootneoqueue******22Receiver1: spirngbootneoqueue******23Receiver2: spirngbootneoqueue******23Receiver1: spirngbootneoqueue******24Receiver2: spirngbootneoqueue******24Receiver1: spirngbootneoqueue******25Receiver2: spirngbootneoqueue******25

结论：和一对多一样，接收端仍然会均匀接收到消息.

4.高级使用

对象的支持

springboot以及完美的支持对象的发送和接收，不需要格外的配置。

//发送者publicvoidsend(Useruser) {
System.out.println("Sender object: "+user.toString());
this.rabbitTemplate.convertAndSend("object", user);
}
...
//接受者@RabbitHandlerpublicvoidprocess(Useruser) {
System.out.println("Receiver object : "+user);
}
结果如下：Senderobject: User{name='neo', pass='123456'}
Receiverobject : User{name='neo', pass='123456'}

Topic Exchange

topic 是RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列

首先对topic规则配置，这里使用两个队列来测试

@ConfigurationpublicclassTopicRabbitConfig {
finalstaticStringmessage="topic.message";
finalstaticStringmessages="topic.messages";
@BeanpublicQueuequeueMessage() {
returnnewQueue(TopicRabbitConfig.message);
    }
@BeanpublicQueuequeueMessages() {
returnnewQueue(TopicRabbitConfig.messages);
    }
@BeanTopicExchangeexchange() {
returnnewTopicExchange("exchange");
    }
@BeanBindingbindingExchangeMessage(QueuequeueMessage, TopicExchangeexchange) {
returnBindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }
@BeanBindingbindingExchangeMessages(QueuequeueMessages, TopicExchangeexchange) {
returnBindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

使用queueMessages同时匹配两个队列，queueMessage只匹配"topic.message"队列

publicvoidsend1() {
Stringcontext="hi, i am message 1";
System.out.println("Sender : "+context);
this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}
publicvoidsend2() {
Stringcontext="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 相关配置

@ConfigurationpublicclassFanoutRabbitConfig {
@BeanpublicQueueAMessage() {
returnnewQueue("fanout.A");
    }
@BeanpublicQueueBMessage() {
returnnewQueue("fanout.B");
    }
@BeanpublicQueueCMessage() {
returnnewQueue("fanout.C");
    }
@BeanFanoutExchangefanoutExchange() {
returnnewFanoutExchange("fanoutExchange");
    }
@BeanBindingbindingExchangeA(QueueAMessage,FanoutExchangefanoutExchange) {
returnBindingBuilder.bind(AMessage).to(fanoutExchange);
    }
@BeanBindingbindingExchangeB(QueueBMessage, FanoutExchangefanoutExchange) {
returnBindingBuilder.bind(BMessage).to(fanoutExchange);
    }
@BeanBindingbindingExchangeC(QueueCMessage, FanoutExchangefanoutExchange) {
returnBindingBuilder.bind(CMessage).to(fanoutExchange);
    }
}

这里使用了A、B、C三个队列绑定到Fanout交换机上面，发送端的routing_key写任何字符都会被忽略：

publicvoidsend() {
Stringcontext="hi, fanout msg ";
System.out.println("Sender : "+context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}
结果如下：Sender : hi, fanoutmsg...
fanoutReceiverB: hi, fanoutmsgfanoutReceiverA  : hi, fanoutmsgfanoutReceiverC: hi, fanoutmsg

结果说明，绑定到fanout交换机上面的队列都收到了消息.

⚠️注意 ~

💯本期内容就结束了，如果内容有误，麻烦大家评论区指出！

如有疑问❓可以在评论区💬或私信💬，尽我最大能力🏃‍♀️帮大家解决👨‍🏫！

如果我的文章有帮助到您，欢迎点赞+关注✔️鼓励博主🏃，您的鼓励是我分享的动力🏃🏃🏃~

SpringBoot集成RabbitMQ

1.RabbitMQ介绍

2.相关概念

3.简单使用

1.配置pom包

2.配置文件

3.队列配置

4.发送者

5.接收者

6.测试

4.高级使用

Topic Exchange

Fanout Exchange

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

SpringBoot集成RabbitMQ

1.RabbitMQ介绍

2.相关概念

3.简单使用

1.配置pom包

2.配置文件

3.队列配置

4.发送者

5.接收者

6.测试

4.高级使用

Topic Exchange

Fanout Exchange

热门文章

最新文章

相关课程

相关电子书

相关实验场景