ThreadPoolTaskScheduler 周期任务原理

Posted 2021-09-28 lxKLj

ThreadPoolTaskScheduler 核心就是schedule 方法

public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
    ScheduledExecutorService executor = getScheduledExecutor();
    try {
        ErrorHandler errorHandler = this.errorHandler;
        if (errorHandler == null) {
            errorHandler = TaskUtils.getDefaultErrorHandler(true);
        }
      // 最终调用ReschedulingRunnable.schedule 方法
        return new ReschedulingRunnable(task, trigger, executor, errorHandler).schedule();
    }
    catch (RejectedExecutionException ex) {
        throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
    }
}

后续进入ReschedulingRunnable.schedule 方法，该类中executor 属性为ScheduledThreadPoolExecutor 类，属性为ScheduledThreadPoolExecutor 类继承了ThreadPoolExecutor 线程池，但是自定义了DelayedWorkQueue 延迟队列，而不是使用ThreadPoolExecutor 类自带的队列，周期任务延迟执行的根本原因就是DelayedWorkQueue 这个延迟队列。

public ScheduledFuture<?> schedule() {
    synchronized (this.triggerContextMonitor) {
        this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);
        if (this.scheduledExecutionTime == null) {
            return null;
        }
        long initialDelay = this.scheduledExecutionTime.getTime() - System.currentTimeMillis();
        this.currentFuture = this.executor.schedule(this, initialDelay, TimeUnit.MILLISECONDS);
        return this;
    }
}

DelayedWorkQueue 类通过一个最小堆来存储ThreadPoolTaskScheduler 中的任务，各任务会进行比较，最快要执行的任务放在最小堆顶部。放入最小堆，通过siftUp，取出最小堆，通过siftDown。

/**
 * 上浮
 */
private void siftUp(int k, RunnableScheduledFuture<?> key) {
    // 一直遍历到根节点下方
    while (k > 0) {
        // 二叉堆，最高节点坐标为0
        // 父节点，(k - 1)/2
        int parent = (k - 1) >>> 1;
        RunnableScheduledFuture<?> e = queue[parent];
        // 目标比父节点大
        // 不需要再上浮，直接跳出
        if (key.compareTo(e) >= 0)
            break;
        // 目标节点比父节点小
        // 继续上浮，当前坐标填入父节点
        queue[k] = e;
        setIndex(e, k);
        // 当前坐标设为父节点坐标
        k = parent;
    }
    // 目标上浮到最小节点坐标，填入该坐标
    queue[k] = key;
    setIndex(key, k);
}

/**
 * 下沉
 */
private void siftDown(int k, RunnableScheduledFuture<?> key) {
    // half = size/2;
    // 二叉堆，最高节点坐标为0
    // 任何节点，其左子节点坐标(k*2)+1，右子节点坐标(k*2)+2
    int half = size >>> 1;
    // 需要拿到子节点，所以只需要到size/2 即可，不需要到最底层
    while (k < half) {
        // 左节点坐标
        int child = (k << 1) + 1;
        RunnableScheduledFuture<?> c = queue[child];
        // 右节点坐标
        int right = child + 1;
        // 左节点比右节点大
        if (right < size && c.compareTo(queue[right]) > 0)
            // 最小 = 右子节点
            c = queue[child = right];
        // 目标最小比子节点小
        // 不再需要下沉，直接退出，目标填入当前坐标
        if (key.compareTo(c) <= 0)
            break;
        // 目标比子节点大
        // 继续下沉，小子节点放到当前节点坐标
        queue[k] = c;
        setIndex(c, k);
        // 当前坐标设为子节点坐标
        // 坐标不断下沉
        k = child;
    }
    // 当前坐标为目标对象最小情况下的坐标
    // 讲目标对象放入该坐标
    queue[k] = key;
    setIndex(key, k);
}

当从DelayedWorkQueue 队列中取出任务时，会取出最小堆顶部的任务，也就是最快要执行的任务，然后线程等待指定时间。等待时间结束后，通过自旋完成任务。

public RunnableScheduledFuture<?> take() throws InterruptedException {
    ...
    for (;;) {
        RunnableScheduledFuture<?> first = queue[0];
        ...
        if (delay <= 0)
            return finishPoll(first);
        ...
        available.awaitNanos(delay);
        ...
    }
   ...
}