SpringCloud学习笔记（九）RabbitMQ

By LeviranJava, Spring Cloud

9.MQ

服务异步通讯

9.1初识MQ

9.1.1 同步通讯

同步通信指发送方和接收方在通信过程中需要同时执行，即要等待发送方完成某个操作之后接收方才能开始处理数据。

微服务间基于Feign的调用就属于同步方式 eg.用户下订单需要调用支付服务，支付服务需要调用并等待其他服务响应才能支付.

优点：

  • 时效性高，立即执行并返回结果

缺点：

  • 耦合度高：每次加入新的需求，都需要修改原来的代码
  • 性能下降：调用者需要等待服务提供者响应
  • 资源浪费：调用链中每个服务在等待响应过程中不能释放请求占用的资源，高并发场景下会极度浪费系统资源
  • 级联失败：若服务提供者出现问题，所有调用方都会跟着出问题
image-20230701145553098

9.1.2 异步通讯

异步通信则是指发送方和接收方在通信过程中不需要同时执行，发送方发送完消息后即可继续进行其他操作，无需等待接收方处理数据。

异步调用常见实现是事件驱动模式.

优点：

  • 服务解耦：服务调用者不直接调用服务，服务提供者只需要订阅或取消订阅事件就可以增加或删除模块
  • 吞吐量提升：服务间不用一直等待响应 eg.业务耗时只包含支付服务+发布支付事件
  • 故障隔离：服务间没有强依赖，故无级联失败
  • 流量削峰：Broker相当于缓冲区，订阅服务按自己的处理能力处理Broker的事件

缺点：

  • 依赖于Broker的可靠性、安全性、吞吐能力
  • 架构复杂，业务没有明显的流程线，不好追踪管理（eg.支付服务不知道谁订阅了自己）
image-20230701150602736

总结：

在时效性要求高的场景下优先使用同步通信例如查询某个信息（实际开发中大多数场景都是同步），对并发要求高时效性要求不高使用异步.

9.1.3 MQ常见框架

MQ（Message Queue），存放消息的队列，即事件驱动架构中的Broker，消息指的是事件.

image-20230701152615064

9.2RabbitMQ

9.2.1安装使用

1.Docker拉取镜像

只有带-management的版本才有管理插件，不带management的版本只包含消息队列等RabbitMQ服务器的核心功能.

 docker pull rabbitmq:3-management

2.运行MQ容器

 docker run \
 -e RABBITMQ_DEFAULT_USER=yuan \
 -e RABBITMQ_DEFAULT_PASS=123321 \
 --name mq \
 --hostname mq1 \ #可以建立集群时再设置
 -p 15672:15672 \ #rabbitMQ管理平台端口
 -p 5672:5672 \  #rabbitMQ消息队列通信端口
 -d \
 rabbitmq:3-management

3.访问管理页面

ip+管理port-16572

image-20230701204857017
image-20230701205553932
image-20230701205700899

MQ架构

  • channel：操作MQ的工具
  • exchange：路由消息到队列中
  • queue：缓存消息
  • virtual hosts：虚拟主机，是对queue、exchange等资源的逻辑分组
image-20230701205817041

9.2.2 常见消息模型

image-20230701213304726

mq-demo示例：以”Hello World”基本消息队列为例

Publisher：

1.建立connection

2.创建channel

3.利用channel声明队列

4.利用channel向队列发送消息

image-20230701214627957
image-20230701214642415
image-20230701214850058
image-20230701215021754

Consumer：

1.建立connection

2.创建channel

3.利用channel声明队列

4.定义consumer的消费行为handleDelivery()

5.利用channel将消费者与队列绑定

image-20230701215557227
image-20230701215708789

注意这里的异步机制.另外一旦消息被消费者接收，queue里的消息就不存在了.

9.3 Spring AMQP

9.3.1 概念

AMQP：Advanced Message Queue Protocol，是用于在应用程序或之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

Spring AMQP是基于AMQP协议定制的一套API规范，提供了模板来发送和接收消息，包括两部分：spring-amqp是抽象基础，spring-rabbit是底层的默认实现。

简单来说就是Spring遵守AMQP的规范封装简化了上面的建立连接、发送接收过程.

