SpringBoot 异步延时任务

1. 引言

异步延时任务在日常工作中是比较常见的。实现方案也比较多。

2. 延迟队列DelayQueue

延迟队列是 jdk1.5 提供的，核心是优先级队列PriorityQueue。

2.1 延时任务

DelayQueue要求队列中的元素必须实现Delayed接口。

import java.util.Date;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class DelayTask<T> implements Delayed {
    final private T data;
    final private long expire;

    /**
     * 构造延时任务
     * @param data      业务数据
     * @param expire    任务延时时间（ms）
     */
    public DelayTask(T data, long expire) {
        super();
        this.data = data;
        this.expire = expire + System.currentTimeMillis();
    }

    public T getData() {
        return data;
    }

    public long getExpire() {
        return expire;
    }


    @Override
    public String toString() {
        return "{" + "data:" + data.toString() + "," + "expire:" + new Date(expire) + "}";
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), unit);
    }

    @Override
    public int compareTo(Delayed o) {
        long delta = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
        return (int) delta;
    }
}

2.2 延时任务管理器

实现CommandLineRunner接口，重写了方法void run(String... args);，这个方法会在Spring容器启动后，所有Bean都已经被创建和初始化之后被调用。这个方法的参数是一个字符串数组，通常包含命令行参数，但是你可以根据需要忽略这个参数。
定义一个静态变量DelayQueue，用于保存延时任务。
在 run 方法中启用线程循环从队列中获取数据，获取不到就阻塞线程，直到有数据为止。
取到任务后，执行processTask方法出来业务逻辑。

@Component
@Slf4j
public class DelayQueueManager implements CommandLineRunner {

    private static DelayQueue<DelayTask> delayQueue = new DelayQueue<>();

    /**
     * 加入到延时队列中
     *
     * @param task
     */
    public void put(DelayTask task) {
        log.info("加入延时任务：{}", task);
        delayQueue.put(task);
    }

    /**
     * 取消延时任务
     *
     * @param task
     * @return
     */
    public boolean remove(DelayTask task) {
        log.info("取消延时任务：{}", task);
        return delayQueue.remove(task);
    }


    @Override
    public void run(String... args) throws Exception {
        log.info("初始化延时队列");
        Executors.newSingleThreadExecutor().execute(new Thread(this::excuteThread));
    }

    /**
     * 延时任务执行线程
     */
    private void excuteThread() {
        while (true) {
            try {
                DelayTask task = delayQueue.take();
                processTask(task);
            } catch (InterruptedException e) {
                break;
            }
        }
    }

    /**
     * 内部执行延时任务
     *
     * @param task
     */
    private void processTask(DelayTask task) {
        log.info("执行延时任务：{}", task);
        //根据task中的data自定义数据来处理相关逻辑，例 if (task.getData() instanceof XXX) {}
        // do something
    }
}