java.util.concurrent 包源码分析之线程池

本文主要是介绍java.util.concurrent 包源码分析之线程池，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

concurrent包中Executor接口的主要类的关系图如下：

这里写图片描述

Executor

Executor接口非常单一，就是执行一个Runnable的命令。

public interface Executor {void execute(Runnable command);
}

ExecutorService

ExecutorService接口扩展了Executor接口，增加状态控制，执行多个任务返回Future。

关于状态控制的方法：

// 发出关闭信号，不会等到现有任务执行完成再返回，但是现有任务还是会继续执行，可以调用awaitTermination等待所有任务执行。不再接受新的任务。void shutdown();// 立刻关闭，尝试取消正在执行的任务（不保证会取消成功），返回未被执行的任务List<Runnable> shutdownNow();// 是否发出关闭信号
boolean isShutdown();// 是否所有任务都执行完毕在shutdown之后，也就是如果不调用shutdownNow或者，shutdown是不可能返回trueboolean isTerminated();// 进行等待直到所有任务完成或者超时
boolean awaitTermination(long timeout, TimeUnit unit)throws InterruptedException;

提交单个任务，立刻返回一个Future存储任务执行的实时状态

<T> Future<T> submit(Callable<T> task);<T> Future<T> submit(Runnable task, T result);Future<?> submit(Runnable task);

执行多个任务的方法，有两种方式，一种等到所有任务执行完成才返回：

 <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>>   tasks) throws InterruptedException;<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit) throws InterruptedException;

另外一种是等到有一个任务完成，取消其他未完成的任务，返回执行完成的任务的执行结果：

<T> T invokeAny(Collection<? extends Callable<T>> tasks)throws InterruptedException, ExecutionException;<T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;

通过上面的代码可以看出ExecutorService可以执行两种类型的任务：Runnable和Callable，而Callable用的更加多。两者区别很简单，前者不会返回执行结果而后者会返回一个执行结果：

public interface Callable<V> {V call() throws Exception;
}

接着说说Future，也就是执行任务的返回类型。Future可以看成是一张发票。比如你送件衣服到洗衣店清洗，他们会开张发票给你，你拿着发票可以去拿回你洗好的衣服或者去洗衣店问衣服是否洗好了等等。

public interface Future<V> {//取消任务，参数mayInterruptIfRunning为true时，如果要取消的任务正在执行，会把执行这个任务的线程设为中断，为false时，正在执行的任务会被允许执行完成boolean cancel(boolean mayInterruptIfRunning);boolean isCancelled();boolean isDone();//获取执行结果，如果任务执行中，会等到任务完成再返回V get() throws InterruptedException, ExecutionException;V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
}

ScheduledExecutorService

ScheduledExecutorService，该接口是ExecutorService的子接口，增加了定时执行任务的功能：

public interface ScheduledExecutorService extends ExecutorService {public ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit);public <V> ScheduledFuture<V> schedule(Callable<V> callable,long delay, TimeUnit unit);// 等待一定时间然后开始执行一个任务，每隔period参数设置的时间// 重复一次，（多线程执行）public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit);// 等待一定时间然后开始执行一个任务，完成后，等待delay参数设置的时间// 然后在执行一次任务。（单线程执行）public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);}

AbstractExecutorService

AbstractExecutorService对ExecutorService的执行任务类型的方法提供了一个默认实现。这些方法包括submit，invokeAny和InvokeAll。

注意的是来自Executor接口的execute方法是未被实现，execute方法是整个体系的核心，所有的任务都是在这个方法里被真正执行的，因此该方法的不同实现会带来不同的执行策略。这个在后面分析ThreadPoolExecutor和ScheduledThreadPoolExecutor就能看出来。

首先来看submit方法，它的基本逻辑是这样的：

生成一个任务类型和Future接口的包装接口RunnableFuture的对象
执行任务
返回future。

    public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;}public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;}

因为submit支持Callable和Runnable两种类型的任务，因此newTaskFor方法有两个重载方法：

   protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable);}protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {return new FutureTask<T>(runnable, value);}

