【Java多线程】ScheduledThreadPoolExecutor实现原理（二十九）

　　ScheduledThreadPoolExecutor，它是一个计划任务线程池，可以执行定时任务或者是计划任务。

　　ScheduledThreadPoolExecutor 继承了ThreadPoolExecutor，需要了解 ThreadPoolExecutor 的原理，参考：【Java多线程】线程池ThreadPoolExecutor实现原理（二十二） 

　　ScheduledThreadPoolExecutor 中的任务队列使用了 阻塞式延迟队列 （DelayedWorkQueue），参考：【Java多线程】DelayQueue源码分析 （二十六） 

　　了解以上2个知识点之后，能更好的理解ScheduledThreadPoolExecutor的原理

　　参考：【Java多线程】ScheduledThreadPoolExecutor详解（二十八） 

　　本章是对其进行补充

一、ScheduledThreadPoolExecutor原理图

　　

　　任务传递示意图

　　

二、属性

　　由于 ScheduledThreadPoolExecutor 是继承了 ThreadPoolExecutor，ThreadPoolExecutor有的属性它都有，以下是另外增加的属性

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

    // shutdown后继续存在周期任务
    private volatile boolean continueExistingPeriodicTasksAfterShutdown;

    // shutdown后执行存在延迟任务
    private volatile boolean executeExistingDelayedTasksAfterShutdown = true;

    // 取消状态下移除标识
    private volatile boolean removeOnCancel = false;

    // 序列器
    private static final AtomicLong sequencer = new AtomicLong();

    ......
}

三、方法

　　在ScheduledThreadPoolExecutor线程池提交任务执行，主要使用的ScheduledThreadPoolExecutor.java、ThreadPoolExecutor.java、ScheduledFutureTask.java、FutureTask.java、DelayedWorkQueue.java等相关类方法，如下：

1、构造方法

// 根据核心线程数创建 ScheduledThreadPool 线程池
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

// 根据核心线程数、线程工厂创建 ScheduledThreadPool 线程池
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory);
}

// 根据核心线程数、拒绝策略创建 ScheduledThreadPool
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), handler);
}

// 根据核心线程数、线程工厂、拒绝策略创建 ScheduledThreadPool 线程池
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory, handler);
}

2、schedule() 方法

public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                       long delay,
                                       TimeUnit unit) {
    if (callable == null || unit == null)
        throw new NullPointerException();
    // 新建一个 RunnableScheduledFuture 对象
    RunnableScheduledFuture<V> t = decorateTask(callable,
        new ScheduledFutureTask<V>(callable,
                                   triggerTime(delay, unit)));

    // 延迟执行任务
    delayedExecute(t);
    return t;
}

　　由上可以看出，主要是新建了一个ScheduledFutureTask对象，然后点用延迟执行任务方法 delayedExecute(t)

3、decorateTask() 方法

// 包装任务成一个 RunnableScheduledFuture 对象
// 实际就是返回了task对象
protected <V> RunnableScheduledFuture<V> decorateTask(
    Callable<V> callable, RunnableScheduledFuture<V> task) {
    return task;
}

　　返回任务对象

4、delayedExecute() 方法

private void delayedExecute(RunnableScheduledFuture<?> task) {
    // 线程池状态是否 Shutdown
    if (isShutdown())
        // 拒绝任务
        reject(task);
    // 
    else {
        // 获取任务队列，添加任务
        super.getQueue().add(task);
        // isShutdown() 再次判断任务状态
        // canRunInCurrentRunState() 是否在当前状态下能运行
        // remove() 移除任务
        if (isShutdown() &&
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            // 任务取消
            task.cancel(false);
        else
            // 确保线程池能启动任务
            ensurePrestart();
    }
}

　　主要是将任务添加到队列中，然后执行ensurePrestart()确保线程池能启动任务

5、ThreadPoolExecutor 类中的 ensurePrestart() 方法

void ensurePrestart() {
    // 获取工作线程数量
    int wc = workerCountOf(ctl.get());
    // 工作线程数小于 核心线程数
    if (wc < corePoolSize)
        // 添加工作线程线，以核心线程池数为上限
        addWorker(null, true);
    else if (wc == 0)
        // 添加工作线程线，以最大线程池线程数为上限
        addWorker(null, false);
}

6、ThreadPoolExecutor 类中的 reject() 方法 拒绝任务

// 拒绝任务
final void reject(Runnable command) {
    // 调用创建线程池时，使用的拒绝策略对象处理
    handler.rejectedExecution(command, this);
}

7、ThreadPoolExecutor 类中的 remove() 方法 移除任务

public boolean remove(Runnable task) {
    // 从任务队列中
    boolean removed = workQueue.remove(task);
    // 尝试终止线程池
    tryTerminate(); // In case SHUTDOWN and now empty
    return removed;
}

8、ScheduledFutureTask 类中的 run() 方法 

// ScheduledFutureTask 的 run() 方法
public void run() {
    // 是否是周期任务
    boolean periodic = isPeriodic();
    // canRunInCurrentRunState() 是否在当前状态下能运行
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    // 非周期任务
    else if (!periodic)
        // 直接运行任务的run()方法
        ScheduledFutureTask.super.run();
    // FutureTask 类中的 runAndReset()方法
    // 运行且重置
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 设置
        setNextRunTime();
        reExecutePeriodic(outerTask);
    }
}

