• 欢迎来到本博客,希望可以y一起学习与分享

延时队列

笔记 benz 2个月前 (04-21) 7次浏览 0个评论 扫描二维码
文章目录[隐藏]

什么是延时队列

延时队列,首先,它是一种队列,队列意味着内部的元素是有序的,元素出队和入队是有方向性的,元素从一端进入,从另一端取出。

其次,延时队列,最重要的特性就体现在它的延时属性上,跟普通的队列不一样的是,普通队列中的元素总是等着希望被早点取出处理,而延时队列中的元素则是希望被在指定时间得到取出和处理,所以延时队列中的元素是都是带时间属性的,通常来说是需要被处理的消息或者任务。

简单来说,延时队列就是用来存放需要在指定时间被处理的元素的队列。

延时队列使用场景

那么什么时候需要用延时队列呢?考虑一下以下场景:

  1. 订单在十分钟之内未支付则自动取消。
  2. 新创建的店铺,如果在十天内都没有上传过商品,则自动发送消息提醒。
  3. 账单在一周内未支付,则自动结算。
  4. 用户注册成功后,如果三天内没有登陆则进行短信提醒。
  5. 用户发起退款,如果三天内没有得到处理则通知相关运营人员。
  6. 预定会议后,需要在预定的时间点前十分钟通知各个与会人员参加会议。

这些场景都有一个特点,需要在某个事件发生之后或者之前的指定时间点完成某一项任务,如:发生订单生成事件,在十分钟之后检查该订单支付状态,然后将未支付的订单进行关闭;发生店铺创建事件,十天后检查该店铺上新商品数,然后通知上新数为0的商户;发生账单生成事件,检查账单支付状态,然后自动结算未支付的账单;发生新用户注册事件,三天后检查新注册用户的活动数据,然后通知没有任何活动记录的用户;发生退款事件,在三天之后检查该订单是否已被处理,如仍未被处理,则发送消息给相关运营人员;发生预定会议事件,判断离会议开始是否只有十分钟了,如果是,则通知各个与会人员。

看起来似乎使用定时任务,一直轮询数据,每秒查一次,取出需要被处理的数据,然后处理不就完事了吗?如果数据量比较少,确实可以这样做,比如:对于“如果账单一周内未支付则进行自动结算”这样的需求,如果对于时间不是严格限制,而是宽松意义上的一周,那么每天晚上跑个定时任务检查一下所有未支付的账单,确实也是一个可行的方案。但对于数据量比较大,并且时效性较强的场景,如:“订单十分钟内未支付则关闭“,短期内未支付的订单数据可能会有很多,活动期间甚至会达到百万甚至千万级别,对这么庞大的数据量仍旧使用轮询的方式显然是不可取的,很可能在一秒内无法完成所有订单的检查,同时会给数据库带来很大压力,无法满足业务要求而且性能低下。

更重要的一点是,不!优!雅!

没错,作为一名有追求的程序员,始终应该追求更优雅的架构和更优雅的代码风格,写代码要像写诗一样优美。【滑稽】

这时候,延时队列就可以闪亮登场了,以上场景,正是延时队列的用武之地。

既然延时队列可以解决很多特定场景下,带时间属性的任务需求,那么如何构造一个延时队列呢?接下来,本文将介绍如何用RabbitMQ来实现延时队列。

RabbitMQ中的TTL

在介绍延时队列之前,还需要先介绍一下RabbitMQ中的一个高级特性——TTL(Time To Live)

TTL是什么呢?TTL是RabbitMQ中一个消息或者队列的属性,表明一条消息或者该队列中的所有消息的最大存活时间,单位是毫秒。换句话说,如果一条消息设置了TTL属性或者进入了设置TTL属性的队列,那么这条消息如果在TTL设置的时间内没有被消费,则会成为“死信”(至于什么是死信,请翻看上一篇)。如果同时配置了队列的TTL和消息的TTL,那么较小的那个值将会被使用。

那么,如何设置这个TTL值呢?有两种方式,第一种是在创建队列的时候设置队列的“x-message-ttl”属性,如下:

这样所有被投递到该队列的消息都最多不会存活超过10s。

另一种方式便是针对每条消息设置TTL,代码如下:

这样这条消息的过期时间也被设置成了10s。

