本文由 简悦 SimpRead 转码, 原文地址 zhuanlan.zhihu.com
延时任务
在开发中,往往会遇到一些关于延时任务的需求。例如
- 生成订单 30 分钟未支付,则自动取消
- 生成订单 60 秒后, 给用户发短信
对上述的任务,我们给一个专业的名字来形容,那就是延时任务。那么这里就会产生一个问题,这个延时任务和定时任务的区别究竟在哪里呢?一共有如下几点区别
定时任务有明确的触发时间,延时任务没有
定时任务有执行周期,而延时任务在某事件触发后一段时间内执行,没有执行周期
定时任务一般执行的是批处理操作是多个任务,而延时任务一般是单个任务
取消订单还涉及到 释放库存,释放优惠券,这是需要调上游服务的;如果单纯在用户登录的时候,或者打开我的订单,或者在下单的时候触发一次,虽然过期订单可以取消,但时效性太差,库存在这之前一直没有释放。
方案分析
(1) 数据库轮询
思路
该方案通常是在小型项目中使用,即通过一个线程定时的去扫描数据库,通过订单时间来判断是否有超时的订单,
下单的时候存一个最晚支付时间,并传给支付中心。一个job专门扫描过最晚支付时间的订单,将命中订单置取消。
然后进行 update 或 delete 等操作;如果要控制在秒级内就取消订单,定时任务需要每隔一秒执行一次。即使这样,依旧存在延时,因为循环扫码数据库需要时间,甚至扫描的时间超过1秒
优缺点
优点: 简单易行,支持集群操作
缺点:
(1) 对服务器内存消耗大
(2) 存在延迟,比如你每隔 3 分钟扫描一次,那最坏的延迟时间就是 3 分钟
(3) 假设你的订单有几千万条,每隔几分钟这样扫描一次,数据库损耗极大
(2)JDK 的延迟队列
思路
该方案是利用 JDK 自带的 DelayQueue 来实现,这是一个无界阻塞队列,该队列只有在延迟期满的时候才能从中获取元素,放入 DelayQueue 中的对象,是必须实现 Delayed 接口的。
DelayedQueue 实现工作流程如下图所示
优缺点
优点: 效率高, 任务触发时间延迟低。
缺点:
(1) 服务器重启后,数据全部消失,怕宕机
(2) 集群扩展相当麻烦
(3) 因为内存条件限制的原因,比如下单未付款的订单数太多,那么很容易就出现 OOM 异常
(4) 代码复杂度较高
(3) 时间轮算法
思路
先上一张时间轮的图
时间轮算法可以类比于时钟,如上图箭头(指针)按某一个方向按固定频率轮动,每一次跳动称为一个 tick。这样可以看出定时轮由个 3 个重要的属性参数,ticksPerWheel(一轮的 tick 数),tickDuration(一个 tick 的持续时间)以及 timeUnit(时间单位),例如当 ticksPerWheel=60,tickDuration=1,timeUnit = 秒,这就和现实中的始终的秒针走动完全类似了。
如果当前指针指在 1 上面,我有一个任务需要 4 秒以后执行,那么这个执行的线程回调或者消息将会被放在 5 上。那如果需要在 20 秒之后执行怎么办,由于这个环形结构槽数只到 8,如果要 20 秒,指针需要多转 2 圈。位置是在 2 圈之后的 5 上面(20 % 8 + 1)
优缺点
优点: 效率高, 任务触发时间延迟时间比 delayQueue 低,代码复杂度比 delayQueue 低。
缺点:
(1) 服务器重启后,数据全部消失,怕宕机
(2) 集群扩展相当麻烦
(3) 因为内存条件限制的原因,比如下单未付款的订单数太多,那么很容易就出现 OOM 异常
(4)redis 缓存
思路
利用 redis 的 zset,zset 是一个有序集合,每一个元素 (member) 都关联了一个 score, 通过 score 排序来取集合中的值
添加元素: ZADD key score member [[score member] [score member] …]
按顺序查询元素: ZRANGE key start stop [WITHSCORES]
查询元素 score:ZSCORE key member
移除元素: ZREM key member [member …]
测试如下
添加单个元素
redis> ZADD page_rank 10 google.com
(integer) 1
添加多个元素
redis> ZADD page_rank 9 baidu.com 8 bing.com
(integer) 2
按顺序查询元素,同时显示分数
redis> ZRANGE page_rank 0 -1 WITHSCORES
1) "bing.com"
2) "8"
3) "baidu.com"
4) "9"
5) "google.com"
6) "10"
查询元素的score值
redis> ZSCORE page_rank bing.com
"8"
移除单个元素
redis> ZREM page_rank google.com
(integer) 1
redis> ZRANGE page_rank 0 -1 WITHSCORES
1) "bing.com"
2) "8"
3) "baidu.com"
4) "9"
那么如何实现呢?
我们将订单超时时间戳与订单号分别设置为 score 和 member, 系统扫描第一个元素判断是否超时,具体如下图所示
优缺点
优点:
(1) 由于使用 Redis 作为消息通道,消息都存储在 Redis 中。如果发送程序或者任务处理程序挂了,重启之后,还有重新处理数据的可能性。
(2) 做集群扩展相当方便
(3) 时间准确度高
缺点
(1) 需要额外进行 redis 维护
(5) 使用消息队列(推荐)
我们可以采用 rabbitMQ 的延时队列。RabbitMQ 具有以下两个特性,可以实现延迟队列
RabbitMQ 可以针对 Queue 和 Message 设置 x-message-tt,来控制消息的生存时间,如果超时,则消息变为 dead letter
RabbitMQ 的 Queue 可以配置 x-dead-letter-exchange 和 x-dead-letter-routing-key(可选)两个参数,用来控制队列内出现了 deadletter,则按照这两个参数重新路由。结合以上两个特性,就可以模拟出延迟消息的功能,
优缺点
优点: 高效, 可以利用 rabbitmq 的分布式特性轻易的进行横向扩展, 消息支持持久化增加了可靠性。
缺点:本身的易用度要依赖于 rabbitMq 的运维. 因为要引用 rabbitMq, 所以复杂度和成本变高
参考
http://blog.csdn.net/hjm4702192/article/details/80519010
淘宝订单未支付30分钟自动取消是如何实现的?
本文由 简悦 SimpRead 转码, 原文地址 zhuanlan.zhihu.com
延迟队列可以实现消息在投递到
Exchange
之后,经过一定的时间之后再投递到相应的Queue
。再被消费者监听消费。 即:生产者投递的消息经过一段时间之后再被消费者消费。
- 业务场景:订单在 30 分钟内还未支付则自动取消。
该业务的其他实现方案:
- 使用
Redis
,设置过期时间,监听过期事件。 使用 RabbitMQ 的过期队列与死信队列,设置消息的存活时间,在设置的时间内未被消费,即会投递到死信队列,我们监听死信队列即可。可参考上一篇文章 RabbitMQ 死信队列在 SpringBoot 中的使用。
使用 RabbitMQ 延迟队列实现:
RabbitMQ 相关代码编写
配置文件 ``` spring: rabbitmq: host: localhost port: 5672 username: futao password: 123456789 virtual-host: delay-vh connection-timeout: 15000
发送确认
publisher-confirms: true
路由失败回调
publisher-returns: true template:
# 必须设置成true 消息路由失败通知监听者,而不是将消息丢弃 mandatory: true
listener:
simple: # 每次从RabbitMQ获取的消息数量 prefetch: 1 default-requeue-rejected: false # 每个队列启动的消费者数量 concurrency: 1 # 每个队列最大的消费者数量 max-concurrency: 1 # 手动签收ACK acknowledge-mode: manual
app: rabbitmq:
# 延迟时长设置
delay:
order: 10S
# 队列定义
queue:
order:
delay: order-delay-queue
# 交换机定义
exchange:
order:
delay: order-delay-exchange
- 延迟交换机,队列定义与绑定
/**
- 队列,交换机定义与绑定
- 延迟队列插件
rabbitmq-delayed-message-exchange
下载地址 https://github.com/rabbitmq/rabbitmq-delayed-message-exchange * - @author futao
@date 2020/4/10. */ @Configuration public class Declare {
/**
- 订单队列 - 接收延迟投递的订单 *
- @param orderQueue 订单队列名称
@return */ @Bean public Queue orderDelayQueue(@Value(“${app.rabbitmq.queue.order.delay}”) String orderQueue) { return QueueBuilder
.durable(orderQueue) .build();
}
/**
- 订单交换机-延迟交换机 - 消息延迟一定时间之后再投递到绑定的队列 *
- @param orderExchange 订单延迟交换机
@return */ @Bean public Exchange orderDelayExchange(@Value(“${app.rabbitmq.exchange.order.delay}”) String orderExchange) { Map
args = new HashMap<>(1); args.put(“x-delayed-type”, “topic”); return new CustomExchange(orderExchange, “x-delayed-message”, true, false, args); } /**
- 订单队列-交换机 绑定 *
- @param orderQueue 订单队列
- @param orderDelayExchange 订单交换机
- @return
*/
@Bean
public Binding orderBinding(Queue orderDelayQueue, Exchange orderDelayExchange) {
return BindingBuilder
} } ```.bind(orderDelayQueue) .to(orderDelayExchange) .with("order.delay.*") .noargs();
可以看出队列就是普通的队列。重点在交换机的设定上。声明延迟交换机需要设置参数_x-delayed-type_
,值为交换机类型,可以是_fanout_
,_topic_
,_direct_
。并且设置交换机的 type 为_x-delayed-message_
。
- 定义完成后可以启动 SpringBoot 应用程序,在 RabbitMQ 管理后台查看 Exchange 和 Queue。
可以看到,除了默认的交换机,SpringBoot 已经帮我们创建好了延迟交换机_order-delay-exchange_
,并且此时 messages delayed 为 0,因为我们还未向交换机投递消息。
- 可以继续查看交换机的路由类型与绑定的队列
- 队列为普通的队列
- 回到代码中,定义消息生产者
在消息投递之前为每条消息都设置了延迟时长_setDelay()_
。 调用消费者的代码在上面_OrderController_
中,下定之后,订单数据落库,并且向 MQ 中投递延迟消息。可以回头看看。
- 消费者 - 监听过期的订单信息,并且将 DB 中相应的订单设置为已过期。
打印出了投递日志,订单主键为_666ae86aabe2a1b3120b34bb5f447bbe_
的订单在_2020-04-14 22:22:04.307_
进行了投递,此时数据库中该订单的状态为_100_
,待支付。
- 此时查看 Exchange 详情可以发现,
messages delayed:1
,即目前有一条消息处于延迟状态。
- 等待 10S 后。
可以看到_OrderConsumer_
在 10S 后_2020-04-14 22:22:14.320_
接收到了主键为_666ae86aabe2a1b3120b34bb5f447bbe_
的订单消息。距离投递时间2020-04-14 22:22:04.307
为 10S。此时查看 DB 中订单状态:
订单状态为_200_
已过期,且过期时间为_2020-04-14 22:22:14_
- 达到了订单在我们指定的时间后过期。
严重风险提示:
在实际业务使用中,如果消费者的消费能力比较低下,会存在已经过期的消息阻塞积压在队列,无法在指定的时间内过期,导致业务出现异常。 实际上,按照我们业务意图,队里 Queue 里是不应该有大量消息存在的,因为投递到过期队列的消息已经是过期了的,应该立即被消费掉。
- 进行测试: 为了降低消费者的消费能力,进行如下处理:
- 设置消费者的最大并发数为 1,并进行手动签收。
listener:
simple:
# 每个队列启动的消费者数量
concurrency: 1
# 每个队列最大的消费者数量
max-concurrency: 1
acknowledge-mode: manual
app:
rabbitmq:
delay:
# 订单过期时间为1分钟
order: 1M
- 消费者在处理消息时休眠 5S
- 向 MQ 投递两条消息,预期两条消息都在 1 分钟后正常过期。
- 结果 (去除了无关信息):
2020-04-15 20:18:05.269 OrderSender : 订单[d6fd965b11f8db0fafb762d305db83b0]投递到MQ
2020-04-15 20:18:05.765 OrderSender : 订单[77ceb7f1bfbbcaf627224ac75e96b0e5]投递到MQ
2020-04-15 20:19:05.279 OrderConsumer : 消费者接收到延迟订单[d6fd965b11f8db0fafb762d305db83b0]
2020-04-15 20:19:15.316 OrderConsumer : 订单业务处理结束.....进行消息ack签收
2020-04-15 20:19:15.318 OrderConsumer : 消费者接收到延迟订单[77ceb7f1bfbbcaf627224ac75e96b0e5]
2020-04-15 20:19:25.330 OrderConsumer : 订单业务处理结束.....进行消息ack签收
第一个订单_d6fd965b11f8db0fafb762d305db83b0_
投递时间为_2020-04-15 20:18:05.269_
。1 分钟后_2020-04-15 20:19:05.279_
接收到了通知,并且处理了 10S 后进行了签收 ack。
第二个订单_77ceb7f1bfbbcaf627224ac75e96b0e5_
投递时间为_2020-04-15 20:18:05.765_
。1 分钟过后并没有收到通知,而是在第一个订单处理完毕之后,_2020-04-15 20:19:15.318_
才收到了通知,比预期的时间长了 10 秒,实际延迟时间为 1 分钟 + 10 秒。出现了业务异常。
导致这个问题的原因就是消费者无法及时消费消息并更新订单状态。所以我们在进行开发时,需要考虑实际的数据量大小,消费者消费能力。及时关注队列消息积压情况,灵活调整消费者并发数量,优化消费者代码,提高消费者消费能力。
系列文章
- RabbitMQ 死信队列在 SpringBoot 中的使用