Features

  • Listener container for asynchronous processing of inbound messages（listener container处理异步入站消息）
  • RabbitTemplate for sending and receiving messages
  • RabbitAdmin for automatically declaring queues, exchanges and bindings（RabbitAdmin自动声明queue等）

9.3.2 基本消息队列

案例HelloWorld-单一消费者

image-20230702162324520

流程：

1.父工程引入spring-amqp依赖

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

2.publisher服务中利用RabbitTemplate发送消息到simple.queue队列

 spring:
   rabbitmq:
     host: xx.xx.xx.xx # rabbitMQ的ip地址
     port: 5672 # 端口
     username: yuan
     password: 123321
     virtual-host: /
 
 @RunWith(SpringRunner.class)
 @SpringBootTest
 public class SpringAmqpTest {
     @Autowired
     private RabbitTemplate rabbitTemplate;
 ​
     @Test
     public void testSimpleQueue(){
         String queueName = "simple.queue";
         String message = "hello,spring amqp";
         rabbitTemplate.convertAndSend(queueName,message);
     }
 }

3.consumer服务中编写消费逻辑，绑定到simple.queue队列

 
 spring:
   rabbitmq:
     host: xx.xx.xx.xx # rabbitMQ的ip地址
     port: 5672 # 端口
     username: yuan
     password: 123321
     virtual-host: /
 //定义简单消费逻辑
 @Component
 public class SpringRabbitListener {
 ​
     @RabbitListener(queues = "simple.queue")
     public void listenSimpleQueue(String msg) {//队列里是String类型所以这里也是String
         System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
     }
 }

运行ConsumerApplication，当监听队列里有消息时，代码自动执行

image-20230702114507079

9.3.3 工作消息队列

案例WorkQueue-多消费者

多个consumer共同处理一个queue.

image-20230702162711090

流程：

1.publisher定义test每秒发送50条消息

  @Test
     public void testSendMessage2WorkQueue() throws InterruptedException {
         String queueName = "simple.queue";
         String message = "hello, message__";
         for (int i = 1; i <= 50; i++) {
             rabbitTemplate.convertAndSend(queueName, message + i);
             Thread.sleep(20);
         }
     }

2.consumer listner定义两个监听器

    @RabbitListener(queues = "simple.queue")
     public void listenWorkQueue1(String msg) throws InterruptedException {
         System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
         Thread.sleep(20);
     }
 ​
     @RabbitListener(queues = "simple.queue")
     public void listenWorkQueue2(String msg) throws InterruptedException {
         System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
         Thread.sleep(200);
     }

结果如下：

image-20230702164327464
image-20230702164406282

结果分析：

结果：消息预取机制，没有考虑consumer处理能力

优化：application.yml中设置预取1条，处理完才能取下一条

image-20230702164555911
image-20230702164102178

9.3.4 发布订阅

前面两个模式都是消息”阅后即焚”，发布订阅模式允许将同一消息发送给多个消费者，实现方式是加入了exchange（交换机）.

（感觉消息中间件MQ和redis的发布订阅模式实在是很像啊，不过MQ应该更专业些，后期出一期redis的发布订阅笔记对比下.）

常见exchange类型：

  • Fanout：广播
  • Direct：路由
  • Topic：话题

注意：

  • exchange类型决定了消息转发规则
  • exchange只负责消息转发，不负责存储，路由失败则消息丢失
image-20230702165147561
FanoutExchange

FanoutExchange将接收到的消息路由到每一个跟其绑定的queue.

image-20230702173321104

案例流程：

1.在consumer中声明queue和exchange，并将queue和exchange绑定

 @Configuration
 public class FanoutConfig {
     // itcast.fanout
     @Bean
     public FanoutExchange fanoutExchange(){
         return new FanoutExchange("itcast.fanout");
     }
 ​
     // fanout.queue1
     @Bean
     public Queue fanoutQueue1(){
         return new Queue("fanout.queue1");
     }
 ​
     // 绑定队列1到交换机
     @Bean
     public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
         return BindingBuilder
                 .bind(fanoutQueue1)
                 .to(fanoutExchange);
     }
 ​
     // fanout.queue2同1
 }