曾经说过Callable和Runnable的区别在于前者带返回值，也就是说Callable=Runnable+返回值。因此java中提供了一种adapter，把Runnable+返回值转换成Callable类型。这点可以在newTaskFor中的FutureTask类型的构造函数的代码中看到：

  public FutureTask(Callable<V> callable) {if (callable == null)throw new NullPointerException();sync = new Sync(callable);}public FutureTask(Runnable runnable, V result) {sync = new Sync(Executors.callable(runnable, result));}

以下是Executors.callable方法的代码：

  public static <T> Callable<T> callable(Runnable task, T result) {if (task == null)throw new NullPointerException();return new RunnableAdapter<T>(task, result);}

那么RunnableAdapter的代码就很好理解了，它是一个Callable的实现，call方法的实现就是执行Runnable的run方法，然后返回那个value。

    static final class RunnableAdapter<T> implements Callable<T> {final Runnable task;final T result;RunnableAdapter(Runnable task, T result) {this.task = task;this.result = result;}public T call() {task.run();return result;}}

接下来先说说较为简单的invokeAll：

为每个task调用newTaskFor方法生成得到一个既是Task也是Future的包装类对象的List
循环调用execute执行每个任务
再次循环调用每个Future的get方法等待每个task执行完成
最后返回Future的list。

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit) throws InterruptedException {if (tasks == null || unit == null)throw new NullPointerException();long nanos = unit.toNanos(timeout);List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());boolean done = false;try {// 为每个task生成包装对象for (Callable<T> t : tasks)futures.add(newTaskFor(t));long lastTime = System.nanoTime();// 循环调用execute执行每个方法// 这里因为设置了超时时间，所以每次执行完成后// 检查是否超时，超时了就直接返回future集合Iterator<Future<T>> it = futures.iterator();while (it.hasNext()) {execute((Runnable)(it.next()));long now = System.nanoTime();nanos -= now - lastTime;lastTime = now;if (nanos <= 0)return futures;}// 等待每个任务执行完成for (Future<T> f : futures) {if (!f.isDone()) {if (nanos <= 0)return futures;try {f.get(nanos, TimeUnit.NANOSECONDS);} catch (CancellationException ignore) {} catch (ExecutionException ignore) {} catch (TimeoutException toe) {return futures;}long now = System.nanoTime();nanos -= now - lastTime;lastTime = now;}}done = true;return futures;} finally {if (!done)for (Future<T> f : futures)f.cancel(true);}}

invokeAny，它的难点在于只要一个任务执行成功就要返回，并且会取消其他任务，也就是说重点在于找到第一个执行成功的任务。

这里我想到了BlockingQueue，当所有的任务被提交后，任务执行返回的Future会被依次添加到一个BlockingQueue中，然后找到第一个执行成功任务的方法就是从BlockingQueue取出第一个元素，这个就是doInvokeAny方法用到的ExecutorCompletionService的基本原理。

因为两个invokeAny方法都是调用doInvokeAny方法，下面是doInvokeAny的代码分析：

  private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,boolean timed, long nanos)throws InterruptedException, ExecutionException, TimeoutException {if (tasks == null)throw new NullPointerException();int ntasks = tasks.size();if (ntasks == 0)throw new IllegalArgumentException();List<Future<T>> futures= new ArrayList<Future<T>>(ntasks);// ExecutorCompletionService负责执行任务，后面调用用poll返回第一个执行结果ExecutorCompletionService<T> ecs =new ExecutorCompletionService<T>(this);// 这里出于效率的考虑，每次提交一个任务之后，就检查一下有没有执行完成的任务try {ExecutionException ee = null;long lastTime = timed ? System.nanoTime() : 0;Iterator<? extends Callable<T>> it = tasks.iterator();// 先提交一个任务futures.add(ecs.submit(it.next()));--ntasks;int active = 1;for (;;) {// 尝试获取有没有执行结果（这个结果是立刻返回的）Future<T> f = ecs.poll();// 没有执行结果if (f == null) {// 如果还有任务没有被提交执行的，就再提交一个任务if (ntasks > 0) {--ntasks;futures.add(ecs.submit(it.next()));++active;}// 没有任务在执行了，而且没有拿到一个成功的结果。else if (active == 0)break;// 如果设置了超时情况else if (timed) {// 等待执行结果直到有结果或者超时f = ecs.poll(nanos, TimeUnit.NANOSECONDS);if (f == null)throw new TimeoutException();// 这里的更新不可少，因为这个Future可能是执行失败的情况，那么还需要再次等待下一个结果，超时的设置还是需要用到。long now = System.nanoTime();nanos -= now - lastTime;lastTime = now;}// 没有设置超时，并且所有任务都被提交了，则一直等到第一个执行结果出来elsef = ecs.take();}// 有返回结果了，尝试从future中获取结果，如果失败了，那么需要接着等待下一个执行结果if (f != null) {--active;try {return f.get();} catch (ExecutionException eex) {ee = eex;} catch (RuntimeException rex) {ee = new ExecutionException(rex);}}}// ExecutorCompletionService执行时发生错误返回了全是null的futureif (ee == null)ee = new ExecutionException();throw ee;} finally {// 尝试取消所有的任务（对于已经完成的任务没有影响）for (Future<T> f : futures)f.cancel(true);}}

ThreadPoolExecutor

ThreadPoolExecutor的execute方法，这个方法能体现出一个Task被加入到线程池之后都发生了什么：

    public void execute(Runnable command) {if (command == null)throw new NullPointerException();/* 如果运行中的worker线程数少于设定的常驻线程数，增加worker线程，把task分配给新建的worker线程 */int c = ctl.get();if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// 如果任务可以被加入到任务队列中，即等待的任务数还在允许的范围内，// 再次检查线程池是否被关闭，如果关闭的话，则移除任务并拒绝该任务if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();if (! isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 如果任务数超过了现有worker线程的承受范围，尝试新建worker线程// 如果无法添加新的worker线程，则会拒绝该任务else if (!addWorker(command, false))reject(command);}

在执行任务时，需要经常检查线程池的状态，那么接下来说说线程池是如何进行状态控制的。上面的代码有个成员变量叫做ctl，它用于标记线程池状态和worker线程的数量，是一个AutomaticInteger对象。

  private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

ctl是一个32位的整数，最高的3位表示状态：

111为running，

000为shutdown，

001为stop，

010为tidying，

011为ternimated。

因此状态值就是这三位加上29个0，因此running的值是个负整数（最高位为1），其他状态都是正整数，后面判断状态会比较值的大小时会用到这点。

剩下的29位表示worker线程的数量（因此最大允许的线程数就是2的29方减1）。

这里是说说这几个状态的意义，这几个状态发生的顺序正好就是上面列出的顺序：

running表示正常运行状态

shutdown状态意味着发出了一个shutdown信号，类似于你点击了windows的关机按钮

stop表示shutdown信号收到，等于windows响应了这个信号，发出正在关机的信息

tidying发生在stop之后做出的响应，表示这个时候在清理一些资源，

ternimated发生在tidying完成之后，表示关闭完成。

接着来看添加一个worker线程时都发生了什么：

  private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// 返回false的情况：// 1. rs>shutdown，即shutdown和running以外的状态// 2. shutdown的状态//     1）firstTask不为null，即有task分配//     2）没有task，但是workQueue（等待任务队列）为空if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {// 1. 如果没有设定线程数的限制，worker线程数不能大于最大值（2的29次方-1）// 2. 如果是固定尺寸的线程池，不能大于固定尺寸// 3. 如果是可扩展的线程池，不能大于规定的线程数的上限int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// 用CAS操作增加线程数量，如果失败，重新循环if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateOf(c) != rs)continue retry;loop}}// 新建worker线程Worker w = new Worker(firstTask);Thread t = w.thread;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {int c = ctl.get();int rs = runStateOf(c);// 检查以下任一状态是否出现：// 1. 创建线程失败// 2. rs>shutdown，即shutdown和running以外的状态// 3. rs==shutdown，有任务分配if (t == null ||(rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null))) {decrementWorkerCount();tryTerminate();return false;}workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;} finally {mainLock.unlock();}t.start();// 这里考虑一种极少出现的情况，如果worker线程调用start没有完成时，// 线程池进入Stop状态，这个时候会调用Thread#interrupt中断每个// worker线程，但是 interrupt对没有start的线程不一定起作用，这样// 就会漏掉了对这个thread的interrupt，因此在worker线程start之后// 检查以下，如果stop了，而这个线程却没有被interrupt，补上这个漏掉// 的interrupt。if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted())t.interrupt();return true;}

ThreadPoolExecutor

ThreadPoolExecutor有一个成员类叫Worker，所起到的作用就是线程池worker线程的作用。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable

这里AbstractQueuedSynchronizer的作用是使Worker具有锁的功能，在执行任务时，会把Worker锁住，这个时候就无法中断Worker。Worker空闲时候是线程池可以通过获取锁，改变Worker的某些状态，在此期间因为锁被占用，Worker就是不会执行任务的。

Worker工作的逻辑在ThreadPoolExecutor#runWorker方法中

      public void run() {runWorker(this);}

因此转到runWorker方法：

final void runWorker(Worker w) {Runnable task = w.firstTask;w.firstTask = null;boolean completedAbruptly = true;try {// 执行分配的任务或者从BlockingQueue中等待获取任务while (task != null || (task = getTask()) != null) {w.lock();clearInterruptsForTaskRun();try {// 执行任务之前的工作beforeExecute(w.thread, task);Throwable thrown = null;// 执行任务，如果发生异常，该Worker就不会再继续执行任务try {task.run();} catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {thrown = x; throw new Error(x);} finally {// 任务执行完的工作afterExecute(task, thrown);}} finally {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {// Worker不再执行任务的处理，completedAbruptly为false// 表示正常结束，否则表示执行任务出错。processWorkerExit(w, completedAbruptly);}}

来看看processWorkerExit，重点看看执行任务发生异常时该如何处理

   private void processWorkerExit(Worker w, boolean completedAbruptly) {// 发生异常，首先要更新Worker数量if (completedAbruptly)decrementWorkerCount();// 移除这个Workerfinal ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}// 尝试停止线程池，正常运行的线程池调用该方法不会有任何动作tryTerminate();int c = ctl.get();// 如果线程池没有被关闭的话，if (runStateLessThan(c, STOP)) {// Worker不是异常退出，检查worker线程数是不是小于最小值// 这个最小值分为几种情况：// 1. allowCoreThreadTimeOut（JDK6新加）表示是否允许线程池在超//     过一定时间没有收到任务后退出，这种情况下，最小值为0，因为如果如//     果一直没有任何任务，worker线程数是0// 2. 最小值为corePoolSize，因为corePoolSize可能为0，因此这种情况//     下，如果有任务的话必然会有Worker，因此最小值为1if (!completedAbruptly) {int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;if (workerCountOf(c) >= min)return;}// 如果Worker线程数小于最小值，新建一个Worker线程addWorker(null, false);}}

这一部分来说说线程池如何进行状态控制，即线程池的开启和关闭。

先来说说线程池的开启，这部分来看ThreadPoolExecutor构造方法：

    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

可以看到，尽管设定了corePoolSize，也就是Worker线程的数量，但是线程池开启的时候，默认是没有创建这些Worker线程的，但是ThreadPoolExecutor提供了prestartAllCoreThreads方法来开启所有的预设的Worker线程，以及prestartCoreThread尝试开启一个预设的Worker线程。

这里重点说说handler，也就是RejectedExecutionHandler，拒绝任务的处理类，ThreadPoolExecutor提供四种策略：

CallerRunsPolicy

该策略会在ThreadPoolExecutor没有关闭的情况，依旧运行任务

AbortPolicy

该策略会抛出一个RejectedExecutionException

DiscardPolicy

该策略直接忽略该任务，不会有任何动作

DiscardOldestPolicy

该策略会在ThreadPoolExecutor没有关闭的情况，丢弃下一个将要执行的任务，把该任务加入到执行队列。

接下来说说关闭，ThreadPoolExecutor提供了shutdown和shutdownNow两种方式，从字面上就能看出区别，后者会尝试结束正在运行的任务。

先来看shutdown：

 public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();advanceRunState(SHUTDOWN);interruptIdleWorkers();onShutdown(); // ScheduledThreadPoolExecutor的回调方法} finally {mainLock.unlock();}tryTerminate();}

再看shutdownNow:

    public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();advanceRunState(STOP);interruptWorkers();tasks = drainQueue();} finally {mainLock.unlock();}tryTerminate();return tasks;}

两个方法的代码非常相似，区别在于：

shutdownNow的状态设置为STOP，shutdown的状态是SHUTDOWN
shutdownNow会中断所有线程，也就是所有任务，而shutdown仅仅中断空闲线程，不会影响正在执行的任务。
shutdownNow会导出未执行的任务。

两个方法都用到的checkShutdownAccess方法主要是检查方法调用者是否有权限中断Worker线程。

advanceRunState方法用于设定线程的状态，如果状态值大于等于该状态值则会返回。Worker线程具备锁的功能，因此可以通过tryLock来判断Worker线程是否处于空闲状态，这是两个方法的区别所在。

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor是ThreadPoolExecutor的子类，同时实现了ScheduledExecutorService接口。

public class ScheduledThreadPoolExecutorextends ThreadPoolExecutorimplements ScheduledExecutorService

ScheduledThreadPoolExecutor的功能主要有两点：在固定的时间点执行（也可以认为是延迟执行），重复执行。

和分析ThreadPoolExecutor时一样，首先来看核心方法execute:

 public void execute(Runnable command) {schedule(command, 0, TimeUnit.NANOSECONDS);}

execute方法调用了另外一个方法schedule，同时我们发现三个submit方法也是同样调用了schedule方法，因为有两种类型的任务：Callable和Runnable，因此schedule也有两个重载方法。

    public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {if (command == null || unit == null)throw new NullPointerException();RunnableScheduledFuture<?> t = decorateTask(command,new ScheduledFutureTask<Void>(command, null,triggerTime(delay, unit)));delayedExecute(t);return t;}public <V> ScheduledFuture<V> schedule(Callable<V> callable,long delay,TimeUnit unit) {if (callable == null || unit == null)throw new NullPointerException();RunnableScheduledFuture<V> t = decorateTask(callable,new ScheduledFutureTask<V>(callable,triggerTime(delay, unit)));delayedExecute(t);return t;}

两个方法逻辑基本一致，都是把任务包装成RunnableScheduledFuture对象，然后调用delayedExecute来实现延迟执行。任务包装类继承自ThreadPoolExecutor的包装类RunnableFuture，同时实现ScheduledFuture接口使包装类具有了延迟执行和重复执行这些功能以匹配ScheduledThreadPoolExecutor。

因此首先来看ScheduledFutureTask，以下是ScheduledFutureTask专有的几个变量：

private class ScheduledFutureTask<V>extends FutureTask<V> implements RunnableScheduledFuture<V> {/** 针对线程池所有任务的序列号 */private final long sequenceNumber;/** 距离任务开始执行的时间，纳秒为单位 */private long time;/*** 重复执行任务的间隔，即每隔多少时间执行一次任务*/private final long period;/** 重复执行任务和排队时用这个类型的对象， */RunnableScheduledFuture<V> outerTask = this;/*** 在延迟队列的索引，这样取消任务时使用索引会加快查找速度*/int heapIndex;

来看核心方法run:

    public void run() {boolean periodic = isPeriodic();// 检测是否可以运行任务，这里涉及到另外两个变量：continueExistingPeriodicTasksAfterShutdown// 和executeExistingDelayedTasksAfterShutdown// 前者允许在shutdown之后继续执行重复执行的任务// 后者允许在shutdown之后继续执行延时执行的任务，// 因此这里根据任务是否为periodic来决定采用哪个选项，然后// 如果线程池正在运行，那么肯定可以执行// 如果正在shutdown，那么要看选项的值是否为true来决定是否允许执行任务// 如果不被允许的话，就会取消任务if (!canRunInCurrentRunState(periodic))cancel(false);// 如果可以执行任务，对于不用重复执行的任务，直接执行即可else if (!periodic)ScheduledFutureTask.super.run();// 对于需要重复执行的任务，则执行一次，然后reset// 更新一下下次执行的时间，调用reExecutePeriodic更新任务在执行队列的// 位置（其实就是添加到队列的末尾）else if (ScheduledFutureTask.super.runAndReset()) {setNextRunTime();reExecutePeriodic(outerTask);}}

因此这里可以得出关于重复执行的实现：任务执行一次，Reset状态，重新加入到任务队列。

回到delayedExecute，它可以保证任务在准确时间点执行，来看delayedExecute是如果实现延迟执行的：

   private void delayedExecute(RunnableScheduledFuture<?> task) {if (isShutdown())reject(task);else {super.getQueue().add(task);if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))task.cancel(false);elseensurePrestart();}}

乍看之下，发现也就是把任务加入到任务队列中，那么这个延时执行的功能是如何实现的，秘密就在任务队列的实现。

 public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,new DelayedWorkQueue());}public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);}

