6种延时队列实现方法：从DelayQueue到Redis、RabbitMQ全解析

五一假期原本计划写两篇文章、看一本技术书，结果五天连电脑都没打开，自律性确实堪忧。不过最近项目招人，我有幸当了回面试官，就整理一道常见面试题来和大家聊聊：如何实现延时队列？

下面会介绍多种实现延时队列的思路，文末提供对应的 GitHub 地址。技术方案没有绝对的好坏，关键看业务场景是否合适。

一、延时队列的应用

什么是延时队列？简单说，它首先具备队列的特性，再附加一个“延迟消费”的功能——可以指定消息在哪个时间点被消费。

电商平台里的应用场景不少：

1、订单成功后 30 分钟内未支付，自动取消订单

2、外卖平台下单成功 60s 后给用户推送短信

3、订单长时间未完结时，及时关单并退还库存

4、淘宝新建商户一个月内未上传商品信息，冻结商铺

这些场景都可以用延时队列来解决。

二、延时队列的实现

我个人一直坚持：能用 JDK 自带 API 实现的功能，就别自己重复造轮子，也别轻易引入三方中间件。自己封装容易出问题，还要花时间调试验证；而三方中间件会让系统复杂度成倍增加，维护成本也高。

1、DelayQueue 延时队列

JDK 在 java.util.concurrent 包下提供了 DelayQueue，这是一组实现延迟队列的 API。

DelayQueue 是一个 BlockingQueue（无界阻塞）队列，本质封装了 PriorityQueue（优先队列），PriorityQueue 内部通过完全二叉堆实现元素排序。入队时需要给元素一个 Delay（延迟时间）作为排序条件，队列中最小的元素优先放在队首，且只有到了 Delay 时间才允许从队列中取出。队列中可以放基本数据类型或自定义实体类——基本类型默认升序排列，自定义实体类需要根据类属性值比较计算。

先简单实现看看效果：添加三个 order 入队 DelayQueue，分别设置订单在当前时间的 5 秒、10 秒、15 秒后取消。

实现 DelayQueue 延时队列，队中元素要实现 Delayed 接口，这个接口只有一个 getDelay 方法用于设置延期时间。Order 类中 compareTo 方法负责对队列中的元素排序。

public class Order implements Delayed {
    /**
     * 延迟时间
     */
    @JsonFormat(locale = "zh", timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
    private long time;
    String name;

    public Order(String name, long time, TimeUnit unit) {
        this.name = name;
        this.time = System.currentTimeMillis() + (time > 0 ? unit.toMillis(time) : 0);
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return time - System.currentTimeMillis();
    }

    @Override
    public int compareTo(Delayed o) {
        Order order = (Order) o;
        long diff = this.time - order.time;
        if (diff <= 0) {
            return -1;
        } else {
            return 1;
        }
    }
}

DelayQueue 的 put 方法是线程安全的，因为内部使用了 ReentrantLock 锁进行线程同步。

DelayQueue 还提供了两种出队方法：poll() 和 take()。poll() 非阻塞获取，没有到期的元素直接返回 null；take() 阻塞获取，没有到期的元素线程会等待。

public class DelayQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        Order order1 = new Order("Order1", 5, TimeUnit.SECONDS);
        Order order2 = new Order("Order2", 10, TimeUnit.SECONDS);
        Order order3 = new Order("Order3", 15, TimeUnit.SECONDS);
        DelayQueue<Order> delayQueue = new DelayQueue<>();
        delayQueue.put(order1);
        delayQueue.put(order2);
        delayQueue.put(order3);

        System.out.println("订单延迟队列开始时间:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        while (delayQueue.size() != 0) {
            // 取队列头部元素是否过期
            Order task = delayQueue.poll();
            if (task != null) {
                System.out.format("订单:{%s}被取消, 取消时间:{%s}\n", task.name, LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            }
            Thread.sleep(1000);
        }
    }
}

执行结果如下，Order1、Order2、Order3 分别在 5 秒、10 秒、15 秒后被执行。实际开发中会有专门的线程负责消息的入队与消费，上面只是简单演示。

订单延迟队列开始时间:2020-05-06 14:59:09
订单:{Order1}被取消, 取消时间:{2020-05-06 14:59:14}
订单:{Order2}被取消, 取消时间:{2020-05-06 14:59:19}
订单:{Order3}被取消, 取消时间:{2020-05-06 14:59:24}

2、Quartz 定时任务

Quartz 是一款非常经典的任务调度框架。在 Redis、RabbitMQ 还未广泛应用时，超时未支付取消订单功能大多由定时任务实现。不过定时任务有固定周期，很多单子可能已经超时，但还没到触发时间点，处理就不够及时。这里简单介绍一下它的定时机制。

引入 quartz 依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>

启动类使用 @EnableScheduling 开启定时任务功能：