2.在consumer中定义不同queue的消费行为

 @RabbitListener(queues = "fanout.queue1")
 public void listenFanoutQueue1(String msg) {
     System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
 }
 @RabbitListener(queues = "fanout.queue2")
 public void listenFanoutQueue2(String msg) {
     System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");
 }

3.在publisher中向exchange发送消息

 @Test
 public void testSendFanoutExchange() {
     // 交换机名称
     String exchangeName = "itcast.fanout";
     // 消息
     String message = "hello, every one!";
     // 发送消息
     rabbitTemplate.convertAndSend(exchangeName, "", message);
 }

结果

image-20230702172933404
image-20230702172640904
image-20230702172602706
DirectExchange

DirectExchange将接收到的消息根据规则路由到指定的queue，因此称为路由模式(routes).

  • 每一个Queue都与Exchange设置一个BindingKey
  • 发布者发送消息时，指定消息的RoutingKey
  • Exchange将消息路由到BindingKey与RoutingKey一致的队列
image-20230702192013203

案例流程：

1.利用@RabbitListener声明Exchange、Queue、RoutingKey；

消费者定义消费行为，监听对应Queue

 //FanoutExchange中使用的是手动声明Exchange、Queue、Binding的Bean的方式，这里采用更简单的写法，即直接使用@RabbitListener注解，Spring根据注解自动生成对应Bean.
 @RabbitListener(bindings = @QueueBinding(
     value = @Queue(name = "direct.queue1"),
     exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
     key = {"red", "blue"}
 ))
 public void listenDirectQueue1(String msg){
     System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
 }

2.发布者发送消息到指定Exchange，同时带上RoutingKey

 @Test
 public void testSendDirectExchange() {
     // 交换机名称
     String exchangeName = "itcast.direct";
     // 消息
     String message = "hello, red!";
     // 发送消息
     rabbitTemplate.convertAndSend(exchangeName, "red", message);
 }

结果

image-20230702193637840
TopicExchange

TopicExchange与DirectExchange类似，区别在于RoutingKey必须是多个单词的列表，并且以.分割。

Queue与Exchange指定BindingKey时可以使用通配符：

  • #代指0个或多个单词
  • *代指一个单词
image-20230702194200909

案例流程：

1.利用@RabbitListener声明Exchange、Queue、RoutingKey；

在consumer服务中定义消费行为并监听对应queue

 @RabbitListener(bindings = @QueueBinding(
     value = @Queue(name = "topic.queue1"),
     exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
     key = "china.#"
 ))
 public void listenTopicQueue1(String msg){
     System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
 }

2.在publisher中向对应exchange发送消息

 @Test
 public void testSendTopicExchange() {
     // 交换机名称
     String exchangeName = "itcast.topic";
     // 消息
     String message = "今天天气不错，我的心情好极了!";
     // 发送消息
     rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);
 }

结果

image-20230702194648507
消息转换器⭐

消息队列中的消息类型其实是Object类型的，支持各种类型的消息传送.

image-20230702195140170

案例：

1.采用Bean方式声明queue（使用Listener会让消息立即被处理）

 @Bean
 public Queue objectQueue(){
     return new Queue("object.queue");
 }
 //Listener写法
  @RabbitListener(queues = "object.queue")
 public void listenObjectQueue(Map<String,Object> msg){
     System.out.println("接收到object.queue的消息：" + msg);
 }

2.发布者向queue发送消息

 @Test
 public void testObjectQueue(){
     Map<String,Object> msg = new HashMap<>();
     msg.put("name","白敬亭");
     msg.put("age","30");
     rabbitTemplate.convertAndSend("object.queue",msg);
 }

3.消息发布者和订阅者都使用其他序列化方式，否则默认使用jdk自带的序列化方式结果是乱码，效率低，安全性也低（注入）⭐

 <dependency>
     <groupId>com.fasterxml.jackson.core</groupId>
     <artifactId>jackson-databind</artifactId>
 </dependency>
 @Bean
 public MessageConverter messageConverter(){
     return new Jackson2JsonMessageConverter();
 }

结果

image-20230702195846539

相似文章：

MySQL报错: INSERT command denied to user ‘root’@’ip’…

OSS对象存储

Spring Security和JWT

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注