ScheduledThreadPoolExecutor的任务队列不是普通的BlockingQueue，而是一个特殊的实现DelayedWorkQueue。

在ScheduledThreadPoolExecutor使用DelayedWorkQueue来存放要执行的任务，因为这些任务是带有延迟的，而每次执行都是取第一个任务执行，因此在DelayedWorkQueue中任务必然按照延迟时间从短到长来进行排序的。

DelayedWorkQueue使用堆来实现的。

和以前分析BlockingQueue的实现类一样，首先来看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;// 这个i就是RunnableScheduledFuture用到的heapIndexsetIndex(e, 0);} else {// 添加元素到堆中siftUp(i, e);}// 如果队列原先为空，那么可能有线程在等待元素，这时候既然添加了元// 素，就需要通过Condition通知这些线程if (queue[0] == e) {// 因为有元素新添加了，第一个等待的线程可以结束等待了，因此这里// 删除第一个等待线程leader = null;available.signal();}} finally {lock.unlock();}return true;}

这里顺带看一下siftUp，熟悉堆的实现的朋友应该很容易看懂这是一个把元素添加已有堆中的算法。

     private void siftUp(int k, RunnableScheduledFuture key) {while (k > 0) {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);}

那么接着就看看poll：

       public RunnableScheduledFuture poll() {final ReentrantLock lock = this.lock;lock.lock();try {// 因为即使拿到任务，线程还是需要等待，而这个等待过程是由队列帮助完成的// 因此poll方法只能返回已经到执行时间点的任务RunnableScheduledFuture first = queue[0];if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)return null;elsereturn finishPoll(first);} finally {lock.unlock();}}

因为poll方法只能返回已经到了执行时间点的任务，所以对于我们理解队列如何实现延迟执行没有意义，因此重点看看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(TimeUnit.NANOSECONDS);// 如果任务不用等待，立刻返回该任务给线程if (delay <= 0)// 从堆中拿走任务return finishPoll(first);// 如果任务需要等待，而且前面有个线程已经等待执行任务（leader线程// 已经拿到任务了，但是执行时间没有到，延迟时间肯定是最短的），// 那么执行take的线程肯定继续等待，else if (leader != null)available.await();// 当前线程的延迟时间是最短的情况，那么更新leader线程// 用Condition等待直到时间到点，被唤醒或者被中断else {Thread thisThread = Thread.currentThread();leader = thisThread;try {available.awaitNanos(delay);} finally {// 重置leader线程以便进行下一次循环if (leader == thisThread)leader = null;}}}}} finally {// 队列不为空发出signal很好理解，这里附带了没有leader线程// 的条件是因为leader线程存在时表示leader线程正在等待执行时间点的// 到来，如果此时发出signal会触发awaitNanos提前返回if (leader == null && queue[0] != null)available.signal();lock.unlock();}}

take方法的重点就是leader线程，因为存在延迟时间，即使拿到任务，线程还是需要等待的，leader线程就那个最先执行任务的线程。

因为线程拿到任务之后还是需要等待一段延迟执行的时间，所以对于超时等待的poll方法来说就有点意思了：

 public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit)throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {for (;;) {RunnableScheduledFuture<?> first = queue[0];if (first == null) {if (nanos <= 0)return null;elsenanos = available.awaitNanos(nanos);} else {long delay = first.getDelay(NANOSECONDS);if (delay <= 0)return finishPoll(first);if (nanos <= 0)return null;first = null; // don't retain ref while waitingif (nanos < delay || leader != null)nanos = available.awaitNanos(nanos);else {Thread thisThread = Thread.currentThread();leader = thisThread;try {long timeLeft = available.awaitNanos(delay);nanos -= delay - timeLeft;} finally {if (leader == thisThread)leader = null;}}}}} finally {if (leader == null && queue[0] != null)available.signal();lock.unlock();}}