@EnableScheduling
@SpringBootApplication
public class DelayqueueApplication {
    public static void main(String[] args) {
        SpringApplication.run(DelayqueueApplication.class, args);
    }
}

编写一个每 5 秒执行一次的定时任务：

@Component
public class QuartzDemo {

    // 每隔五秒
    @Scheduled(cron = "0/5 * * * * ? ")
    public void process() {
        System.out.println("我是定时任务！");
    }
}

3、Redis sorted set

Redis 的数据结构 Zset 同样可以实现延迟队列，主要利用它的 score 属性——Redis 通过 score 为集合中的成员从小到大排序。

通过 zadd 命令向队列 delayqueue 添加元素，score 值表示元素过期时间。向 delayqueue 添加 order1、order2、order3，分别是 10 秒、20 秒、30 秒后过期。

zadd delayqueue 3 order3

消费端轮询队列 delayqueue，将元素排序后取最小时间与当前时间比对，如小于当前时间则移除 key。

// 消费消息
public void pollOrderQueue() {

    while (true) {
        Set<Tuple> set = jedis.zrangeWithScores(DELAY_QUEUE, 0, 0);

        String value = ((Tuple) set.toArray()[0]).getElement();
        int score = (int) ((Tuple) set.toArray()[0]).getScore();

        Calendar cal = Calendar.getInstance();
        int nowSecond = (int) (cal.getTimeInMillis() / 1000);
        if (nowSecond >= score) {
            jedis.zrem(DELAY_QUEUE, value);
            System.out.println(sdf.format(new Date()) + " removed key:" + value);
        }

        if (jedis.zcard(DELAY_QUEUE) <= 0) {
            System.out.println(sdf.format(new Date()) + " zset empty ");
            return;
        }
        Thread.sleep(1000);
    }
}

执行结果符合预期：

2020-05-07 13:24:09 add finished.
2020-05-07 13:24:19 removed key:order1
2020-05-07 13:24:29 removed key:order2
2020-05-07 13:24:39 removed key:order3
2020-05-07 13:24:39 zset empty 

4、Redis 过期回调

Redis 的 key 过期回调事件也能达到延迟队列的效果。简单来说，开启监听 key 是否过期的事件，一旦 key 过期就会触发一个 callback 事件。

修改 redis.conf 开启 notify-keyspace-events Ex：

notify-keyspace-events Ex

Redis 监听配置，注入 Bean RedisMessageListenerContainer：

@Configuration
public class RedisListenerConfig {
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }
}

编写 Redis 过期回调监听方法，必须继承 KeyExpirationEventMessageListener，类似于 MQ 的消息监听：

@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {

    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String expiredKey = message.toString();
        System.out.println("监听到key：" + expiredKey + "已过期");
    }
}

到这里代码就完成了，非常简单。在 redis-cli 中添加一个 key 并设置 3s 过期时间：

set xiaofu 123 ex 3

控制台成功监听到了这个过期的 key：

监听到过期的key为：xiaofu

5、RabbitMQ 延时队列

利用 RabbitMQ 做延时队列是比较常见的方式。不过 RabbitMQ 本身并没有直接支持延迟队列功能，而是通过 TTL 和 DLX 两个属性间接实现的。

先认识两个概念：

Time To Live（TTL）：

TTL 指消息的存活时间。RabbitMQ 可以通过 x-message-tt 参数设置 Queue（队列）和 Message（消息）上消息的存活时间，值为非负整数，单位微秒。

RabbitMQ 可以从两个维度设置消息过期时间：

• 设置队列过期时间，队列中所有消息都具有相同过期时间
• 设置消息过期时间，每条消息 TTL 都可以不同

如果同时设置队列和消息的 TTL，取两者中较小的值为准。队列中的消息一旦超过 TTL 过期时间就成为 Dead Letter（死信）。

Dead Letter Exchanges（DLX）：

DLX 即死信交换机，绑定在死信交换机上的即死信队列。RabbitMQ 的 Queue 可以配置 x-dead-letter-exchange 和 x-dead-letter-routing-key（可选）两个参数，一旦队列内出现 Dead Letter，就按这两个参数将消息重新路由到另一个 Exchange，让消息重新被消费。

• x-dead-letter-exchange：队列中出现 Dead Letter 后，将 Dead Letter 重新路由转发到指定 exchange
• x-dead-letter-routing-key：指定 routing-key 发送，一般为要指定转发的队列

队列出现 Dead Letter 的情况：

• 消息或队列的 TTL 过期
• 队列达到最大长度
• 消息被消费端拒绝（basic.reject 或 basic.nack）

下图展示了如何实现超 30 分钟未支付关单功能：将订单消息 A0001 发送到延迟队列 order.delay.queue，设置 x-message-tt 消息存活时间为 30 分钟；到期后 A0001 成为 Dead Letter，延迟队列检测到死信后，通过 x-dead-letter-exchange 将死信转发到能正常消费的关单队列，监听关单队列处理关单逻辑即可。