9、ScheduledFutureTask 类中的 setNextRuntime() 方法 

// 设置下次运行时间
private void setNextRunTime() {
    // 周期时间
    long p = period;

    if (p > 0)
        // time 任务到期时间
        // 任务完成后，在任务到期时间上 + 一个周期时间，进行延迟。
        time += p;
    else
        // 现在的时间 + 一个周期时间，进行延迟
        time = triggerTime(-p);
}

　　重新设置任务的执行时间

10、ScheduledFutureTask 类中的reExecutePeriodic() 方法 

// 重新设置周期任务
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    // canRunInCurrentRunState() 是否在当前状态下能运行
    if (canRunInCurrentRunState(true)) {
        // 将周期任务加入队列
        super.getQueue().add(task);
        // 再次判断是否在当前状态下能运行
        // 不能就移除任务，并取消
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            // 确保线程池能启动任务
            ensurePrestart();
    }
}

11、ScheduledFutureTask 类中的 compareTo() 方法 任务比较优先级

　　在放入任务到队列或取出时，用到比较方法

12、ScheduledFutureTask 类中的getDelay() 方法  获取延迟时间

// 返回当前元素还需要延迟多长时间，单位是纳秒
public long getDelay(TimeUnit unit) {
    return unit.convert(time - now(), NANOSECONDS);
}

13、ScheduledFutureTask 类中的 cancel() 方法

// ScheduledFutureTask类中的 cancel() 方法
// 参数 mayInterruptIfRunning ：如果任务运行是否可以被中断
// 取消任务
public boolean cancel(boolean mayInterruptIfRunning) {
    // 调用 FutureTask类中的 cancel() 方法
    boolean cancelled = super.cancel(mayInterruptIfRunning);
    // 取消成功 且 移除取消的成功 且 heapIndex(数组中位置下表) 大于等于0
    if (cancelled && removeOnCancel && heapIndex >= 0)
        // 移除自己
        remove(this);

    return cancelled;
}

14、FutureTask类中的run() 方法

// FutureTask 类中的 run(); 方法
public void run() {
    // 给任务对象添加runner属性值，值为当前线程
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;

    try {
        Callable<V> c = callable;
        // 判断callable 不为空以及 state状态 为NUM
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                // 调用c的 call返回，并接受返回值
                result = c.call();
                // 运行完成标识
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            // 完成之后，设置结果给属性
            if (ran)
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        // 是否中断
        if (s >= INTERRUPTING)
            // 处理中断情况
            handlePossibleCancellationInterrupt(s);
    }
}

15、FutureTask类中的run() 方法

// FutureTask 类中的 runAndReset(); 方法
protected boolean runAndReset() {
    // 给任务对象添加runner属性值，值为当前线程
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        // 失败返回false                           
        return false;
    
    boolean ran = false;
    // 任务状态
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                // 调用sft任务对象的callable属性对应的对象，call()方法
                // 由于是周期任务，调用call返回之后不会接收返回值
                c.call(); // don't set result
                // 调用完成标识
                ran = true;
            } catch (Throwable ex) {
                setException(ex);
            }
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        // 任务的运行线程属性置空
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        // 重置后任务状态不变
        s = state;
        if (s >= INTERRUPTING)
            // 状态改变成中断，进行中断处理
            handlePossibleCancellationInterrupt(s);
    }
    // 运行完成 且 状态还是 NEW 返回成功
    return ran && s == NEW;
}

