点击上方“Java精选”，选择“设为星标”

别问别人为什么，多问自己凭什么！

下方有惊喜，留言必回，有问必答！

每天 08:15 更新文章，每天进步一点点...

一、背景

在业务发展过程中，会出现一些需要延时处理的场景，比如：

1. 订单下单之后超过30分钟用户未支付，需要取消订单

2. 订单一些评论，如果48h用户未对商家评论，系统会自动产生一条默认评论

3. 点我达订单下单后，超过一定时间订单未派出，需要超时取消订单等。。

处理这类需求，比较直接简单的方式就是定时任务轮训扫表。这种处理方式在数据量不大的场景下是完全没问题，但是当数据量大的时候高频的轮训数据库就会比较的耗资源，导致数据库的慢查或者查询超时。所以在处理这类需求时候，采用了延时队列来完成。

二、几种延时队列

延时队列就是一种带有延迟功能的消息队列。下面会介绍几种目前已有的延时队列：

推荐下自己几个月熬夜整理的各个大厂面试资料：

https://gitee.com/yoodb/eboo‍ks

1.Java中java.util.concurrent.DelayQueue

优点：JDK自身实现，使用方便，量小适用

缺点：队列消息处于jvm内存，不支持分布式运行和消息持久化

2.Rocketmq延时队列

优点：消息持久化，分布式

缺点：不支持任意时间精度，只支持特定level的延时消息

3.Rabbitmq延时队列（TTL+DLX实现）

优点：消息持久化，分布式

缺点：延时相同的消息必须扔在同一个队列

根据自身业务和公司情况，如果实现一个自己的延时队列服务需要考虑一下几点:

* 消息存储
* 过期延时消息实时获取
* 高可用性

三、 基于Redis实现

1.0版本

• 功能特性

* 消息可靠性，消息持久化，消息至少被消费一次

* 实时性：存在一定的时间误差（定时任务间隔）

* 支持指定消息remove

* 高可用性

• 整体结构

- Messages Pool所有的延时消息存放，结构为KV结构，key为消息ID，value为一个具体的message（这里选择Redis Hash结构主要是因为hash结构能存储较大的数据量，数据较多时候会进行渐进式rehash扩容，并且对于HSET和HGET命令来说时间复杂度都是O(1)）

- Delayed Queue是16个有序队列（队列支持水平扩展），结构为ZSET，value为messages pool中消息ID，score为过期时间（分为多个队列是为了提高扫描的速度）

推荐下自己做的 Spring boot 的实战项目：
https://gitee.com/yoodb/jing-xuan

- Timed Task定时任务，负责扫描处理每个队列过期消息

• 消息结构

每个延时消息必须包括以下参数:

* tags：消息过期之后发送mq的tags

* keys：消息过期之后发送mq的keys

* body：消息过期之后发送mq的body，提供给消费这做具体的消息处理

* delayTime：延时发送时间（默认，delayTime、expectDate有一个即可）

* expectDate：期望发送时间

* 另外，推荐公众号Java精选，回复java面试，获取最新redis面试题资料，支持在线随时随地刷题。

• 流程

注：上图1、2、3或者2、3是一个事务操作

取出过期消息过程是通过一个外部定时任务每隔1min分钟去查询队列中过期的消息，然后发送mq && remove

2.0版本

1.0上有一个可改进的地方就是队列中过期的消息是通过定时任务触发查询。所有有了2.0

2.0版本在1.0上做了一个优化，废弃掉了1min定时任务触发过期消息发送，采用了java Lock await/singlal方式实现过期消息的实时发送低延时

• 多节点部署结构：

- pull job：这里分别为每一个队列创建了一个pull job thread，功能很简单，就是负责去队列中拉取过期的消息数据（这里保证一个队列有且只有一个pull job）

- worker：pull job拉取到的过期消息会交给一个worker thread去处理，这样的好处是处理过期的消息实时性更高（pull job不必等去除过期消息全部处理完成在继续去拉取新的过期数据）。另外，关注公众号Java精选，回复渗透工具四个汉字，可以获取十种渗透工具。

- zookeeper coordinate：通过zk的操作来完成对队列的重新分配工作，daemon thread监听zk节点的创建和删除。

• 主要流程：

服务启动会注册zk，获取分配处理的queues，启动后台线程监听zk 。
为每个分配queue创建一个pull job 。

pull job首先会去queue中查询是否有过期消息：

Y：将取出消息交给worker处理

N：查询queue中最后一个成员（zset结构默认按score递增排序），如果为空，则await；不为空则await（成员score-System.currentTimeMillis()）

由于过期消息发送成功才会从队列中remove，所以pull job会记录上一次查询队列的一个offset，每次获取到过期消息会将offset向前偏移，过期消息交给worker处理，当worker由于某些异常原因处理失败会重置pull job中offset，这样可以避免消息发送一次失败之后没办法在继续处理（除了新节点add || remove时候）。

当部署服务有新增，延时队列服务会重新计算得到当前处理队列，并将之前创建pull job cancel，为新处理队列重新创建pull job。删除同理。

作者：程序人生ly

https://www.cnblogs.com/lylife/p/7881950.html

公众号“Java精选”所发表内容注明来源的，版权归原出处所有（无法查证版权的或者未注明出处的均来自网络，系转载，转载的目的在于传递更多信息，版权属于原作者。如有侵权，请联系，笔者会第一时间删除处理！

------ THE END ------

精品资料，超赞福利！

>Java精选面试题<
3000+ 道面试题在线刷，最新、最全 Java 面试题！

期往精选  点击标题可跳转

【232期】面试官：如何保护 Spring Boot 配置文件敏感信息？

【233期】Java8 stream 处理 List 集合的相同部分(交集)、去重！

【234期】新来的同事问我 where 1=1 是什么意思？

【235期】不同并发场景下 LongAdder 与 AtomicLong 如何选择？

【236期】ElasticSearch 进阶：一文全览各种 ES 查询在 Java 中的实现

【237期】Java 8 判空新写法

【238期】Java 8 中 Lambda 实现原理及源码剖析！

【239期】面试官问：你觉得 ThreadLocalRandom 这玩意安全吗？

技术交流群！

最近有很多人问，有没有读者&异性交流群，你懂的！想知道如何加入。加入方式很简单，有兴趣的同学，只需要点击下方卡片，回复“加群”，即可免费加入交流群！

文章有帮助的话，在看，转发吧！