但这两种方式是有区别的,如果设置了队列的TTL属性,那么一旦消息过期,就会被队列丢弃,而第二种方式,消息即使过期,也不一定会被马上丢弃,因为消息是否过期是在即将投递到消费者之前判定的,如果当前队列有严重的消息积压情况,则已过期的消息也许还能存活较长时间。

另外,还需要注意的一点是,如果不设置TTL,表示消息永远不会过期,如果将TTL设置为0,则表示除非此时可以直接投递该消息到消费者,否则该消息将会被丢弃。

如何利用RabbitMQ实现延时队列

实现延时队列的基本要素

存在一个倒计时机制:Time To Live(TTL)
当到达时间点的时候会触发一个发送消息的事件:Dead Letter Exchanges(DLX)
基于第一点,我利用的是消息存在过期时间这一特性, 消息一旦过期就会变成dead letter,可以让单独的消息过期,也可以设置整个队列消息的过期时间
而rabbitmq会有限取两个值的最小值

基于第二点,是用到了rabbitmq的过期消息处理机制:
x-dead-letter-exchange: 将过期的消息发送到指定的 exchange 中
x-dead-letter-routing-key: 将过期的消息发送到自定的 route当中
x-message-ttl:消息存活时间

想想看,延时队列,不就是想要消息延迟多久被处理吗,TTL则刚好能让消息在延迟多久之后成为死信,另一方面,成为死信的消息都会被投递到死信队列里,这样只需要消费者一直消费死信队列里的消息就万事大吉了,因为里面的消息都是希望被立即处理的消息。

生产者生产一条延时消息,根据需要延时时间的不同,利用不同的routingkey将消息路由到不同的延时队列,每个队列都设置了不同的TTL属性,并绑定在同一个死信交换机中,消息过期后,根据routingkey的不同,又会被路由到不同的死信队列中,消费者只需要监听对应的死信队列进行处理即可。

不过,等等,如果这样使用的话,岂不是每增加一个新的时间需求,就要新增一个队列,这里只有6s和60s两个时间选项,如果需要一个小时后处理,那么就需要增加TTL为一个小时的队列,如果是预定会议室然后提前通知这样的场景,岂不是要增加无数个队列才能满足需求??

RabbitMQ延时队列优化

显然,需要一种更通用的方案才能满足需求,那么就只能将TTL设置在消息属性里了。增加一个延时队列,用于接收设置为任意延时时长的消息,增加一个相应的死信队列和routingkey:

看起来似乎没什么问题,但不要高兴的太早,在最开始的时候,就介绍过,如果使用在消息属性上设置TTL的方式,消息可能并不会按时“死亡“,因为RabbitMQ只会检查第一个消息是否过期,如果过期则丢到死信队列,索引如果第一个消息的延时时长很长,而第二个消息的延时时长很短,则第二个消息并不会优先得到执行。

利用RabbitMQ插件实现延迟队列

上文中提到的问题,确实是一个硬伤,如果不能实现在消息粒度上添加TTL,并使其在设置的TTL时间及时死亡,就无法设计成一个通用的延时队列。

那如何解决这个问题呢?不要慌,安装一个插件即可:https://www.rabbitmq.com/community-plugins.html

安装

下载后解压,并将其拷贝至(使用Linux Debian/RPM部署)rabbitmq服务器目录:/usr/local/rabbitmq/plugins中( windows安装目录\rabbitmq_server-version\plugins ).

启用插件

使用命令

启用插件
输出如下:

通过rabbitmq-plugins list查看已安装列表,如下:

机制解释

安装插件后会生成新的Exchange类型x-delayed-message,该类型消息支持延迟投递机制,接收到消息后并未立即将消息投递至目标队列中,而是存储在mnesia(一个分布式数据系统)表中,检测消息延迟时间,如达到可投递时间时并将其通过x-delayed-type类型标记的交换机类型投递至目标队列。

生产者

消费者

只需改Exchange即可,无需修改其他

参考

【RabbitMQ】一文带你搞定RabbitMQ延迟队列
RabbitMQ 入门教程(PHP版) 使用rabbitmq-delayed-message-exchange插件实现延迟功能
ghigt/rabbitmq-delayed
基于rabbitmq实现的延时队列(golang版)


文章 延时队列 转载需要注明出处
喜欢 (0)

您必须 登录 才能发表评论!