0


微服务的异步通信技术RabbitMQ

文章目录

前言

MQ的出现进一步降低了微服务模块之间的耦合度,相比于同步通信而言减少了关联服务的等待时间,使消息的传递更加多变,灵活
不管什么东西,只要被Spring整合就会变得十分简单,RabbitMQ也不例外
使用SpringAMQP来实现消息收发,不需要重复地配置连接参数,解决了一部分“硬编码”的问题。可以说和MyBatis整合JDBC非常相似。
在以前,使用原生的RabbitMQ收发消息是这样的:

使用SpringAMQP后收发消息是这样的:
这就是一个基本队列(Basic-Queue)
在这里插入图片描述
可以看到,只要引入依赖

spring-boot-starter-amqp

,写好yml配置文件,建立连接、创建通道的工作Spring都为我们做好了,而我们要做的仅仅就是利用工具类发送、监听消息,可以说相当的方便!
针对不同的场景,我们要使用不同的队列模型:

1.WorkQueue(工作队列)

在这里插入图片描述
对于单一消费者情况(简单队列),当生产者每秒发送50条消息,消费者每秒处理40条消息,这样每秒钟就会多出10条消息无法处理,由此就会产生生产过剩而导致消息堆积在队列中,一旦达到队列内存的上限,新来的消息就无法被处理而被丢弃。
为了提高消息处理的速度,避免队列中消息的堆积可以将队列绑定多个消费者,即WorkQueue
在这里插入图片描述
为了方便观察控制台,一般这样设计:
生产者:

@TestpublicvoidtestSendMessage2WorkQueue()throwsInterruptedException{String queueName ="work.queue";String message ="hello, MQQ";for(int i =1; i <=50; i++){
        rabbitTemplate.convertAndSend(queueName, message + i);Thread.sleep(20);}}

消费者:
为了尽可能的模拟真实场景(消费者处理消息的能力不同),所以设置两个消费者的sleep参数为不同的两个时间

@RabbitListener(queues ="work.queue")publicvoidlistenWorkQueue1(String msg)throwsInterruptedException{System.out.println("消费者1接收到消息——【"+ msg +"】"+"At "+LocalTime.now());Thread.sleep(20);}@RabbitListener(queues ="work.queue")publicvoidlistenWorkQueue2(String msg)throwsInterruptedException{System.err.println("消费者2接收到消息——【"+ msg +"】"+"At "+LocalTime.now());Thread.sleep(200);}

消息预取机制

