ScheduledThreadPoolExecutor --- DelayedWorkQueue源码解析

我们使用Java实现周期任务或者延时任务，一般都会采用ScheduledThreadPoolExecutor来实现。而翻看了源码以后发现，周期任务和延时任务实现的核心就是一个特殊的阻塞队列 — DelayedWorkQueue

ThreadPoolExecutor中，每一个线程都被封装成一个Worker对象，通过Worker对象中的getTask()方法不断拉取任务执行。

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {        	
        	// ... 调用task的run方法
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

从以上代码看出，getTask()取出来后会立刻调用run方法，因此没法做到延迟或周期效果。所以ScheduledThreadPoolExecutor自定义了一个队列以实现功能。

实现周期任务通常调用scheduleAtFixedRate()方法或者scheduleWithFixedDelay()方法。

public ScheduledFuture scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (period <= 0)
        throw new IllegalArgumentException();
    ScheduledFutureTask sft =
        new ScheduledFutureTask(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period));
    RunnableScheduledFuture t = decorateTask(command, sft);
    sft.outerTask = t;
    delayedExecute(t);
    return t;
}

从以上代码可以看出来，ScheduledThreadPoolExecutor提交任务时，会把任务封装成一个ScheduledFutureTask对象，这个类有两个关键的字段，用于标记这个任务的周期还有执行前的延时。

// 执行前的延时
private long time;

// 执行周期
private final long period

而这个对象重写了run方法，有了不一样的执行逻辑。

public void run() {
	// 判断这个任务是否需要周期执行
    boolean periodic = isPeriodic();
    // 判断当前状态是否可以运行
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    // 如果是一次性的任务，则直接调用一次run方法即可
    else if (!periodic)
        ScheduledFutureTask.super.run();
    // 如果是周期任务的话,调用run方法并且重置其返回值
    else if (ScheduledFutureTask.super.runAndReset()) {
    	// 设置这个任务下一次的执行时间
        setNextRunTime();
        // 重新加入到队列中
        reExecutePeriodic(outerTask);
    }
}

也就是说，周期任务的执行流程为:

1. 从队列中取出周期任务，调用run方法执行
2. 计算下一次运行的时间
3. 重新放入到队列中

DelayedWorkQueue的存储结果类似Java中的PriorityQueue，都是基于数组实现的堆。而内部实现了自动扩容的功能，每次扩容会变为原来容量的1.5倍。最大容量为Integer.MAX_VALUE，换而言之就是一个无界的队列。

在DelayedWorkQueue中有Leader线程这一概念。这是Leader-Follower模型的一个变种。
在这里的具体实现为，当第一个线程（A）来线程池中取任务时，A会被指定为Leader线程，假若线程池中的第一个任务还没到可执行的时间，那么A就会调用awaitNanos()进入有限时间的阻塞，直到任务可执行。而其他非Leader线程来取任务的话就会调用await()，一直阻塞被唤醒。当Leader线程执行完毕后去唤醒其他的线程消费任务。

为什么要这样子设计？

假设没有Leader线程，多个线程来取任务，而任务还没到达可执行的时间，于是所有的线程都会调用awaitNanos()进入有限时间的阻塞，而最终只有一个线程能拿到任务并且执行，然后其他线程需要重新取别的任务然后循环这个过程。这样子会造成很多次不必要的定时等待。

往队列加入任务时无论是add、put方法都是调用了offer方法，因此核心就是offer方法 (优先队列实现插入和删除元素的逻辑在我另一篇关于优先队列的文章里有详细分析，因此在此不再赘述)

public boolean offer(Runnable x) {
	// 为空判断
    if (x == null)
        throw new NullPointerException();
    // 向下转型
    RunnableScheduledFuture e = (RunnableScheduledFuture)x;
    // 由于数组是非线程安全的，所以对其进行操作的时候需要加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
    	// 判断是否需要扩容
        int i = size;
        if (i >= queue.length)
            grow();
        size = i + 1;
        // 堆插入
        if (i == 0) {
            queue[0] = e;
            setIndex(e, 0);
        } else {
            siftUp(i, e);
        }
        // 假如队列里只有当前任务，则需要唤醒在等待的线程
        if (queue[0] == e) {
            leader = null;
            available.signal();
        }
    } finally {
        lock.unlock();
    }
    return true;
}

用于获取队列头的任务

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);
                if (delay <= 0)
                    return finishPoll(first);
                // 如果任务还没到执行时间的话，先释放资源
                first = null;
                // 判断是否有Leader线程，有的话则进入阻塞。
                if (leader != null)
                    available.await();
                else {
               	// 没有的话当前线程成为Leader线程并且进入定时阻塞
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
    	// 假如没有leader线程并且有任务可以消费的话，则唤醒别的线程
        if (leader == null && queue[0] != null)
            available.signal();
        lock.unlock();
    }
}

与take方法相似，不同的是，如果当前任务还没到执行时间，则直接返回null

public RunnableScheduledFuture poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        RunnableScheduledFuture first = queue[0];
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            return null;
        else
            return finishPoll(first);
    } finally {
        lock.unlock();
    }
}

以上内容只是对一些比较难理解的核心源码进行分析，一些比较简单的细节和过程还是需要自己去看源码理解的。