1. RabbitMQ简介
官网地址:https://www.rabbitmq.com/
2. 安装方式
安装前置准备:
此处基于Linux平台 + Docker进行安装,前置准备如下:
- Linux云服务器 / 虚拟机
- Docker环境
安装命令:
docker run \-e RABBITMQ_DEFAULT_USER=ricejson \-e RABBITMQ_DEFAULT_PASS=123321 \-v mq-plugins:/plugins \--name mq \--hostname mq \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:3.12.10-management-alpine
我们观察一下安装命令可以发现,配置了两个端口映射:
- 15672:该端口是RabbitMQ的控制台端口,用户可以在浏览器地址栏进行登录
- 5672:该端口用户不可见,供RabbitMQ内部数据传输使用
如果在地址栏中输入http://{ip}:15672
能够观察到如下界面,证明安装成功!
2.1 配置参考文档
需要验证:https://blog.csdn.net/dzqxwzoe/article/details/138517406
3. 快速入门
3.1 RabbitMQ架构图
上述就是RabbitMQ的架构图,其中涉及到几个重要角色:
- publisher:生产者,即发送消息的一方
- consumer:消费者,即接收消息进行业务处理的一方
- queue:队列,用于暂存生产者投递的消息,等待消费者处理
- exchange:交换机,负责将生产者投递的消息路由给具有binding关系的queue
3.2 收发消息
Step1:配置queue:
- 在RabbitMQ的控制台中选择
Queues
- 配置两个队列,名称例如
hello.queue1
、hello.queue2
- 点击
Add queue
添加该队列
Step2:配置exchange:
- 在RabbitMQ控制台中选择
Exchanges
- 配置交换机信息,例如名称:
hello.exchange
,类型为fanout
- 点击
Add exchange
添加交换机
Step3:配置binding关系:
就像在架构图中所示,我们的exchange会根据一定规则将生产者生产的消息路由给特定的queue,因此此处我们需要配置对应的绑定关系:
- 在exchange显示台中选择新添加的
hello.exchange
- 进入其中,选择
binding
项,配置两个binding规则,分别将hello.queue1与hello.queue2绑定到hello.exchange上
Step4:publish投递消息:
- 在配置的exchange中选择
publish message
列表 - 配置message的payload载荷字段
- 点击
Publish message
发送消息
Step5:验证结果:
- 选择
Queues
,选择其中一个queue例如hello.queue1
进入其中 - 查看
Get Message
选项,查看接收到的消息
此时我们就完成了基于RabbitMQ提供的控制台来完成收发消息的操作!
3.3 SpringAMQP
我们已经实现了基于控制台的方式完成消息的收发,但是我们实际开发中仍然选用编程的方式来实现消息的收发,因此接下来就要学习如何使用RabbitMQ系列的客户端完成Java代码的编写。事实上,RabbitMQ针对不同的语言提供了不同的客户端可供使用,而针对Java语言提供的客户端编码较为复杂,因此此处使用Spring整合后的 **SpringAMQP **客户端进行Java编程!
- 在maven配置文件中引入
SpringAMQP
依赖
<!--AMQP依赖,包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- 在
application.yml
配置RabbitMQ连接地址(publisher以及consumer均需要配置)
# 配置RabbitMQ地址
spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guest
- 在
publisher
模块中编写单元测试,测试消息发送
@SpringBootTest
@Slf4j
class PublisherApplicationTest {@Resourceprivate RabbitTemplate rabbitTemplate;@Testpublic void testPublish() {// 1. 定义发送队列名String queueName = "hello.queue1";// 2. 定义消息内容String msg = "hello, SpringAMQP";// 3. 发送消息rabbitTemplate.convertAndSend(queueName, msg);}
}
- 在
consumer
模块中编写组件类,使用@RabbitListener
注解接收消息
@Component
@Slf4j
public class RabbitMQListener {@RabbitListener(queues = "hello.queue1")public void listen(String msg) {log.info("从【hello.queue1】接收到消息:" + msg);}
}
启动项目后,观察日志文件,发现消费者确实监听到了Queuehello.queue1
中的消息:
4. WorkQueues模型
**WorkQueues模型:**即任务模型,简单来说就是让多个消费者监听同一个队列中的消息,共同消费队列中的消息,该模型可用于防止消息堆积,在消息处理速度慢以及消息量大的情况非常适用,下面我们就来模拟这种场景:
需求:
- 创建一个新的队列
work.queue
- 创建两个消费者监听同一个队列
work.queue
- 创建一个生产者,生产50条消息,观察结果
步骤:
- 在RabbitMQ控制台中创建一个新的Queue:
work.queue
- 在RabbitMQListener类中创建两个方法用于模拟两个消费者监听同一个队列
@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String msg) {log.info("listener1 从【work.queue】接收到消息:" + msg);
}@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String msg) {log.info("listener2 从【work.queue】接收到消息:" + msg);
}
- 在PublisherApplicationTest类中创建一个测试方法,模拟一个生产者生产50条消息
@Test
public void testWorkQueue() {// 1. 定义发送队列名String queueName = "work.queue";for (int i = 1; i <= 50; i++) {// 2. 定义消息内容String msg = "消息" + i;// 3. 发送消息rabbitTemplate.convertAndSend(queueName, msg);}
}
观察消费者结果如下:
可以发现默认的工作模式是采用类似于轮询方式的,那么这种方式在两个消费者消费速率相当的场景下没有问题,但是以后如果多个消费者部署在不同的机器上,多个机器性能不同造成,效率最低的及其拖慢了整个消费进程,下面就来模拟该情况:
@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {log.info("listener1 从【work.queue】接收到消息:" + msg);Thread.sleep(40);
}@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {log.info("listener2 从【work.queue】接收到消息:" + msg);Thread.sleep(200);
}
此时发现消费者1消费速度很快,1s内就完成了,但是消费者2消费速度很慢,需要5s完成,基于默认的工作模型,消息仍旧均摊给两个消费者,拖慢了整体效率。
解决方案:
- 在消费者配置文件中引入如下配置:
spring:rabbitmq:listener:simple:prefetch: 1
这样一来就改成了"能者多劳"的工作模式了