16、FutureTask类中的 cancel() 方法

// FutureTask类中的 cancel() 方法
// 参数 mayInterruptIfRunning ：如果任务运行是否可以被中断
// 取消任务
public boolean cancel(boolean mayInterruptIfRunning) {
    // 任务状态是 新建NEW状态，且CAS修改任务状态为 INTERRUPTING(中断) 或: CANCELLED(取消)
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        // CAS失败，返回false
        return false;

    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                // 获取运行线程，
                // runner由来：任务FutureTask，运行时，会CAS设置 runner = Thread.currentThread();
                Thread t = runner;
                if (t != null)
                    // 中断线程
                    t.interrupt();
            } finally { // final state
                // 修改任务状态为INTERRUPTED 中断
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        // 
        finishCompletion();
    }
    return true;
}

17、FutureTask 类中的 finishCompletion() 方法

// FutureTask类中的 完成节点设置
private void finishCompletion() {
    // assert state > COMPLETING;
    // 遍历等待节点
    for (WaitNode q; (q = waiters) != null;) {
        // 修改将等待节点设为null，循环节点
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                // 节点线程不为空
                if (t != null) {
                    // 将节点thread属性设为null
                    q.thread = null;
                    // 唤醒线程t
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                // q指向下一个节点
                q = next;
            }
            break;
        }
    }

    // 钩子方法
    done();

    callable = null;        // to reduce footprint
}

18、DelayedWorkQueue 类中的 take() 方法

public RunnableScheduledFuture<?> take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    // 获取锁
    lock.lockInterruptibly();
    try {
        for (;;) {
            // 查看队列头的元素
            RunnableScheduledFuture<?> first = queue[0];
            if (first == null)
                // 队列无用元素，线程等待可用
                available.await();
            else {
                // 第一个，也是最近一个可用元素的延迟时间
                long delay = first.getDelay(NANOSECONDS);
                // <= 0表示队头元素有过期或者已到期
                if (delay <= 0)
                    // 完成拉取相关操作
                    return finishPoll(first);
                // 不存在过期元素
                first = null; // don't retain ref while waiting
                // 前面是否还有消费线程等待消费
                if (leader != null)
                    // 等待
                    available.await();
                else {
                    // 设置本线程为最先消费线程
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 本线程等待 delay 时长，后被唤醒
                        // delay 时长 后，队列头第一个元素会过期
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 如果 leader 为空，且队列不为空，唤醒一个消费线程
        if (leader == null && queue[0] != null)
            // 通知消费线程，延迟队列可用
            available.signal();
        // 释放锁
        lock.unlock();
    }
}

19、DelayedWorkQueue 类中的 finishPoll() 方法

// 完成拉取相关操作
private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
    int s = --size;
    // 队列最后一个元素，由于数组第一位置空出来了，最后一个元素就需要上浮
    RunnableScheduledFuture<?> x = queue[s];
    // 数组位置空出来
    queue[s] = null;
    if (s != 0)
        // 二叉堆中的最小堆，下浮操作
        siftDown(0, x);
    // 设置数组下标，f是已取出来的元素，下标设为-1
    setIndex(f, -1);
    return f;
}

四、示例

public class TestScheduledThreadPool {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);

        ScheduledFuture<Integer> future1 = scheduledThreadPoolExecutor.schedule(() -> {
            System.out.println("我要延迟30s执行");
            return 1;
        }, 30_000, TimeUnit.MILLISECONDS);

        ScheduledFuture<Integer> future2 = scheduledThreadPoolExecutor.schedule(() -> {
            System.out.println("我要延迟20s执行");
            return 2;
        }, 20_000, TimeUnit.MILLISECONDS);

        ScheduledFuture<Integer> future3 = scheduledThreadPoolExecutor.schedule(() -> {
            System.out.println("我要延迟8s执行");
            return 1;
        }, 8_000, TimeUnit.MILLISECONDS);


        scheduledThreadPoolExecutor.shutdown();

        System.out.println(future1.get());
        System.out.println(future2.get());
        System.out.println(future3.get());

//        while (!scheduledThreadPoolExecutor.isTerminated()) {
//            Thread.sleep(1000);
//        }
        System.out.println("Game Over~~~");


    }
}