【SpringCloud学习笔记】RabbitMQ（上）

1. RabbitMQ简介

官网地址：https://www.rabbitmq.com/

2. 安装方式

安装前置准备：
此处基于Linux平台 + Docker进行安装，前置准备如下：

• Linux云服务器 / 虚拟机
• Docker环境

安装命令：

docker run \-eRABBITMQ_DEFAULT_USER=ricejson \-eRABBITMQ_DEFAULT_PASS=123321\-v mq-plugins:/plugins \--name mq \--hostname mq \-p15672:15672 \-p5672: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：

1. 在RabbitMQ的控制台中选择Queues
2. 配置两个队列，名称例如hello.queue1、hello.queue2
3. 点击Add queue添加该队列

Step2：配置exchange：

1. 在RabbitMQ控制台中选择Exchanges
2. 配置交换机信息，例如名称：hello.exchange，类型为fanout
3. 点击Add exchange添加交换机

Step3：配置binding关系：
就像在架构图中所示，我们的exchange会根据一定规则将生产者生产的消息路由给特定的queue，因此此处我们需要配置对应的绑定关系：

1. 在exchange显示台中选择新添加的hello.exchange
2. 进入其中，选择binding项，配置两个binding规则，分别将hello.queue1与hello.queue2绑定到hello.exchange上

Step4：publish投递消息：

1. 在配置的exchange中选择publish message列表
2. 配置message的payload载荷字段
3. 点击Publish message发送消息

Step5：验证结果：

1. 选择Queues，选择其中一个queue例如hello.queue1进入其中
2. 查看Get Message选项，查看接收到的消息

此时我们就完成了基于RabbitMQ提供的控制台来完成收发消息的操作！

3.3 SpringAMQP

我们已经实现了基于控制台的方式完成消息的收发，但是我们实际开发中仍然选用编程的方式来实现消息的收发，因此接下来就要学习如何使用RabbitMQ系列的客户端完成Java代码的编写。事实上，RabbitMQ针对不同的语言提供了不同的客户端可供使用，而针对Java语言提供的客户端编码较为复杂，因此此处使用Spring整合后的 **SpringAMQP **客户端进行Java编程！

1. 在maven配置文件中引入SpringAMQP依赖

<!--AMQP依赖，包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

1. 在application.yml配置RabbitMQ连接地址（publisher以及consumer均需要配置）

# 配置RabbitMQ地址spring:rabbitmq:host: 127.0.0.1
    port:5672username: guest
    password: guest

1. 在publisher模块中编写单元测试，测试消息发送

@SpringBootTest@Slf4jclassPublisherApplicationTest{@ResourceprivateRabbitTemplate rabbitTemplate;@TestpublicvoidtestPublish(){// 1. 定义发送队列名String queueName ="hello.queue1";// 2. 定义消息内容String msg ="hello, SpringAMQP";// 3. 发送消息
        rabbitTemplate.convertAndSend(queueName, msg);}}

1. 在consumer模块中编写组件类，使用@RabbitListener注解接收消息

@Component@Slf4jpublicclassRabbitMQListener{@RabbitListener(queues ="hello.queue1")publicvoidlisten(String msg){
        log.info("从【hello.queue1】接收到消息："+ msg);}}

启动项目后，观察日志文件，发现消费者确实监听到了Queue

hello.queue1

中的消息：

4. WorkQueues模型

WorkQueues模型：即任务模型，简单来说就是让多个消费者监听同一个队列中的消息，共同消费队列中的消息，该模型可用于防止消息堆积，在消息处理速度慢以及消息量大的情况非常适用，下面我们就来模拟这种场景：
需求：

1. 创建一个新的队列work.queue
2. 创建两个消费者监听同一个队列work.queue
3. 创建一个生产者，生产50条消息，观察结果

步骤：

1. 在RabbitMQ控制台中创建一个新的Queue：work.queue

1. 在RabbitMQListener类中创建两个方法用于模拟两个消费者监听同一个队列

@RabbitListener(queues ="work.queue")publicvoidlistenWorkQueue1(String msg){
    log.info("listener1 从【work.queue】接收到消息："+ msg);}@RabbitListener(queues ="work.queue")publicvoidlistenWorkQueue2(String msg){
    log.info("listener2 从【work.queue】接收到消息："+ msg);}

1. 在PublisherApplicationTest类中创建一个测试方法，模拟一个生产者生产50条消息

@TestpublicvoidtestWorkQueue(){// 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")publicvoidlistenWorkQueue1(String msg)throwsInterruptedException{
    log.info("listener1 从【work.queue】接收到消息："+ msg);Thread.sleep(40);}@RabbitListener(queues ="work.queue")publicvoidlistenWorkQueue2(String msg)throwsInterruptedException{
    log.info("listener2 从【work.queue】接收到消息："+ msg);Thread.sleep(200);}

此时发现消费者1消费速度很快，1s内就完成了，但是消费者2消费速度很慢，需要5s完成，基于默认的工作模型，消息仍旧均摊给两个消费者，拖慢了整体效率。

解决方案：

• 在消费者配置文件中引入如下配置：

spring:rabbitmq:listener:simple:prefetch:1

这样一来就改成了"能者多劳"的工作模式了