发送消息时指定延迟时间：

public void send(String delayTimes) {
    amqpTemplate.convertAndSend("order.pay.exchange", "order.pay.queue", "大家好我是延迟数据", message -> {
        // 设置延迟毫秒值
        message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
        return message;
    });
}

设置延迟队列出现死信后的转发规则：

/**
 * 延时队列
 */
@Bean(name = "order.delay.queue")
public Queue getMessageQueue() {
    return QueueBuilder
            .durable(RabbitConstant.DEAD_LETTER_QUEUE)
            // 配置到期后转发的交换
            .withArgument("x-dead-letter-exchange", "order.close.exchange")
            // 配置到期后转发的路由键
            .withArgument("x-dead-letter-routing-key", "order.close.queue")
            .build();
}

6、时间轮

前面几种实现方式相对简单，时间轮算法稍微抽象一些。Kafka、Netty 都有基于时间轮算法实现延时队列，下面以 Netty 的延时队列为例讲一下原理。

先看时间轮原理图，解读几个基本概念：

• wheel：时间轮，圆盘可以看作钟表的刻度。比如一圈 round 长度为 24 秒，刻度数为 8，那么每个刻度表示 3 秒，时间精度就是 3 秒。时间长度 / 刻度数值越大，精度越高。
• 当添加一个延时任务 A，假如延迟 25 秒后执行，可时间轮一圈 round 长度才 24 秒，此时会根据时间轮长度和刻度得到一个圈数 round 和对应的指针位置 index，也就是任务 A 会绕一圈指向 0 格子上，此时时间轮记录该任务的 round 和 index 信息。当 round=0，index=0，指针指向 0 格子时，任务 A 并不会执行，因为 round=0 不满足要求。
• 所以每个格子代表一段时间，比如 1 秒和 25 秒的任务都会指向 0 格子上，而任务则放在每个格子对应的链表中，这点和 HashMap 的数据结构有些类似。

Netty 构建延时队列主要用 HashedWheelTimer，底层数据结构依然使用 DelayedQueue，只是采用时间轮算法实现。

用 Netty 简单实现延时队列，先解释 HashedWheelTimer 构造函数参数：

• ThreadFactory：表示用于生成工作线程，一般采用线程池
• tickDuration 和 unit：每格的时间间隔，默认 100ms
• ticksPerWheel：一圈下来有几格，默认 512。如果传入值不是 2 的 N 次方，会调整为大于等于该参数的一个 2 的 N 次方数值，有利于优化 hash 值的计算

public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) {
    this(threadFactory, tickDuration, unit, ticksPerWheel, true);
}

• TimerTask：定时任务的实现接口，run 方法包装了定时任务的逻辑
• Timeout：定时任务提交到 Timer 后返回的句柄，通过它可以取消定时任务，并对任务状态进行基本判断
• Timer：HashedWheelTimer 实现的父接口，仅定义了如何提交定时任务和如何停止整个定时机制

public class NettyDelayQueue {

    public static void main(String[] args) {

        final Timer timer = new HashedWheelTimer(Executors.defaultThreadFactory(), 5, TimeUnit.SECONDS, 2);

        // 定时任务
        TimerTask task1 = new TimerTask() {
            public void run(Timeout timeout) throws Exception {
                System.out.println("order1  5s 后执行 ");
                timer.newTimeout(this, 5, TimeUnit.SECONDS); // 结束时再次注册
            }
        };
        timer.newTimeout(task1, 5, TimeUnit.SECONDS);
        TimerTask task2 = new TimerTask() {
            public void run(Timeout timeout) throws Exception {
                System.out.println("order2  10s 后执行");
                timer.newTimeout(this, 10, TimeUnit.SECONDS); // 结束时再注册
            }
        };

        timer.newTimeout(task2, 10, TimeUnit.SECONDS);

        // 延迟任务
        timer.newTimeout(new TimerTask() {
            public void run(Timeout timeout) throws Exception {
                System.out.println("order3  15s 后执行一次");
            }
        }, 15, TimeUnit.SECONDS);

    }
}

执行结果：order3 延迟任务只执行一次，order2、order1 为定时任务，按不同周期重复执行。

order1  5s 后执行 
order2  10s 后执行
order3  15s 后执行一次
order1  5s 后执行 
order2  10s 后执行

总结

为了便于理解，上面代码写得都比较简单。几种实现方式的 demo 已经全部提交到 GitHub：https://github.com/chengxy-nds/delayqueue ，感兴趣的朋友可以下载跑一跑。

这篇文章整理确实费了不少功夫，查证资料、反复验证 demo 可行性、搭建 RabbitMQ、Redis 环境。有写得不够完善的地方，欢迎大家指正。如果你对延时队列或 Java 并发实战还有疑问，欢迎到云栈社区和更多开发者一起交流。