运行后观察控制台,发现所有的消息处理完时间竟然花费了差不多五秒钟,很显然这样的效率是非常低下的:
在这里插入图片描述
为何绑定了两个消费者消费消息的速度不快反慢了呢?
仔细观察控制台会发现50条消息是被平均分配的,两个消费者分别消费id为偶数、奇数的消息,这样一想好像是处理能力差的消费者拖了后腿(200msX25=5000ms=5s),为什么会出现这样的情况?
这是由于MQ存在消息预取机制,即消费者会在处理之前预先拿到消息的通道,然后逐个处理消息,这个过程是与处理消息相隔离的!
如果还有人不明白就想想使用原生的RabbitMQ时我们是怎么处理消息的:
在这里插入图片描述
当执行完回调函数有可能消息都不会被处理,这时程序会继续向下执行,过段时间才会开始处理消息(其实我认为
这也是体现RabbitMQ异步的一个地方

这样的机制存在是保证异步性的关键,通过人为的设置参数也可以将消息预取的方式做出调整,来保证处理的效率,就像这样:

listener:simple:prefetch:1

通过prefetch参数来保证消费者每次获取消息的个数,以及处理完成后才能获取下一个批次的消息
进行数据预取设置后消费者在一秒之内处理完了所有的消息:
在这里插入图片描述
由此可见对于WorkQueue中消费者的设置要进行“按劳分配”的策略才较为完美
使用工作队列WorkQueue之后处理消息的效率得到了很大的提升,并且也不会出现消息堆积的情况

2.Publish&Subscribe(发布-订阅)

对于简单队列和工作队列模型,生产者发布消息,消费者一旦消费完,消息就会被销毁。这样无法做到将一个消息同时发送给多个消费者。
在这里插入图片描述
对于一个微服务项目,在支付订单的模型中当支付服务完成,会发消息同时去通知短信服务、订单服务…这就需要保证消息的高可用,不能一个服务消费完消息就被销毁而导致其他服务接收不到消息
如何做到将同一消息发送给多个消费者并让其各自接收到?采用发布&订阅的工作模型即可
在这里插入图片描述
通过交换机(exchange)将消息路由到不同的队列中,再由消费者来消费各自订阅队列中的消息
针对不同的交换机种类会有不同的发布策略:

1.Fanout(广播)

SpringAMQPA提供了声明交换机、队列、绑定关系的API
在这里插入图片描述
所以使用Exchange接口下的实现类就可以实现将消息路由到每一个绑定的Queue中,使用代码就会变得非常简单,声明交换机并绑定队列即可:

@BeanpublicFanoutExchangefanoutExchange(){returnnewFanoutExchange("yu7.fanout");}// fanout.queue1@BeanpublicQueuefanoutQueue1(){returnnewQueue("fanout.queue1");}// 绑定队列1到交换机@BeanpublicBindingfanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){returnBindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}// fanout.queue2@BeanpublicQueuefanoutQueue2(){returnnewQueue("fanout.queue2");}// 绑定队列2到交换机@BeanpublicBindingfanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){returnBindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}

在这里插入图片描述
注意:交换机只能用作消息的转发路由,不能用作消息的存储,一旦路由失败消息就会丢失!

2.DirectExchange(路由)

DirectExchange会将接收到的消息根据规则路由到指定的Queue中,生产者发布消息时指定消息的RoutingKey与消费者声明的bindingKey相匹配,从而达到“精确制导”
在这里插入图片描述
为了简化开发不使用声明Bean的方式来完成配置,通过@RabbitListener注解即可一键完成,所以根本不需要使用配置类:

生产者:

@TestpublicvoidtestSendDirectExchange(){// 交换机名称String exchangeName ="yu7.direct";// 消息String message ="hello, MQ!";// 发送消息
    rabbitTemplate.convertAndSend(exchangeName,"A", message);}

消费者:

@RabbitListener(bindings =@QueueBinding(
        value =@Queue(name ="direct.queue1"),
        exchange =@Exchange(name ="yu7.direct", type =ExchangeTypes.DIRECT),
        key ={"A","B"}))publicvoidlistenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息:【"+ msg +"】");}@RabbitListener(bindings =@QueueBinding(
        value =@Queue(name ="direct.queue2"),
        exchange =@Exchange(name ="yu7.direct", type =ExchangeTypes.DIRECT),
        key ={"A","C"}))publicvoidlistenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue2的消息:【"+ msg +"】");}

在这里插入图片描述
当队列的bindingKey都相同时就变成了广播模型

3.TopicExchange(话题)

Topic与Direct非常相似,他允许RoutingKey-BindingKey以通配符的形式进行匹配,这样就可以更加有针对性的路由、订阅更多消息,并且以前用多个BindingKey的情况现在只需要用一个就能解决:
#:代表0或者多个单词
*:代指一个单词
在这里插入图片描述

MQ的优点

1、耦合度低:每次有新需求,只需要添加对应的订阅即可
2、吞吐量提升:各自处理自己订阅的事件,不需要等待执行完毕后再释放资源
3、故障隔离:因为没有强依赖,中间某一环节出了问题,不会影响整个流程
4、流量削峰:MQ就像—根管道,大量请求来了,你们给我排好队,依次执行


本文转载自: https://blog.csdn.net/weixin_57535055/article/details/128794993
版权归原作者 懒羊羊.java 所有, 如有侵权,请联系我们删除。

“微服务的异步通信技术RabbitMQ”的评论:

还没有评论