文章目录
- 一、初识 MQ
- 1.1 同步调用:
- 1.2 异步调用:
- 二、RabbitMQ
- 三、SpringAMQP
- 3.1 依赖和配置文件
- 3.2 消息发送和接收:
- 3.2.1 消息发送:
- 3.2.2 消息接收:
- 3.3 WorkQueues 模型:
- 3.4 交换机类型:
- 3.4.1 Fanout 交换机:
- 3.4.2 Direct 交换机:
- 3.4.3 Topic 交换机:
- 3.5 声明交换机和队列:
- 3.5.1 使用 API:
- 3.5.1.1 fanout 示例:
- 3.5.1.2 direct 示例:
- 3.5.2 使用注解:
- 3.5.2.1 fanout 示例:
- 3.5.2.2 direct 示例:
- 3.6 消息转化器:
一、初识 MQ
1.1 同步调用:
我们观察下,下面这个余额支付功能的流程图:
如果我们采用的是基于 OpenFeign 的同步调用,也就是说业务执行流程是这样的:
- 支付服务需要先调用用户服务完成余额扣减。
- 然后支付服务自己要更新支付流水单的状态。
- 然后支付服务调用交易服务,更新业务订单状态为已支付。
三个步骤依次执行。
这其中就存在3个问题:
- 拓展性差:
我们目前的业务相对简单,但是随着业务规模扩大,产品的功能也在不断完善。
在大多数电商业务中,用户支付成功后都会以短信或者其它方式通知用户,告知支付成功。假如后期产品经理提出这样新的需求,你怎么办?是不是要在上述业务中再加入通知用户的业务?
某些电商项目中,还会有积分或金币的概念。假如产品经理提出需求,用户支付成功后,给用户以积分奖励或者返还金币,你怎么办?是不是要在上述业务中再加入积分业务、返还金币业务?
最终你的支付业务会越来越臃肿:
也就是说每次有新的需求,现有支付逻辑都要跟着变化,代码经常变动,不符合开闭原则,拓展性不好。
- 性能下降:
由于我们采用了同步调用,调用者需要等待服务提供者执行完返回结果后,才能继续向下执行,也就是说每次远程调用,调用者都是阻塞等待状态。最终整个业务的响应时长就是每次远程调用的执行时长之和:
假如每个微服务的执行时长都是 50ms,则最终整个业务的耗时可能高达 300ms,性能太差了。
- 级联失败:
由于我们是基于 OpenFeign 调用交易服务、通知服务。当交易服务、通知服务出现故障时,整个事务都会回滚,交易失败。
这其实就是同步调用的级联失败问题。
不能因为短信通知、更新订单状态失败而回滚整个事务(这些都不是支付服务的主线任务)。
而要解决这些问题,我们就必须用异步调用的方式来代替同步调用。
1.2 异步调用:
异步调用方式其实就是基于消息通知的方式,一般包含三个角色:
- 消息发送者:投递消息的人,就是原来的调用方。
- 消息Broker:管理、暂存、转发消息。
- 消息接收者:接收和处理消息的人,就是原来的服务提供方。
还是以余额支付业务为例:
除了扣减余额、更新支付流水单状态以外,其它调用逻辑全部取消。而是改为发送一条消息到 Broker。而相关的微服务都可以订阅消息通知,一旦消息到达 Broker,则会分发给每一个订阅了的微服务,处理各自的业务。
假如产品经理提出了新的需求,比如要在支付成功后更新用户积分。支付代码完全不用变更,而仅仅是让积分服务也订阅消息即可:
不管后期增加了多少消息订阅者,作为支付服务来讲,执行问扣减余额、更新支付流水状态后,发送消息即可。业务耗时仅仅是这三部分业务耗时,仅仅 100ms,大大提高了业务性能。
另外,不管是交易服务、通知服务,还是积分服务,他们的业务与支付关联度低。现在采用了异步调用,解除了耦合,他们即便执行过程中出现了故障,也不会影响到支付服务。
综上,异步调用的优势包括:
- 耦合度更低
- 性能更好
- 业务拓展性强
- 故障隔离,避免级联失败
当然,异步通信也并非完美无缺,它存在下列缺点:
- 完全依赖于 Broker 的可靠性、安全性和性能
- 架构复杂,后期维护和调试麻烦
消息 Broker,目前常见的实现方案就是消息队列(MessageQueue),简称为 MQ。
下面我们就来进行 RabbitMQ 的学习。
二、RabbitMQ
RabbitMQ 是基于 Erlang 语言开发的开源消息通信中间件,官网地址:
https://www.rabbitmq.com/
接下来,我们就学习它的基本概念和基础用法。
安装:
基于 Docker 来安装 RabbitMQ,使用下面的命令即可:
docker run \-e RABBITMQ_DEFAULT_USER=itheima \-e RABBITMQ_DEFAULT_PASS=123321 \-v mq-plugins:/plugins \--name mq \--hostname mq \-p 15672:15672 \-p 5672:5672 \--network hm-net\-d \rabbitmq:3.8-management
可以看到在安装命令中有两个映射的端口:
- 15672:RabbitMQ提供的管理控制台的端口
- 5672:RabbitMQ的消息发送处理接口
安装完成后,我们访问 http://ip地址:15672
即可看到管理控制台。首次访问需要登录,默认的用户名和密码在配置文件中已经指定了。
登录后即可看到管理控制台总览页面:
RabbitMQ 对应的架构如图:
其中包含几个概念:
publisher
:生产者。consumer
:消费者。queue
:队列,存储消息。生产者投递的消息会暂存在消息队列中,等待消费者处理。exchange
:交换机,负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。virtual host
:虚拟主机,起到数据隔离的作用。每个虚拟主机相互独立,有各自的 exchange、queue。这个就像一个 mysql 有多个数据库每个数据库有自己的表。
在 MQ 的客户端也可以实现收发消息,比较简单,所以这里跳过。
三、SpringAMQP
将来我们开发业务功能的时候,肯定不会在控制台收发消息,而是应该基于编程的方式。由于RabbitMQ
采用了 AMQP 协议,因此它具备跨语言的特性。任何语言只要遵循 AMQP 协议收发消息,都可以与RabbitMQ
交互。并且RabbitMQ
官方也提供了各种不同语言的客户端。
但是,RabbitMQ 官方提供的 Java 客户端编码相对复杂,一般生产环境下我们更多会结合 Spring 来使用。而 Spring 的官方刚好基于RabbitMQ 提供了这样一套消息收发的模板工具:SpringAMQP。并且还基于 SpringBoot 对其实现了自动装配,使用起来非常方便。
SpringAmqp 的官方地址:
https://spring.io/projects/spring-amqp
SpringAMQP 提供了三个功能:
- 自动声明队列、交换机及其绑定关系。
- 基于注解的监听器模式,异步接收消息。
- 封装了 RabbitTemplate 工具,用于发送消息。
3.1 依赖和配置文件
要进行下面的操作,首先我们需要引入相关依赖:
<!--AMQP依赖,包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
其次进行配置文件:
在application.yml
中添加配置:
spring:rabbitmq:host: xxxxx # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码
注意:不论是消息的发送者还是消息的接收者都需要进行上面的配置。
3.2 消息发送和接收:
3.2.1 消息发送:
利用RabbitTemplate
实现消息发送,RabbitTemplate 在我们引入依赖和配置完文件启动项目就自动创建了,我们直接进行依赖注入即可。
下面的代码中,没有实现交换机,因为比较复杂,我们下面讲,这里其实就是感受 RabbitTemplate 的使用。队列需要我们在 MQ 控制台手动创建。
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;public void testSimpleQueue() {// 队列名称String queueName = "simple.queue";// 消息String message = "hello, spring amqp!";// 发送消息rabbitTemplate.convertAndSend(queueName, message);}
}
3.2.2 消息接收:
@Component
public class SpringRabbitListener {// 利用RabbitListener来声明要监听的队列信息// 将来一旦监听的队列中有了消息,就会推送给当前服务,调用当前方法,处理消息。@RabbitListener(queues = "simple.queue")public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println("spring 消费者接收到消息:【" + msg + "】");}
}
注意:消息接收的方法参数类型要和消息的发送的参数类型一致。例如上面都是 String 类型。
3.3 WorkQueues 模型:
Work queues,任务模型。简单来说就是让多个消费者绑定到一个队列,共同消费队列中的消息。
当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。
此时就可以使用work 模型,多个消费者共同处理消息处理,同一条消息只会被一个消费者处理,这样,消息处理的速度就能大大提高了。
不过在 RabbitMQ 中该模型还存在一个问题:消息是平均分配给每个消费者,并没有考虑到消费者的处理能力。
这显然是不合理的。
改进:
在spring中有一个简单的配置,可以解决这个问题,在 application.yml 文件中添加配置:
spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息
这样就能实现能者多劳的效果,充分利用了每一个消费者的处理能力,可以有效避免消息积压问题。
总结:
Work模型的使用:
- 多个消费者绑定到一个队列,同一条消息只会被一个消费者处理。
- 通过设置prefetch来控制消费者预取的消息数量。
3.4 交换机类型:
在之前的案例中,都没有交换机,生产者直接发送消息到队列。而一旦引入交换机,消息发送的模式会有很大变化:
可以看到,在订阅模型中,多了一个exchange角色,而且过程略有变化:
- Publisher:生产者,不再发送消息到队列中,而是发给交换机。
- Exchange:交换机,一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。
- Queue:消息队列也与以前一样,接收消息、缓存消息。不过队列一定要与交换机绑定。
- Consumer:消费者,与以前一样,订阅队列,没有变化。
Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!
交换机的类型有四种:
- Fanout:广播,将消息交给所有绑定到交换机的队列。
- Direct:订阅,基于 RoutingKey(路由 key)发送给订阅了消息的队列。
- Topic:通配符订阅,与 Direct 类似,只不过 RoutingKey 可以使用通配符。
- Headers:头匹配,基于 MQ 的消息头匹配,用的较少。
我们这里讲解前面三种交换机。
3.4.1 Fanout 交换机:
在广播模式下,消息发送流程是这样的:
总结:
交换机的特点:
- 接收 publisher 发送的消息,将消息按照规则路由到与之绑定的队列。
- 不能缓存消息,路由失败,消息丢失。
- FanoutExchange 的会将消息路由到每个绑定的队列。
3.4.2 Direct 交换机:
在 Fanout 模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到 Direct 类型的 Exchange。
在 Direct 模型下:
- 队列与交换机的绑定,不能是任意绑定了,而是要指定一个
RoutingKey
(路由key)。 - 消息的发送方在向 Exchange 发送消息时,也必须指定消息的
RoutingKey
。 - Exchange 不再把消息交给每一个绑定的队列,而是根据消息的
Routing Key
进行判断,只有队列的Routingkey
与消息的Routing key
完全一致,才会接收到消息。
总结:
描述下 Direct 交换机与 Fanout 交换机的差异?
- Fanout 交换机将消息路由给每一个与之绑定的队列。
- Direct 交换机根据 RoutingKey 判断路由给哪个队列。
- 如果 Direct 交换机绑定的多个队列具有相同的 RoutingKey,则与 Fanout 功能类似。
3.4.3 Topic 交换机:
Topic
类型的Exchange
与Direct
相比,都是可以根据RoutingKey
把消息路由到不同的队列。
只不过Topic
类型Exchange
可以让队列在绑定BindingKey
的时候使用通配符!
**BindingKey 一般都是由一个或多个单词组成,多个单词之间以.
分割。**例如: item.insert
通配符规则:
#
:匹配 0 个或多个词。*
:匹配不多不少恰好 1 个词。
总结:
描述下 Direct 交换机与 Topic 交换机的差异?
- Topic 交换机与队列绑定时的 bindingKey 可以指定通配符。
#
:代表 0 个或多个词。*
:代表 1 个词。
3.5 声明交换机和队列:
在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时,队列和交换机是程序员定义的,将来项目上线,又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来,交给运维。在这个过程中是很容易出现错误的。
因此推荐的做法是由程序启动时检查队列和交换机是否存在,如果不存在自动创建。
**声明交换机和队列一般是消息的接收者来做的。**这里的背景是微服务,消息的发送者和接收者一般是在不同的项目里面。
注意:如果项目中没有消费者(使用 @RabbitListener),可能创建交换机和队列会失败(MQ 上显示不出来,但是 spring 中对象成功创建),至于为什么会这样,我也不清楚,去网上找,问 AI 都没有这方面的资料。这个问题是我在项目中遇到的,解决了很久,算是比较小众的问题,如果有相同情况的友友,可以试试)。
3.5.1 使用 API:
SpringAMQP 提供了一个 Queue 类,用来创建队列:
SpringAMQP 还提供了一个 Exchange 接口,来表示所有不同类型的交换机:
我们可以自己创建队列和交换机,不过 SpringAMQP 还提供了 ExchangeBuilder 来简化这个过程:
而在绑定队列和交换机时,则需要使用 BindingBuilder 来创建 Binding 对象:
3.5.1.1 fanout 示例:
package com.itheima.consumer.config;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类型交换机*/@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("hmall.fanout");}/*** 第1个队列*/@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}
3.5.1.2 direct 示例:
direct 模式由于要绑定多个 KEY,会非常麻烦,每一个 Key 都要编写一个 binding:
package com.itheima.consumer.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DirectConfig {/*** 声明交换机* @return Direct类型交换机*/@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange("hmall.direct").build();}/*** 第1个队列*/@Beanpublic Queue directQueue1(){return new Queue("direct.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");}/*** 第2个队列*/@Beanpublic Queue directQueue2(){return new Queue("direct.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");}
}
由于 Topic 交换机的声明和 Direct 差不多,大家照着上面的 Direct ,修改一下类型就能成功创建,这里就不演示了。
3.5.2 使用注解:
基于 @Bean 的方式声明队列和交换机比较麻烦,尤其是 direct 交换机,Spring 还提供了基于注解方式来声明。
3.5.2.1 fanout 示例:
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "fanout.queue1"),exchange = @Exchange(name = "hmall.fanout", type = ExchangeTypes.FANOUT)
))
public void listenFanoutQueue1(String msg){System.out.println("消费者1接收到fanout.queue1的消息:【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "fanout.queue2"),exchange = @Exchange(name = "hmall.fanout", type = ExchangeTypes.FANOUT)
))
public void listenFanoutQueue2(String msg){System.out.println("消费者2接收到fanout.queue2的消息:【" + msg + "】");
}
3.5.2.2 direct 示例:
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消费者1接收到direct.queue1的消息:【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){System.out.println("消费者2接收到direct.queue2的消息:【" + msg + "】");
}
3.6 消息转化器:
convertAndSend 方法可以任意类型的消息。
而在数据传输时,它会把你发送的消息序列化为字节发送给 MQ,接收消息的时候,还会把字节反序列化为 Java 对象。
只不过,默认情况下 Spring 采用的序列化方式是 JDK 序列化。JDK 序列化存在下列问题:
- 数据体积过大
- 有安全漏洞
- 可读性差
所以我们需要换一个消息转化器。
这里采用 JSON 方式来做序列化和反序列化。
在发送者和接受者两个服务中都引入依赖。
<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>
注意,如果项目中引入了spring-boot-starter-web
依赖,则无需再次引入Jackson
依赖。
配置消息转换器,在发送者和接受者两个服务的启动类中添加一个 Bean 即可:
@Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;
}
消息转换器中添加的 messageId 可以便于我们将来做幂等性判断。
参考文献:
- 黑马程序员
结语:
其实写博客不仅仅是为了教大家,同时这也有利于我巩固知识点,和做一个学习的总结,由于作者水平有限,对文章有任何问题还请指出,非常感谢。如果大家有所收获的话,还请不要吝啬你们的点赞收藏和关注,这可以激励我写出更加优秀的文章。