Java实现一个延时队列

前言

业务中经常会出现各种涉及到定时，延迟执行的需求任务。

有一种队列专门处理这种情况。那就是延时队列。

本文提供两种实现方式：

1. java原生的内存型延时队列；
2. redisson 的内置延时队列；

正文

一、基本概念

延时队列（Delay Queue）是一种特殊的消息队列，用于处理需要在将来某个时间点执行的任务。

与普通的队列不同，延时队列中的消息在指定的时间之前是不可见的，只有当消息的延时时间到达后，消息才会被消费。

1.1 延时队列的特点

• 延时性：消息在进入队列后并不会立即被消费，而是需要等待一段时间后才能被消费。
• 有序性：消息按照延时时间的先后顺序被消费。
• 可靠性：通常需要保证消息不丢失，即使在系统故障的情况下也能恢复。（这里对于内存型的延时队列不太适合，一旦内存释放就会丢失消息）

1.2 常见的实现方式

• 数据库：使用数据库的定时任务或触发器。
• 消息队列：使用支持延时消息的消息队列，如 RabbitMQ、Kafka、RocketMQ 等。
• 内存队列：使用内存中的数据结构，如 Java 中的 DelayQueue。
• Redis：使用 Redis 的 sorted set 或 Redisson 的 RDelayedQueue。

二、Java原生的内存型延时队列

使用 Java 的 DelayQueue

• 生产者：将任务封装成 Delayed 接口的实现类，添加到 DelayQueue 中。
• 消费者：使用 take 或 poll 方法从 DelayQueue 中取出任务进行处理。

2.1 定义延时元素DelayedElement

package com.pine.common.util.delayqueue;import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;/*** 延迟元素** @author fengjinsong*/
public class DelayedElement implements Delayed {/*** 延迟时间（单位：毫秒）*/private final AtomicLong delayTime;/*** 到期时间*/private final AtomicLong expire;/*** 任务数据*/private final Object data;/*** 执行次数*/private final AtomicInteger executionFrequency;public DelayedElement(long delayTime, Object data) {this.delayTime = new AtomicLong(delayTime);this.expire = new AtomicLong(System.currentTimeMillis() + delayTime);this.data = data;this.executionFrequency = new AtomicInteger(0);}public Object getData() {return this.data;}public AtomicInteger getExecutionFrequency() {return executionFrequency;}public void setExecutionFrequency() {this.executionFrequency.incrementAndGet();}@Overridepublic long getDelay(TimeUnit unit) {return unit.convert(this.expire.longValue() - System.currentTimeMillis(), TimeUnit.MILLISECONDS);}@Overridepublic int compareTo(Delayed o) {return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));}/*** 重置延迟时间*/public void resetDelay(long delayTime) {this.delayTime.set(delayTime);this.expire.set(System.currentTimeMillis() + this.delayTime.longValue());}/*** 重置延迟时间*/public void resetDelay() {resetDelay(this.delayTime.longValue());}@Overridepublic String toString() {return "DelayedElement{" +"delayTime=" + delayTime +", expire=" + expire +", data=" + data +", executionFrequency=" + executionFrequency +'}';}
}

2.2 定义延时队列管理器DelayedQueueManager

package com.pine.common.util.delayqueue;import java.util.List;
import java.util.concurrent.DelayQueue;/*** 延时队列管理器** @author fengjinsong*/
public class DelayedQueueManager {private DelayedQueueManager() {}/*** 延时队列*/private static final DelayQueue<DelayedElement> DELAY_QUEUE = new DelayQueue<>();/*** 添加元素** @param element 元素*/public static void addElement(DelayedElement element) {DELAY_QUEUE.add(element);}public static void addElement(List<DelayedElement> elements) {DELAY_QUEUE.addAll(elements);}/*** 获取元素，并从队列中移除该元素** @return 元素*/public static DelayedElement pollElement() {return DELAY_QUEUE.poll();}
}

2.3 消费元素

