1.延迟消息
在电商的支付业务中,对于一些库存有限的商品,为了更好的用户体验,通常都会在用户下单时立刻扣减商品库存。例如电影院购票、高铁购票,下单后就会锁定座位资源,其他人无法重复购买。
但是这样就存在一个问题,假如用户下单后一直不付款,就会一直占有库存资源,导致其他客户无法正常交易,最终导致商户利益受损!
因此,电商中通常的做法就是:对于超过一定时间未支付的订单,应该立刻取消订单并释放占用的库存。
例如,订单支付超时时间为30分钟,则我们应该在用户下单后的第30分钟检查订单支付状态,如果发现未支付,应该立刻取消订单,释放库存。
但问题来了:如何才能准确的实现在下单后第30分钟去检查支付状态呢?
像这种在一段时间以后才执行的任务,我们称之为延迟任务,而要实现延迟任务,最简单的方案就是利用MQ的延迟消息了。
在RabbitMQ中实现延迟消息也有两种方案:
- 死信交换机+TTL
- 延迟消息插件
这一章我们就一起研究下这两种方案的实现方式,以及优缺点。
1.1.死信交换机和延迟消息
首先我们来学习一下基于死信交换机的延迟消息方案。
1.1.1.死信交换机
什么是死信?
当一个队列中的消息满足下列情况之一时,可以成为死信(dead letter):
- 消费者使用
basic.reject
或basic.nack
声明消费失败,并且消息的requeue
参数设置为false - 消息是一个过期消息,超时无人消费
- 要投递的队列消息满了,无法投递
如果一个队列中的消息已经成为死信,并且这个队列通过**dead-letter-exchange**
属性指定了一个交换机,那么队列中的死信就会投递到这个交换机中,而这个交换机就称为死信交换机(Dead Letter Exchange)。而此时加入有队列与死信交换机绑定,则最终死信就会被投递到这个队列中。
死信交换机有什么作用呢?
- 收集那些因处理失败而被拒绝的消息
- 收集那些因队列满了而被拒绝的消息
- 收集因TTL(有效期)到期的消息
1.1.2.延迟消息
前面两种作用场景可以看做是把死信交换机当做一种消息处理的最终兜底方案,与消费者重试时讲的RepublishMessageRecoverer
作用类似。
而最后一种场景,大家设想一下这样的场景:
如图,有一组绑定的交换机(ttl.fanout
)和队列(ttl.queue
)。但是ttl.queue
没有消费者监听,而是设定了死信交换机hmall.direct
,而队列direct.queue1
则与死信交换机绑定,RoutingKey是blue:
假如我们现在发送一条消息到ttl.fanout
,RoutingKey为blue,并设置消息的有效期为5000毫秒:
:::warning
注意:尽管这里的ttl.fanout
不需要RoutingKey,但是当消息变为死信并投递到死信交换机时,会沿用之前的RoutingKey,这样hmall.direct
才能正确路由消息。
:::
消息肯定会被投递到ttl.queue
之后,由于没有消费者,因此消息无人消费。5秒之后,消息的有效期到期,成为死信:
死信被再次投递到死信交换机hmall.direct
,并沿用之前的RoutingKey,也就是blue
:
由于direct.queue1
与hmall.direct
绑定的key是blue,因此最终消息被成功路由到direct.queue1
,如果此时有消费者与direct.queue1
绑定, 也就能成功消费消息了。但此时已经是5秒钟以后了:
也就是说,publisher发送了一条消息,但最终consumer在5秒后才收到消息。我们成功实现了延迟消息。
1.1.3.总结
:::warning
注意:
RabbitMQ的消息过期是基于追溯方式来实现的,也就是说当一个消息的TTL到期以后不一定会被移除或投递到死信交换机,而是在消息恰好处于队首时才会被处理。
当队列中消息堆积很多的时候,过期消息可能不会被按时处理,因此你设置的TTL时间不一定准确。
:::
1.2.DelayExchange插件
基于死信队列虽然可以实现延迟消息,但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果。
官方文档说明:
Scheduling Messages with RabbitMQ | RabbitMQ - Blog
1.2.1.下载
插件下载地址:
GitHub - rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ
由于我们安装的MQ是3.8
版本,因此这里下载3.8.17
版本:
当然,也可以直接使用课前资料提供好的插件:
1.2.2.安装
因为我们是基于Docker安装,所以需要先查看RabbitMQ的插件目录对应的数据卷。
1 | docker volume inspect mq-plugins |
结果如下:
1 | [ |
插件目录被挂载到了/var/lib/docker/volumes/mq-plugins/_data
这个目录,我们上传插件到该目录下。
接下来执行命令,安装插件:
1 | docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange |
运行结果如下:
1.2.3.声明延迟交换机
基于注解方式:
1 |
|
基于@Bean
的方式:
1 | package com.itheima.consumer.config; |
1.2.4.发送延迟消息
发送消息时,必须通过x-delay属性设定延迟时间:
1 |
|
:::warning
注意:
延迟消息插件内部会维护一个本地数据库表,同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长,可能会导致堆积的延迟消息非常多,会带来较大的CPU开销,同时延迟消息的时间会存在误差。
因此,不建议设置延迟时间过长的延迟消息。
:::
1.3.订单状态同步问题
接下来,我们就在交易服务中利用延迟消息实现订单支付状态的同步。其大概思路如下:
假如订单超时支付时间为30分钟,理论上说我们应该在下单时发送一条延迟消息,延迟时间为30分钟。这样就可以在接收到消息时检验订单支付状态,关闭未支付订单。
但是大多数情况下用户支付都会在1分钟内完成,我们发送的消息却要在MQ中停留30分钟,额外消耗了MQ的资源。因此,我们最好多检测几次订单支付状态,而不是在最后第30分钟才检测。
例如:我们在用户下单后的第10秒、20秒、30秒、45秒、60秒、1分30秒、2分、…30分分别设置延迟消息,如果提前发现订单已经支付,则后续的检测取消即可。
这样就可以有效避免对MQ资源的浪费了。
优化后的实现思路如下:
由于我们要多次发送延迟消息,因此需要先定义一个记录消息延迟时间的消息体,处于通用性考虑,我们将其定义到hm-common
模块下:
代码如下:
1 | package com.hmall.common.domain; |
1.3.1.定义常量
无论是消息发送还是接收都是在交易服务完成,因此我们在trade-service
中定义一个常量类,用于记录交换机、队列、RoutingKey等常量:
内容如下:
1 | package com.hmall.trade.constants; |
1.3.2.抽取共享mq配置
我们将mq的配置抽取到nacos中,方便各个微服务共享配置。
在nacos中定义一个名为shared-mq.xml
的配置文件,内容如下:
1 | spring: |
这里只添加一些基础配置,至于生产者确认,消费者确认配置则由微服务根据业务自己决定。
在trade-service
模块添加共享配置:
1.3.3.改造下单业务
接下来,我们改造下单业务,在下单完成后,发送延迟消息,查询支付状态。
1)引入依赖
在trade-service
模块的pom.xml
中引入amqp的依赖:
1 | <!--amqp--> |
2)改造下单业务
修改trade-service
模块的com.hmall.trade.service.impl.OrderServiceImpl
类的createOrder
方法,添加消息发送的代码:
1.3.4.编写查询支付状态接口
由于MQ消息处理时需要查询支付状态,因此我们要在pay-service模块定义一个这样的接口,并提供对应的FeignClient.
首先,在hm-api模块定义三个类:
说明:
- PayOrderDTO:支付单的数据传输实体
- PayClient:支付系统的Feign客户端
- PayClientFallback:支付系统的fallback逻辑
PayOrderDTO代码如下:
1 | package com.hmall.api.dto; |
PayClient代码如下:
1 | package com.hmall.api.client; |
PayClientFallback代码如下:
1 | package com.hmall.api.client.fallback; |
最后,在pay-service模块的PayController中实现该接口:
1 |
|
1.3.5.消息监听
接下来,我们在trader-service编写一个监听器,监听延迟消息,查询订单支付状态:
代码如下:
1 | package com.hmall.trade.listener; |
注意,这里要在OrderServiceImpl中实现cancelOrder方法,留作作业大家自行实现。