本文由 简悦 SimpRead 转码, 原文地址 zhuanlan.zhihu.com

延时任务

在开发中,往往会遇到一些关于延时任务的需求。例如

  • 生成订单 30 分钟未支付,则自动取消
  • 生成订单 60 秒后, 给用户发短信

对上述的任务,我们给一个专业的名字来形容,那就是延时任务。那么这里就会产生一个问题,这个延时任务和定时任务的区别究竟在哪里呢?一共有如下几点区别
定时任务有明确的触发时间,延时任务没有
定时任务有执行周期,而延时任务在某事件触发后一段时间内执行,没有执行周期
定时任务一般执行的是批处理操作是多个任务,而延时任务一般是单个任务

取消订单还涉及到 释放库存,释放优惠券,这是需要调上游服务的;如果单纯在用户登录的时候,或者打开我的订单,或者在下单的时候触发一次,虽然过期订单可以取消,但时效性太差,库存在这之前一直没有释放。

方案分析

(1) 数据库轮询

思路

该方案通常是在小型项目中使用,即通过一个线程定时的去扫描数据库,通过订单时间来判断是否有超时的订单,
下单的时候存一个最晚支付时间,并传给支付中心。一个job专门扫描过最晚支付时间的订单,将命中订单置取消。

然后进行 update 或 delete 等操作;如果要控制在秒级内就取消订单,定时任务需要每隔一秒执行一次。即使这样,依旧存在延时,因为循环扫码数据库需要时间,甚至扫描的时间超过1秒

优缺点

优点: 简单易行,支持集群操作

缺点:
(1) 对服务器内存消耗大
(2) 存在延迟,比如你每隔 3 分钟扫描一次,那最坏的延迟时间就是 3 分钟
(3) 假设你的订单有几千万条,每隔几分钟这样扫描一次,数据库损耗极大

(2)JDK 的延迟队列

思路
该方案是利用 JDK 自带的 DelayQueue 来实现,这是一个无界阻塞队列,该队列只有在延迟期满的时候才能从中获取元素,放入 DelayQueue 中的对象,是必须实现 Delayed 接口的。
DelayedQueue 实现工作流程如下图所示

超时订单自动关闭 - 图1

优缺点
优点: 效率高, 任务触发时间延迟低。

缺点:
(1) 服务器重启后,数据全部消失,怕宕机
(2) 集群扩展相当麻烦
(3) 因为内存条件限制的原因,比如下单未付款的订单数太多,那么很容易就出现 OOM 异常
(4) 代码复杂度较高

(3) 时间轮算法

思路
先上一张时间轮的图

超时订单自动关闭 - 图2

时间轮算法可以类比于时钟,如上图箭头(指针)按某一个方向按固定频率轮动,每一次跳动称为一个 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 …]

测试如下

  1. 添加单个元素
  2. redis> ZADD page_rank 10 google.com
  3. (integer) 1
  4. 添加多个元素
  5. redis> ZADD page_rank 9 baidu.com 8 bing.com
  6. (integer) 2
  7. 按顺序查询元素,同时显示分数
  8. redis> ZRANGE page_rank 0 -1 WITHSCORES
  9. 1) "bing.com"
  10. 2) "8"
  11. 3) "baidu.com"
  12. 4) "9"
  13. 5) "google.com"
  14. 6) "10"
  15. 查询元素的score
  16. redis> ZSCORE page_rank bing.com
  17. "8"
  18. 移除单个元素
  19. redis> ZREM page_rank google.com
  20. (integer) 1
  21. redis> ZRANGE page_rank 0 -1 WITHSCORES
  22. 1) "bing.com"
  23. 2) "8"
  24. 3) "baidu.com"
  25. 4) "9"

那么如何实现呢?
我们将订单超时时间戳与订单号分别设置为 score 和 member, 系统扫描第一个元素判断是否超时,具体如下图所示
超时订单自动关闭 - 图3

优缺点

优点:
(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。

超时订单自动关闭 - 图4

可以看到,除了默认的交换机,SpringBoot 已经帮我们创建好了延迟交换机_order-delay-exchange_,并且此时 messages delayed 为 0,因为我们还未向交换机投递消息。

  • 可以继续查看交换机的路由类型与绑定的队列

超时订单自动关闭 - 图5

  • 队列为普通的队列

超时订单自动关闭 - 图6

  • 回到代码中,定义消息生产者

超时订单自动关闭 - 图7

在消息投递之前为每条消息都设置了延迟时长_setDelay()_ 调用消费者的代码在上面_OrderController_中,下定之后,订单数据落库,并且向 MQ 中投递延迟消息。可以回头看看。

  • 消费者 - 监听过期的订单信息,并且将 DB 中相应的订单设置为已过期。

超时订单自动关闭 - 图8

打印出了投递日志,订单主键为_666ae86aabe2a1b3120b34bb5f447bbe_的订单在_2020-04-14 22:22:04.307_进行了投递,此时数据库中该订单的状态为_100_,待支付。

  • 此时查看 Exchange 详情可以发现,messages delayed:1,即目前有一条消息处于延迟状态。

超时订单自动关闭 - 图9

  • 等待 10S 后。

超时订单自动关闭 - 图10

可以看到_OrderConsumer_在 10S 后_2020-04-14 22:22:14.320_接收到了主键为_666ae86aabe2a1b3120b34bb5f447bbe_的订单消息。距离投递时间2020-04-14 22:22:04.307为 10S。此时查看 DB 中订单状态:

超时订单自动关闭 - 图11

订单状态为_200_已过期,且过期时间为_2020-04-14 22:22:14_

  • 达到了订单在我们指定的时间后过期。

严重风险提示:

在实际业务使用中,如果消费者的消费能力比较低下,会存在已经过期的消息阻塞积压在队列,无法在指定的时间内过期,导致业务出现异常。 实际上,按照我们业务意图,队里 Queue 里是不应该有大量消息存在的,因为投递到过期队列的消息已经是过期了的,应该立即被消费掉。

  • 进行测试: 为了降低消费者的消费能力,进行如下处理:
  1. 设置消费者的最大并发数为 1,并进行手动签收。
listener:
      simple:
        # 每个队列启动的消费者数量
        concurrency: 1
        # 每个队列最大的消费者数量
        max-concurrency: 1
        acknowledge-mode: manual

app:
  rabbitmq:
    delay:
      # 订单过期时间为1分钟
      order: 1M
  1. 消费者在处理消息时休眠 5S

超时订单自动关闭 - 图12

  1. 向 MQ 投递两条消息,预期两条消息都在 1 分钟后正常过期。
  2. 结果 (去除了无关信息):

超时订单自动关闭 - 图13

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 秒。出现了业务异常。