SpringCloud学习笔记(九)RabbitMQ

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

发表回复

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