RabbitMQ快速入门
MeassageQueue | Kyle’s Blog (cyborg2077.github.io)
初识MQ
同步和异步通讯
- 微服务间通讯有同步和异步两种方式
- 同步通讯：就像打电话，需要实时响应，而且通话期间，不能响应其他的电话，当有其他妹子给你打电话的时候，就容易错失
良机。时效性强 - 异步通讯：就像发邮件，QQ/微信聊天，不需要马上回复。可以同时给多个妹子发消息，支持多线操作，
时间管理大师的必备技能。时效性弱
- 同步通讯：就像打电话，需要实时响应，而且通话期间，不能响应其他的电话，当有其他妹子给你打电话的时候，就容易错失
- 两种方式各有优劣，打电话可以立即得到响应，但是却不能与多个人同时通话，发送邮件可以同时与多个人收发邮件，但是往往响应会有延迟
同步通讯
我们之前学习的（RestTemplate、OpenFeign）Feign调用就属于同步方式，虽然调用可以实时得到结果，但是存在一些问题。订单服务，仓储服务，短信服务是和我们的支付服务耦合在一起的。
public void PaymentService() {
orderService.doSth();
storageService.doSth();
messageService.doSth();
...
// 一升级 这个就需要加业务，改动代码。。。耦合度高
}- 综上所述，同步调用存在以下问题
- 耦合度高：每次加入新的请求，都需要修改原来的代码
- 性能下降：调用者需要等待服务提供者响应，如果调用链过长，则响应时间等于每次调用服务的时间之和
- 资源浪费：调用链中的每个服务在等待响应过程中，不能释放请求占用资源，高并发场景下会极度浪费系统资源
- 级联失败：如果服务提供者出现问题，那么调用方都会跟着出现问题，就像多米诺骨牌一样，迅速导致整个微服务故障
异步通讯（事件驱动模式）
为了解除事件发布者与订阅者之间的耦合，两者并不是直接通信，而是由一个中间人(Broker)来代理。发布者发布时间到Broker，不关心谁来订阅的事件（只要时间发布了，就不管我的事了，忙其他东西去了）。而订阅者从Broker订阅事件，不关心是谁发布的事件
异步通讯的优点
- 耦合度低（发布者只管发布业务； 新增业务无需改代码，新增微服务订阅Broker发布的事件即可）
- 吞吐量提升（发布者发布完就不用管了，认为任务执行完成了，可以立即接收其他请求了）
- 故障隔离（服务没有强依赖，下游微服务挂了不会强影响到上游，下游微服务自己解决后再连接）
- 流量削峰（不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件）
异步通讯的缺点
- 依赖于Broker的可靠性、安全性、吞吐能力（如大量任务发布，Broker需要能存储上）
- 架构复杂时，业务没有明确的流程线，不好追踪管理（出了bug都不好找）
- 好在现在开源平台上的 Broker 的软件是非常成熟的，比较常见的一种就是我们这里要学习的MQ技术。
技术对比
- MQ(MessageQueue)中文是消息队列，字面意思就是存放消息的队列，也就是事件驱动中的Broker
- 比较常见的MQ实现
- ActiveMQ
- RabbitMQ
- RockerMQ
- Kafka
- 几种常见的MQ对比
| RabbitMQ | ActiveMQ | RocketMQ | Kafka | |
|---|---|---|---|---|
| 公司/社区 | Rabbit | Apache | 阿里 | Apache |
| 开发语言 | Erlang | Java | Java | Scala&Java |
| 协议支持 | AMQP，XMPP，SMTP，STOMP | OpenWire,STOMP，REST,XMPP,AMQP | 自定义协议 | 自定义协议 |
| 可用性 | 高 | 一般 | 高 | 高 |
| 单机吞吐量 | 一般 | 差 | 高 | 非常高 |
| 消息延迟 | 微秒级 | 毫秒级 | 毫秒级 | 毫秒以内 |
| 消息可靠性 | 高 | 一般 | 高 | 一般 |
- 追求可用性：Kafka、RockerMQ、RabbitMQ
- 追求可靠性：RabbitMQ、RocketMQ
- 追求吞吐能力：RocketMQ、Kafka
- 追求消息低延迟：RabbitMQ、Kafka
快速入门
这里是在CentOS 7虚拟机中使用Docker安装的
拉取镜像
BASH docker pull rabbitmq:3-management使用docker images查看是否已经成功拉取，之后启动一个RabbitMQ容器
BASH docker run \ # 启动镜像-> 容器 -e RABBITMQ_DEFAULT_USER=root \ # -e配置环境变量 -e RABBITMQ_DEFAULT_PASS=root \ --name mq \ # 容器的名称 --hostname mq1 \ # 配置主机名 集群用 -p 15672:15672 \ # 端口映射（控制台窗口） 冒号左侧是宿主机端口（外部访问用的），右侧是容器端口（具体看写这个镜像的人怎么设置的） -p 5672:5672 \ -d \ rabbitmq:3-management # 镜像名称其中：两个环境变量分别配置登录用户和密码，15672是rabbitMQ的管理平台的端口，5672是将来做消息通信的端口
容器启动成功之后，我们输入
虚拟机ip:15672访问RabbitMQ的管理平台其中：两个环境变量分别配置登录用户和密码，15672是rabbitMQ的管理平台的端口，5672是将来做消息通信的端口
容器启动成功之后，我们输入
虚拟机ip:15672访问RabbitMQ的管理平台
RabbitMQ中的一些角色：
- publisher：生产者
- consumer：消费者
- exchange：交换机，负责消息路由
- queue：队列，存储消息
- virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离（类似于 linux的不同用户）
RabbitMQ消息类型
RabbitMQ官方提供了5个不同的Demo实例，对应了不同的消息模型
1.基本消息类型(BasicQueue)
2.工作消息队列(WorkQueue)
其中发布订阅(Publish、Subscribe)，又根据交换机类型不同，分为三种
3.广播(Fanout Exchange)
4.路由(Direct Exchange)
5.主题(Topic Exchange)
前置操作
导入准备好的工程
包括三部分：
mq-demo：整体父工程，管理项目依赖
publisher：子工程 消息的发送者 微服务
consumer：子工程 消息的消费者 微服务
基本消息类型(BasicQueue)
只有三个部分，
- publisher 消息发送者，我们用上面工程的publisher微服务来实现，消息发布者，将消息发送到队列queue
- consumer 消息消费者，我们用 上面工程的consumer微服务类实现，订阅队列，处理队列中的消息
- queue消息队列 ，这里部分“不归”代码管理，全权交予 rabbitMQ管理维护，负责接受并缓存消息
publisher实现
- 建立连接
- 创建Channel
- 声明队列
- 发送消息
- 关闭连接和Channel
JAVA
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class PublisherTest {
@Test
public void testSendMessage() throws IOException, TimeoutException {
// 1.建立连接 将本为服务与 RabbitMQ建立连接
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.150.101"); // 虚拟机地址
factory.setPort(5672);
factory.setVirtualHost("/"); // MQ用户
factory.setUsername("root");
factory.setPassword("root");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 执行完这里 观察 RabbitMQ 会发现已经出现了连接，但是通道队列都没有
// 2.创建通道Channel （发送的通道）
Channel channel = connection.createChannel();
// 3.声明队列
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 创建队列MQ中的队列
// 4.发送消息
String message = "hello, rabbitmq!";
channel.basicPublish("", queueName, null, message.getBytes());
System.out.println("发送消息成功：【" + message + "】");
// 5.关闭通道和连接
channel.close();
connection.close();
}
}consumer实现
- 建立连接
- 创建Channel
- 声明队列
- 订阅消息
JAVA
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ConsumerTest {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.建立连接
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.150.101");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("root");
factory.setPassword("root");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 2.创建通道Channel （接收的通道）
Channel channel = connection.createChannel();
// 3.声明队列，声明一个队列是幂等的——只有当它不存在时才会被创建，我们在这里再次声明队列可以防止队列为null
// 生产者消费者 启动的时机不一定
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.订阅消息
channel.basicConsume(queueName, true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
// 5.处理消息
String message = new String(body);
System.out.println("接收到消息：【" + message + "】");
}
});
System.out.println("等待接收消息。。。。");
// 你会发现（"等待接收消息。。。。" 先被打印出来，等接收到了消息处理才会回调 handleDelivery
// 消费完消息，会发现MQ中queue中的消息也没了 （阅后即焚）
}
}- 但是这种写法，看着就很麻烦，其中设置连接参数时，和我们之前学JDBC一样，手动在代码中设置连接四要素，但是后来我们数据库的连接参数都是写在yml文件中来简化代码的，这里同理，所以我们继续往下学SpringAMQP
总结
- 基本消息队列的消息发送流程
- 建立connection
- 创建channel
- 使用channel声明队列
- 使用channel向队列发送消息
- 基本消息队列的消息接收流程
- 建立connection
- 创建channel
- 使用channel声明队列
- 定义consumer的消费行为handleDelivery()，也就是接收到消息后要执行的业务逻辑
- 利用channel将消费者与队列绑定
SpringAMQP
- SpringAMQP官方地址：https://spring.io/projects/spring-amqp， 它是基于RabbitMQ封装的一套实现模板，提供了模板来发送和接收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现，并且利用SpringBoot对其实现了自动装配，使用起来十分方便 （也就是 SpringAMQP 其中的 spring-amqp 对 spring-rabbit 进行了代码的接口抽象，底层还是由 spring-rabbit 实现，接口的标准是按照AMQP协议来写的 ）
AMQP(Advanced Message Queuing Portocol)，是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求
SpringAMQP提供了三个功能
- Rabbit Admin 自动声明队列、交换机及其绑定关系
- 基于注解的监听模式，异步接收消息
- 封装了RabbitTemplate工具，用于发送消息(之前在Redis中我们也接触过RedisTemplate）
再次回顾
RabbitMQ消息类型
RabbitMQ官方提供了5个不同的Demo实例，对应了不同的消息模型
1.基本消息类型(BasicQueue)
2.工作消息队列(WorkQueue)
其中发布订阅(Publish、Subscribe)，又根据交换机类型不同，分为三种
3.广播(Fanout Exchange)
4.路由(Direct Exchange)
5.主题(Topic Exchange)
下面使用RabbitMQ来优雅的实现上面的模型吧
基本消息类型(BasicQueue)
在父工程mq-demo中引入依赖（子工程就不用引入了）
<!--AMQP依赖，包含RabbitMQ--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
消息发送
首先配置MQ地址，在publisher服务的application.yml中添加配置
spring: rabbitmq: host: 192.168.128.130 # 主机名 port: 5672 #端口 username: root # 用户名 password: root # 密码 virtual-host: / # 虚拟主机然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送，如果没有创建simple.queue，可以在RabbitMQ管理平台
import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest // 关于 @RunWith //如果测试类中涉及到@Autowired注入，那就必须加上@RunWith注解，不然自带的junit测试环境也还是报错空指针 // 在IDEA里去掉@RunWith仍然能跑,因为 IDEA自动识别 配置类，不用其他ide不一定了 public class SpringAmqpTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void testSimpleQueue() { String queueName = "simple.queue"; String message = "Hello, SpringAMQP! "; rabbitTemplate.convertAndSend(queueName, message); } }
消息接收
刚刚我们已经完成了消息的发送，那么现在我们来完成消息的监听
首先配置MQ地址，在consumer服务的application.yml中添加配置
spring: rabbitmq: host: 192.168.128.130 # 主机名 port: 5672 #端口 username: root # 用户名 password: root # 密码 virtual-host: / # 虚拟主机消费者要做的事情是监听消息，Spring已经帮我们跟MQ建立了连接，所以消费者只需要关心要监听哪个队列，以及监听到队列了，要干什么事儿
在consumer服务新建一个类，编写消费逻辑，其中
@Component注解是将该类声明为一个Bean，这样Spring才能找到它，然后使用@RabbitListener注解声明要监听的队列，msg参数就是接收到的消息，我们可以在方法体内编写对消息的处理逻辑，但这里只是简单的输出了一下@Component public class SpringRabbitListener { @RabbitListener(queues = "simple.queue") public void listenSimpleQueueMessage(String msg) { System.out.println("Spring 消费者接收到消息：【" + msg + "】"); } }启动consumer服务，控制台可以看到接收到的消息，并且如果多次使用publisher服务发送消息，consumer服务也会接收多次消息，表现为控制台也输出多次语句
Spring 消费者接收到消息：【Hello, SpringAMQP! 】 Spring 消费者接收到消息：【Hello, SpringAMQP! 】
小结
SpringAMQP如何发送消息？
引入amqp的starter依赖
配置RabbitMQ地址
rabbitTemplate.convertAndSend(queueName, message);
SpringAMQP如何接收消息？
引入amqp的starter依赖
配置RabbitMQ地址
定义类，添加@Component注解
类中声明方法，添加
@RabbitListener(queues = "simple.queue")注解，方法参数 消息队列
注意：消息一旦消费就会从队列删除，RabbitMQ没有消息回溯功能
工作消息队列(WorkQueue)
Work Queue，也被称为(Task queues)，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的信息。- 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度，长此以往，消息就会堆积的越来越多，无法及时处理。
- 此时就可以使用
Work模型，多个消费者共同处理消息，那么处理速度就能大大提高了
消息发送
这次我们循环发送消息，模拟大量消息堆积的场景
在
publisher服务中的SpringAmqpTest类中新增测试方法@Test public void testWorkQueue() throws InterruptedException { String queueName = "simple.queue"; String message = "Hello, SpringAMQP! __ "; for (int i = 1; i <= 50; i++) { // 循环发送50条消息，带上消息编号 rabbitTemplate.convertAndSend(queueName, message + i); // 休眠20ms，模拟在1s内发送完 Thread.sleep(20); } }
消息接收
要模拟多个消费者绑定同一个队列，我们在
consumer服务的SpringRabbitListener中添加两个方法@RabbitListener(queues = "simple.queue") public void listenWorkQueueMessage1(String msg) throws InterruptedException { System.out.println("消费者1 接收到消息：【" + msg + "】" + LocalDateTime.now()); // 休眠20ms，1s大致能处理50条消息 Thread.sleep(20); } @RabbitListener(queues = "simple.queue") public void listenWorkQueueMessage2(String msg) throws InterruptedException { System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalDateTime.now()); // 休眠200ms，1s大概能处理5条消息 Thread.sleep(200); }
测试
启动
consumer服务后，执行publisher服务中刚编写的测试方法，发送50条消息，查看控制台输出消费者2........接收到消息：【Hello, SpringAMQP! __ 1】2022-12-23T13:16:41.407 消费者1 接收到消息：【Hello, SpringAMQP! __ 2】2022-12-23T13:16:41.414 消费者1 接收到消息：【Hello, SpringAMQP! __ 4】2022-12-23T13:16:41.461 消费者1 接收到消息：【Hello, SpringAMQP! __ 6】2022-12-23T13:16:41.502 消费者1 接收到消息：【Hello, SpringAMQP! __ 8】2022-12-23T13:16:41.549 消费者1 接收到消息：【Hello, SpringAMQP! __ 10】2022-12-23T13:16:41.592 消费者2........接收到消息：【Hello, SpringAMQP! __ 3】2022-12-23T13:16:41.609 消费者1 接收到消息：【Hello, SpringAMQP! __ 12】2022-12-23T13:16:41.635 消费者1 接收到消息：【Hello, SpringAMQP! __ 14】2022-12-23T13:16:41.680 消费者1 接收到消息：【Hello, SpringAMQP! __ 16】2022-12-23T13:16:41.722 消费者1 接收到消息：【Hello, SpringAMQP! __ 18】2022-12-23T13:16:41.767 消费者2........接收到消息：【Hello, SpringAMQP! __ 5】2022-12-23T13:16:41.811 消费者1 接收到消息：【Hello, SpringAMQP! __ 20】2022-12-23T13:16:41.813 消费者1 接收到消息：【Hello, SpringAMQP! __ 22】2022-12-23T13:16:41.858 消费者1 接收到消息：【Hello, SpringAMQP! __ 24】2022-12-23T13:16:41.902 消费者1 接收到消息：【Hello, SpringAMQP! __ 26】2022-12-23T13:16:41.944 消费者1 接收到消息：【Hello, SpringAMQP! __ 28】2022-12-23T13:16:41.990 消费者2........接收到消息：【Hello, SpringAMQP! __ 7】2022-12-23T13:16:42.013 消费者1 接收到消息：【Hello, SpringAMQP! __ 30】2022-12-23T13:16:42.035 消费者1 接收到消息：【Hello, SpringAMQP! __ 32】2022-12-23T13:16:42.078 消费者1 接收到消息：【Hello, SpringAMQP! __ 34】2022-12-23T13:16:42.122 消费者1 接收到消息：【Hello, SpringAMQP! __ 36】2022-12-23T13:16:42.165 消费者1 接收到消息：【Hello, SpringAMQP! __ 38】2022-12-23T13:16:42.210 消费者2........接收到消息：【Hello, SpringAMQP! __ 9】2022-12-23T13:16:42.215 消费者1 接收到消息：【Hello, SpringAMQP! __ 40】2022-12-23T13:16:42.256 消费者1 接收到消息：【Hello, SpringAMQP! __ 42】2022-12-23T13:16:42.300 消费者1 接收到消息：【Hello, SpringAMQP! __ 44】2022-12-23T13:16:42.343 消费者1 接收到消息：【Hello, SpringAMQP! __ 46】2022-12-23T13:16:42.387 消费者2........接收到消息：【Hello, SpringAMQP! __ 11】2022-12-23T13:16:42.416 消费者1 接收到消息：【Hello, SpringAMQP! __ 48】2022-12-23T13:16:42.431 消费者1 接收到消息：【Hello, SpringAMQP! __ 50】2022-12-23T13:16:42.473 消费者2........接收到消息：【Hello, SpringAMQP! __ 13】2022-12-23T13:16:42.617 消费者2........接收到消息：【Hello, SpringAMQP! __ 15】2022-12-23T13:16:42.819 消费者2........接收到消息：【Hello, SpringAMQP! __ 17】2022-12-23T13:16:43.021 消费者2........接收到消息：【Hello, SpringAMQP! __ 19】2022-12-23T13:16:43.224 消费者2........接收到消息：【Hello, SpringAMQP! __ 21】2022-12-23T13:16:43.425 消费者2........接收到消息：【Hello, SpringAMQP! __ 23】2022-12-23T13:16:43.626 消费者2........接收到消息：【Hello, SpringAMQP! __ 25】2022-12-23T13:16:43.827 消费者2........接收到消息：【Hello, SpringAMQP! __ 27】2022-12-23T13:16:44.027 消费者2........接收到消息：【Hello, SpringAMQP! __ 29】2022-12-23T13:16:44.229 消费者2........接收到消息：【Hello, SpringAMQP! __ 31】2022-12-23T13:16:44.431 消费者2........接收到消息：【Hello, SpringAMQP! __ 33】2022-12-23T13:16:44.634 消费者2........接收到消息：【Hello, SpringAMQP! __ 35】2022-12-23T13:16:44.837 消费者2........接收到消息：【Hello, SpringAMQP! __ 37】2022-12-23T13:16:45.040 消费者2........接收到消息：【Hello, SpringAMQP! __ 39】2022-12-23T13:16:45.240 消费者2........接收到消息：【Hello, SpringAMQP! __ 41】2022-12-23T13:16:45.444 消费者2........接收到消息：【Hello, SpringAMQP! __ 43】2022-12-23T13:16:45.646 消费者2........接收到消息：【Hello, SpringAMQP! __ 45】2022-12-23T13:16:45.846 消费者2........接收到消息：【Hello, SpringAMQP! __ 47】2022-12-23T13:16:46.048 消费者2........接收到消息：【Hello, SpringAMQP! __ 49】2022-12-23T13:16:46.250可以看到消费者1很快就完成了自己的25条消息，消费者2却在缓慢的处理自己的25条消息
也就是说当前的处理方式是平均分配给每个消费者，并没有按照我们设定的处理能力来处理消息，这样显然是有问题的，消费者2很大概率会出现消息积压的风险
能者多劳
在spring中有一个简单的配置，可以解决这个问题。我们修改
consumer服务中的application.yml文件，添加如下配置spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息重启服务，查看控制台输出
消费者2........接收到消息：【Hello, SpringAMQP! __ 1】2022-12-23T13:37:15.252 消费者1 接收到消息：【Hello, SpringAMQP! __ 2】2022-12-23T13:37:15.262 消费者1 接收到消息：【Hello, SpringAMQP! __ 3】2022-12-23T13:37:15.284 消费者1 接收到消息：【Hello, SpringAMQP! __ 4】2022-12-23T13:37:15.307 消费者1 接收到消息：【Hello, SpringAMQP! __ 5】2022-12-23T13:37:15.331 消费者1 接收到消息：【Hello, SpringAMQP! __ 6】2022-12-23T13:37:15.354 消费者1 接收到消息：【Hello, SpringAMQP! __ 7】2022-12-23T13:37:15.377 消费者1 接收到消息：【Hello, SpringAMQP! __ 8】2022-12-23T13:37:15.399 消费者1 接收到消息：【Hello, SpringAMQP! __ 9】2022-12-23T13:37:15.422 消费者1 接收到消息：【Hello, SpringAMQP! __ 10】2022-12-23T13:37:15.444 消费者2........接收到消息：【Hello, SpringAMQP! __ 11】2022-12-23T13:37:15.464 消费者1 接收到消息：【Hello, SpringAMQP! __ 12】2022-12-23T13:37:15.486 消费者1 接收到消息：【Hello, SpringAMQP! __ 13】2022-12-23T13:37:15.509 消费者1 接收到消息：【Hello, SpringAMQP! __ 14】2022-12-23T13:37:15.533 消费者1 接收到消息：【Hello, SpringAMQP! __ 15】2022-12-23T13:37:15.559 消费者1 接收到消息：【Hello, SpringAMQP! __ 16】2022-12-23T13:37:15.583 消费者1 接收到消息：【Hello, SpringAMQP! __ 17】2022-12-23T13:37:15.606 消费者1 接收到消息：【Hello, SpringAMQP! __ 18】2022-12-23T13:37:15.630 消费者1 接收到消息：【Hello, SpringAMQP! __ 19】2022-12-23T13:37:15.653 消费者2........接收到消息：【Hello, SpringAMQP! __ 20】2022-12-23T13:37:15.668 消费者1 接收到消息：【Hello, SpringAMQP! __ 21】2022-12-23T13:37:15.687 消费者1 接收到消息：【Hello, SpringAMQP! __ 22】2022-12-23T13:37:15.710 消费者1 接收到消息：【Hello, SpringAMQP! __ 23】2022-12-23T13:37:15.732 消费者1 接收到消息：【Hello, SpringAMQP! __ 24】2022-12-23T13:37:15.755 消费者1 接收到消息：【Hello, SpringAMQP! __ 25】2022-12-23T13:37:15.777 消费者1 接收到消息：【Hello, SpringAMQP! __ 26】2022-12-23T13:37:15.801 消费者1 接收到消息：【Hello, SpringAMQP! __ 27】2022-12-23T13:37:15.825 消费者1 接收到消息：【Hello, SpringAMQP! __ 28】2022-12-23T13:37:15.847 消费者1 接收到消息：【Hello, SpringAMQP! __ 29】2022-12-23T13:37:15.871 消费者2........接收到消息：【Hello, SpringAMQP! __ 30】2022-12-23T13:37:15.885 消费者1 接收到消息：【Hello, SpringAMQP! __ 31】2022-12-23T13:37:15.908 消费者1 接收到消息：【Hello, SpringAMQP! __ 32】2022-12-23T13:37:15.930 消费者1 接收到消息：【Hello, SpringAMQP! __ 33】2022-12-23T13:37:15.955 消费者1 接收到消息：【Hello, SpringAMQP! __ 34】2022-12-23T13:37:15.977 消费者1 接收到消息：【Hello, SpringAMQP! __ 35】2022-12-23T13:37:16 消费者1 接收到消息：【Hello, SpringAMQP! __ 36】2022-12-23T13:37:16.021 消费者1 接收到消息：【Hello, SpringAMQP! __ 37】2022-12-23T13:37:16.043 消费者1 接收到消息：【Hello, SpringAMQP! __ 38】2022-12-23T13:37:16.065 消费者1 接收到消息：【Hello, SpringAMQP! __ 39】2022-12-23T13:37:16.087 消费者2........接收到消息：【Hello, SpringAMQP! __ 40】2022-12-23T13:37:16.107 消费者1 接收到消息：【Hello, SpringAMQP! __ 41】2022-12-23T13:37:16.130 消费者1 接收到消息：【Hello, SpringAMQP! __ 42】2022-12-23T13:37:16.153 消费者1 接收到消息：【Hello, SpringAMQP! __ 43】2022-12-23T13:37:16.174 消费者1 接收到消息：【Hello, SpringAMQP! __ 44】2022-12-23T13:37:16.197 消费者1 接收到消息：【Hello, SpringAMQP! __ 45】2022-12-23T13:37:16.219 消费者1 接收到消息：【Hello, SpringAMQP! __ 46】2022-12-23T13:37:16.242 消费者1 接收到消息：【Hello, SpringAMQP! __ 47】2022-12-23T13:37:16.264 消费者1 接收到消息：【Hello, SpringAMQP! __ 48】2022-12-23T13:37:16.286 消费者1 接收到消息：【Hello, SpringAMQP! __ 49】2022-12-23T13:37:16.310 消费者2........接收到消息：【Hello, SpringAMQP! __ 50】2022-12-23T13:37:16.330这次处理消息的比例差不多就是10:1了，能者多劳
总结
- Wrok模型的使用：
- 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
- 通过设置
prefetch来控制消费者预取的消息数量
发布（Publish）订阅（Subscribe）(同一个消息发给多个队列)
发布订阅的模型如图
可以看到，在订阅模型中，多了一个
exchange角色，而且过程略有变化Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发送给exchange(交换机)
Exchange：交换机。一方面，接收生产者发送的消息；另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或者是将消息对其。到底如何操作，取决于Exchange的类型。Exchange有以下三种类型
Fanout：广播，将消息交给所有绑定到交换机的队列Direct：定向，把消息交给符合指定routing key的队列Topic：通配符，把消息交给符合routing pattern(路由模式)的队列
Consumer：消费者，与以前一样，订阅队列，没有变化
Queue：消息队列也与之前一样，接收消息，缓存消息
Exchange(交换机)只负责转发消息，不具备存储消息的能力。因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
广播(Fanout Exchange)
- Fanout直译是扇出，但是在MQ中叫
广播更合适
- 在广播模式下，消息发送流程是这样的
- 可以存在多个队列
- 每个队列都要绑定到Exchange(交换机)
- 生产者发送消息，只能发送到交换机，交换机来决定要发给哪个队列，而不是生产者来决定
- 交换机把消息发送给绑定过的所有队列
- 订阅队列的消费者都能拿到消息
声明队列和交换机
Spring提供了一个接口
Exchange，来表示所有不同类型的交换机
在
consumer中创建一个配置类，声明队列和交换机import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class FanoutConfig { /** * 声明交换机 * @return Fanout类型交换机 */ @Bean public FanoutExchange fanoutExchange() { return new FanoutExchange("exchange"); } /** * 第1个队列 绑定 */ @Bean public Queue queue1() { return new Queue("queue1"); } @Bean public Binding bindingQueue1(Queue queue1, FanoutExchange fanoutExchange) { return BindingBuilder.bind(queue1).to(fanoutExchange); } /** * 第2个队列 绑定第2个队列和交换机 */ @Bean Queue queue2() { return new Queue("queue2"); } @Bean public Binding bindingQueue2(Queue queue2, FanoutExchange fanoutExchange) { return BindingBuilder.bind(queue2).to(fanoutExchange); } }
消息发送
在
publisher服务的SpringAmqpTest类中添加测试方法@Test public void testFanoutExchange() { String exchangeName = "exchange"; String message = "Hello Everyone~"; rabbitTemplate.convertAndSend(exchangeName, "", message); }
消息接收
在
consumer服务的SpringRabbitListener中添加两个方法作为消费者@RabbitListener(queues = "queue1") public void listenFanoutQueue1(String msg) { System.out.println("消费者1收到广播消息：【" + msg + "】"); } @RabbitListener(queues = "queue2") public void listenFanoutQueue2(String msg) { System.out.println("消费者2收到广播消息：【" + msg + "】"); }重启
consumer服务，运行publisher中新编写的测试方法，控制台输出如下消费者1收到广播消息：【Hello Everyone~】 消费者2收到广播消息：【Hello Everyone~】
总结
- 交换机的作用是什么？
- 接收
publisher发送的消息 - 将消息按照规则路由到与之绑定的队列
- 不能缓存消息，路由失败，消息丢失
- FanoutExchange会将消息路由到每个绑定的队列
- 接收
- 声明队列、交换机、绑定关系的Bean是什么？
QueueFanoutExchangeBinding
路由(Direct Exchange) - 定向路由
- 在
Fanout模式中，一条消息，会被所有订阅的队列都消费，但是在某些场景下，我们希望不同的消息被不同的队列消费，这时就要用到Direct类型的Exchange - 在
Direct模型下- 队列与虚拟机绑定，不能是任意绑定了，而是需要指定一个
RoutingKey(路由key) - 消息的发送方在向Exchange发送消息时，也必须指定消息的
RoutingKey。 - Exchange不再把消息交给每一个绑定的队列，而是根据消息的
RoutingKey进行判断，只有队列的RoutingKey与消息的RoutingKey完全一致，才会收到消息
- 队列与虚拟机绑定，不能是任意绑定了，而是需要指定一个
案例需求如下：
- 路由
RabbitListener声明Exchange、Queue、RoutingKey - 在
consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2 - 在
publisher中编写测试方法，向directExchange发送消息
消息接收 ： 消费者基于注解声明队列和交换机
基于
Bean的方式声明队列与交换机比较麻烦，所以Spring还提供了基于注解方式来声明！！！！在
consumer的SpringRabbitListener中添加两个消费者，同时基于注解来声明队列和交换机
@RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue1"), exchange = @Exchange(name = "directExchange", type = ExchangeTypes.DIRECT), key = {"Apex", "DeepRock"} )) public void listenDirectQueue1(String msg) { System.out.println("消费者收到direct.queue1的消息：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue2"), exchange = @Exchange(name = "directExchange", type = ExchangeTypes.DIRECT), key = {"Apex", "Terraria"} )) public void listenDirectQueue2(String msg) { System.out.println("消费者收到direct.queue2的消息：【" + msg + "】"); }
消息发送
在
publisher服务的SpringAmqpTest类中添加测试方法
- Apex
- DeepRock
- Terraria