package com.pine.common.util.delayqueue;import java.time.LocalDateTime;public class DelayedElementConsumer implements Runnable {private final static int[] FREQUENCY_SEQUENCE = new int[]{1, 2, 3, 6, 12, 24, 48, 96, 192, 384, 768};@Overridepublic void run() {boolean hasDelayedElement = true;while (hasDelayedElement) {// 获取元素DelayedElement element = DelayedQueueManager.pollElement();try {if (element != null) {System.out.println(LocalDateTime.now() + "消费了延迟元素：" + element);if (element.getData().toString().contains("3")) {throw new RuntimeException("模拟报错");}} else {hasDelayedElement = false;}} catch (Exception e) {retry(element);}}}private void retry(DelayedElement element) {element.setExecutionFrequency();System.out.println("执行出错：" + element);//出错3次后，不再重试if (element.getExecutionFrequency().intValue() > 3) {System.out.println("出错3次后，不再重试");} else {element.resetDelay(FREQUENCY_SEQUENCE[element.getExecutionFrequency().intValue() + 3] * 1000);// 重试DelayedQueueManager.addElement(element);}}}

2.4 调试

package com.pine.common.redis.delayqueue;import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;public class Client {public static void main(String[] args) {// 模拟生产数据RedissonDelayedQueueManager.offer("hello22", 3000);RedissonDelayedQueueManager.offer("hello33", 5000);// 模拟消费数据System.out.println(LocalDateTime.now() + "开始消费数据");while (true) {Object object = RedissonDelayedQueueManager.poll(10, TimeUnit.SECONDS);if (object != null) {System.out.println("-----------------------" + LocalDateTime.now() + ":" + object);}}}
}

2.5 调试结果

2024-11-06T16:57:39.342358开始消费数据
-----------------------2024-11-06T16:57:42.285383:hello22
-----------------------2024-11-06T16:57:44.378298:hello33

可以观察到 hello22 延时了3秒；hello33延时了5秒；

2.6 精髓之 DelayQueue.poll()

检索并删除此队列的头部，如果此队列没有延迟过期的元素，则返回null。

public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {E first = q.peek();return (first == null || first.getDelay(NANOSECONDS) > 0)? null: q.poll();} finally {lock.unlock();}}

三、基于Redisson的延时队列

使用 Redisson 的 RDelayedQueue

• 生产者：使用 RDelayedQueue 的 offer 方法将任务添加到队列中，指定延时时间。
• 消费者：使用 RQueue 的 poll 方法从队列中取出任务进行处理。

3.1 定义延时队列管理器

package com.pine.common.redis.delayqueue;import org.redisson.Redisson;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;import java.io.IOException;
import java.util.concurrent.TimeUnit;public class RedissonDelayedQueueManager {private static final String QUEUE_NAME = "delay_queue";private static final RedissonClient REDISSON_CLIENT;static {try {String content = """singleServerConfig:address: "redis://10.189.64.136:8379"""";Config config = Config.fromYAML(content);REDISSON_CLIENT = Redisson.create(config);} catch (IOException e) {throw new RuntimeException(e);}}/*** 获取延迟队列* <p>* 本方法通过Redisson客户端创建一个阻塞队列，并基于该阻塞队列创建一个延迟队列* 延迟队列用于处理需要延迟执行的任务，例如任务重试机制、任务调度等场景** @param <T> 队列中元素的类型* @return 返回一个延迟队列实例，用于后续的操作和管理*/private static <T> RDelayedQueue<T> getDelayedQueue() {// 创建一个阻塞队列，这是后续创建延迟队列的基础RBlockingQueue<T> queue = REDISSON_CLIENT.getBlockingQueue(QUEUE_NAME);// 基于阻塞队列创建延迟队列并返回return REDISSON_CLIENT.getDelayedQueue(queue);}/*** 向延迟队列中添加元素，并设置延迟时间** @param task     要添加的元素* @param delayTime 延迟时间，单位为毫秒* @param <T>      元素类型*/public static <T> void offer(T task, long delayTime) {RDelayedQueue<T> delayedQueue = getDelayedQueue();delayedQueue.offer(task, delayTime, TimeUnit.MILLISECONDS);}/*** 从延迟队列中获取元素，并设置超时时间** @param timeout 超时时间，单位为毫秒* @param unit    超时时间单位* @param <T>     元素类型* @return 返回获取到的元素，如果没有获取到元素则返回null*/public static <T> T poll(long timeout, TimeUnit unit) {RBlockingQueue<T> queue = REDISSON_CLIENT.getBlockingQueue(QUEUE_NAME);try {return queue.poll(timeout, unit);} catch (InterruptedException e) {throw new RuntimeException(e);}}
}

3.2 调试

package com.pine.common.redis.delayqueue;import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;public class Client {public static void main(String[] args) {// 模拟生产数据RedissonDelayedQueueManager.offer("hello22", 3000);RedissonDelayedQueueManager.offer("hello33", 5000);// 模拟消费数据System.out.println(LocalDateTime.now() + "开始消费数据");while (true) {Object object = RedissonDelayedQueueManager.poll(10, TimeUnit.SECONDS);if (object != null) {System.out.println("-----------------------" + LocalDateTime.now() + ":" + object);}}}
}

3.3 调试结果

2024-11-06T17:05:31.630768开始消费数据
-----------------------2024-11-06T17:05:34.548032:hello22
-----------------------2024-11-06T17:05:36.732607:hello33

可以观察到 hello22 延时了3秒；hello33延时了5秒；