Java并发包线程池之ScheduledThreadPoolExecutor

前言

它是一种可以安排在给定的延迟之后执行一次或周期性执行任务的ThreadPoolExecutor。因为它继承了ThreadPoolExecutor, 当然也具有处理普通Runnable、Callable 任务的能力，当需要多个工作线程辅助时，或者当需要 ThreadPoolExecutor 具有额外的灵活性或功能(该类扩展的这些功能)时，这个类通常比Timer更好。

延迟或周期任务一旦延迟/周期时间到达变得可执行时，它们具体在何时会被线程池调度执行是无法实时保障的，因为同一时期可能有多个任务都需要被执行，该类会按照先进先出FIFO的原则调度执行那些被安排在同一执行时间的任务。

当提交的任务在执行之前被取消，则不会再被执行，但是默认情况下，被取消的任务不会自动从工作队列中被删除，直到它的延迟或者周期时间到达被线程池调度的时候才会从任务队列中被取出，发现已经取消则放弃执行并且不安排周期任务下一次执行，为了避免这种情况，可以调用setRemoveOnCancelPolicy(true)来启用被取消的任务立即从工作队列中删除的特性，默认不开启此特性。

通过scheduleAtFixedRate或scheduleWithFixedDelay提交的周期任务的连续执行不会重叠(即由于某次执行时间较长，前后两次执行不会同时进行)。虽然每一次的执行都是由不同的线程来执行，但是前一次执行的效果会 happen-before 后续执行的效果。周期任务一旦某一次执行任务抛出异常将取消后续的周期执行。

虽然此类继承了ThreadPoolExecutor,但使用了一个无界的任务队列（延迟工作队列DelayedWorkQueue），因此对maximumPoolSize的调整将对此类没有任何作用。此外，将corePoolSize设置为零或使用allowCoreThreadTimeOut几乎从来都不是一个好主意，因为这可能会使线程池在需要运行任务时没有线程来处理这些任务。

扩展注意事项，该类重写了execute和submit方法来生成内部ScheduledFuture对象，以控制每个任务的延迟和调度。为了保持功能，子类中对这些方法的任何进一步重写都必须依赖父类的版本，这将有效地避免自定义额外的任务类型。但是该类提供了另一种可被重写的方法decorateTask (Runnable和Callable各有一个版本)，可定制用于通过execute、submit、schedule、scheduleAtFixedRate和scheduleWithFixedDelay方法传入的任务的具体类型，默认情况下，ScheduledThreadPoolExecutor使用一个扩展了FutureTask、RunnableScheduledFuture的任务类型ScheduledFutureTask。但是，可以使用下列形式的子类修改或替换该类型：

 1 public class CustomScheduledExecutor extends ScheduledThreadPoolExecutor {
 2     
 3     //自定义RunnableScheduledFuture
 4    static class CustomTask<V> implements RunnableScheduledFuture<V> { ... }
 5 
 6    //重写decorateTask，返回自定义的RunnableScheduledFuture
 7    protected <V> RunnableScheduledFuture<V> decorateTask(
 8                 Runnable r, RunnableScheduledFuture<V> task) {
 9        return new CustomTask<V>(r, task);
10    }
11 
12    protected <V> RunnableScheduledFuture<V> decorateTask(
13                 Callable<V> c, RunnableScheduledFuture<V> task) {
14        return new CustomTask<V>(c, task);
15    }
16    // ... add constructors, etc.
17 }

View Code

源码解读

属性字段

 1 public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService{
 2 
 3 //是否在线程池shutdown之后继续执行周期任务，为false时，将会在线程池shutdown时取消/废弃周期任务
 4 private volatile boolean continueExistingPeriodicTasksAfterShutdown;
 5 
 6 
 7 //是否在线程池shutdown之后继续执行延迟任务，为false时，将会在线程池shutdown时取消非周期任务(即取消延迟任务)
 8 private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
 9 
10 //为true时，将会在取消任务时立即从任务队列中删除该任务
11 private volatile boolean removeOnCancel = false;
12 
13 //保证任务的先进先出的序列
14 private static final AtomicLong sequencer = new AtomicLong();
15 
16 //返回当前纳秒时间。
17 final long now() {
18     return System.nanoTime();
19 }

View Code

他的属性字段只有四个，continueExistingPeriodicTasksAfterShutdown，executeExistingDelayedTasksAfterShutdown用于指示当线程池shutdown之后(仅限线程池处于SHUTDOWN状态的时候)是否继续执行周期/延迟任务，它们并不能阻止线程池即将被终结的命运。removeOnCancel指示是否在任务被取消后立即从任务队列中移除。sequencer用于记录任务进入工作队列的顺序，以便将来两个任务撞到同一时间需要被执行时，决定谁先谁后的问题。

虽然 continueExistingPeriodicTasksAfterShutdown会导致即使线程池处于SHUTDOWN状态也会继续将周期任务加入到线程池队列等待下一次被调度执行，但一旦线程池调度之后发现任务队列为空，线程池依然会走向终结，因为最后一个周期任务从队列中取走执行的时候，还没来得及将周期任务再次添加到任务队列，线程池可能就因为发现任务队列为空，从SHUTDOWN过渡到STOP或者TIDYING或者TERMINATED状态了，这个时候即使任务已经进入了任务队列也不会被调度执行。这两个参数仅仅可以使处于SHUTDOWN状态的线程池可以继续执行延迟或周期任务，但线程池从SHUTDOWN状态过渡到后面的状态是自动进行的可能就是瞬息之间，并不能保证这两个参数就一定会发挥作用，我觉得没什么用处显得很鸡肋。

RunnableScheduledFuture实现--- ScheduledFutureTask

ThreadPoolExecutor默认将Runnable、Callable的任务封装成了FutureTask的任务调度执行，该类为了实现任务的延迟和周期调度执行扩展了默认的FutureTask并实现了RunnableScheduledFuture：

  1 private class ScheduledFutureTask<V>
  2         extends FutureTask<V> implements RunnableScheduledFuture<V> {
  3 
  4     /** FIFO的序列号 */
  5     private final long sequenceNumber;
  6 
  7     /** 以纳秒为单位的下一次任务执行的时间 */
  8     private long time;
  9 
 10     //以纳秒为单位的重复任务的周期:
 11     //正数表示固定频率的周期任务(scheduleAtFixedRate)，
 12     //负数表示固定延迟的周期任务(scheduleWithFixedDelay)。
 13     //0表示不是一个重复执行的周期任务。
 14     private final long period;
 15 
 16     /** 被周期执行的任务，指向的自身 */
 17     RunnableScheduledFuture<V> outerTask = this;
 18 
 19     /**
 20      * 任务在延迟数组中的索引，以便快速查找、取消任务.
 21      */
 22     int heapIndex;
 23 
 24     //创建一个基于给定纳秒数延迟的一次性Runnable任务
 25     ScheduledFutureTask(Runnable r, V result, long ns) {
 26         super(r, result);
 27         this.time = ns;
 28         this.period = 0;
 29         this.sequenceNumber = sequencer.getAndIncrement();
 30     }
 31 
 32 
 33     //创建一个基于给定纳秒数延迟和周期的周期性Runnable任务 
 34     ScheduledFutureTask(Runnable r, V result, long ns, long period) {
 35         super(r, result);
 36         this.time = ns;
 37         this.period = period;
 38         this.sequenceNumber = sequencer.getAndIncrement();
 39     }
 40 
 41     //创建一个基于给定纳秒数延迟的一次性Callable任务 
 42     ScheduledFutureTask(Callable<V> callable, long ns) {
 43         super(callable);
 44         this.time = ns;
 45         this.period = 0;
 46         this.sequenceNumber = sequencer.getAndIncrement();
 47     }
 48 
 49     //Delayed接口实现，返回值小于等于0时表示延迟已经结束
 50     public long getDelay(TimeUnit unit) {
 51         return unit.convert(time - now(), NANOSECONDS);
 52     }
 53 
 54     //判断两个任务的先后执行顺序
 55     public int compareTo(Delayed other) {
 56         if (other == this) // 同一个对象
 57             return 0;
 58         if (other instanceof ScheduledFutureTask) { //是ScheduledFutureTask任务
 59             ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
 60             //直接比较延迟时间，若延迟时间相同则按先进先出的策略
 61             long diff = time - x.time;
 62             if (diff < 0)
 63                 return -1;
 64             else if (diff > 0)
 65                 return 1;
 66             else if (sequenceNumber < x.sequenceNumber) //按FIFO决定顺序
 67                 return -1;
 68             else
 69                 return 1;
 70         }
 71         //非ScheduledFutureTask，即其它类型的延迟任务，借助getDelay比较
 72         long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
 73         return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
 74     }
 75 
 76     //实现RunnableScheduledFuture接口，如果是周期重复执行的任务返回true
 77     public boolean isPeriodic() {
 78         return period != 0; //只要不是0，就是一个周期执行的任务
 79     }
 80 
 81 
 82     //设置周期任务下一次的运行时间。 
 83     private void setNextRunTime() {
 84         long p = period;
 85         if (p > 0) //固定速率 fixed-rate：time,time + p,time +2*p,time + 3*p......
 86             time += p;
 87         else       //固定延迟 fixed-delay: now + |p| 即当前系统时间 加周期时间
 88             time = triggerTime(-p); //这是p是负数，所以取反传递的是正数
 89     }
 90 
 91     //取消任务，重写FutureTask是为了实现实时将被取消的任务从工作队列中移除
 92     public boolean cancel(boolean mayInterruptIfRunning) {
 93         //直接使用FutureTask的cancel
 94         boolean cancelled = super.cancel(mayInterruptIfRunning);
 95         //取消任务之后，如果removeOnCancel为true，并且任务还存在延迟队列中(heapIndex >=0)
 96         if (cancelled && removeOnCancel && heapIndex >= 0)
 97             remove(this);    //将被取消的任务从工作队列中移除
 98         return cancelled;
 99     }
100 
101     //重写 FutureTask的run方法，是为了将需要周期执行的任务重置/再次入队。
102     public void run() {
103         boolean periodic = isPeriodic();
104         //如果线程池的状态至少是STOP，或者是SHUTDOWN但是参数指定不在shutdown状态继续执行任务，则取消任务
105         if (!canRunInCurrentRunState(periodic))
106             cancel(false);
107         //不是周期任务，直接执行FutureTask.run
108         else if (!periodic)        
109             ScheduledFutureTask.super.run();
110         //是周期执行的任务，
111         else if (ScheduledFutureTask.super.runAndReset()) { //任务正常结束
112             setNextRunTime();    //设置周期任务的下次执行时间
113             reExecutePeriodic(outerTask); //将当前任务再次加入线程池队列
114         }
115     }
116 }

View Code

ScheduledFutureTask就是被ScheduledThreadPoolExecutor线程池调度执行的任务实体，其成员属性中，sequenceNumber表示创建该任务的顺序，在构造方法中原子的自增。time是该任务下一次被调度执行的剩余延迟纳秒时长，period则是通过scheduleAtFixedRate和scheduleWithFixedDelay提交的周期任务的周期，非周期任务该字段为0。heapIndex是该任务在任务队列（平衡二叉堆实现的延迟队列DelayedWorkQueue）中的索引，方便快速查找和取消任务，如果任务已经从任务队列中移除，该值为-1。其构造方法与FutureTask一样支持Runnable（可以指定返回结果result）、Callable的任务，只是扩展多了延迟时间和周期时间参数。

该类实现的Delayed接口的getDelay方法以及compareTo是根据下一次任务执行的延迟剩余时间进行比较的，对于剩余延迟时间相同的根据sequenceNumber按先进先出FIFO的策略决定顺序。其run、cancel方法后面再说。

任务队列--DelayedWorkQueue

不像ThreadPoolExecutor那样可以支持任意阻塞队列作为任务队列，为了实现任务的延迟和周期执行，ScheduledThreadPoolExecutor只能使用内部实现的延迟任务队列DelayedWorkQueue，该类已经在前面的Java同步数据结构之DelayQueue/DelayedWorkQueue学习过了。其原理就是数组实现的平衡二叉堆，根据元素的延迟剩余时间多少分布成二叉堆，只有当延迟时间结束，元素才能被取出队列，并且取出的元素一定是当前队列中延迟时间最短的，这里的元素当然就是一个个ScheduledFutureTask任务实体，这里就不再对该内部类进行分析了，需要注意的是，DelayedWorkQueue在内部实现的时候，会将ScheduledFutureTask任务实体在数组中的索引记录到ScheduledFutureTask的heapIndex字段中，以方便快速查找和取消任务，并且当从任务队列中移除该任务时会将heapIndex置为-1.

构造方法

介绍完了ScheduledThreadPoolExecutor的内部字段与任务实体和任务队列这些基础设施，下面开始其构造方法的分析。这才是用户的使用入口。

 1 //创建具有给定核心线程数的ScheduledThreadPoolExecutor线程池
 2 public ScheduledThreadPoolExecutor(int corePoolSize) {
 3     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
 4           new DelayedWorkQueue());
 5 }
 6 
 7 //创建具有给定核心线程数和线程工厂的ScheduledThreadPoolExecutor线程池
 8 public ScheduledThreadPoolExecutor(int corePoolSize,
 9                                    ThreadFactory threadFactory) {
10     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
11           new DelayedWorkQueue(), threadFactory);
12 }
13 
14 //创建具有给定核心线程数和拒绝策略的ScheduledThreadPoolExecutor线程池
15 public ScheduledThreadPoolExecutor(int corePoolSize,
16                                        RejectedExecutionHandler handler) {
17     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
18           new DelayedWorkQueue(), handler);
19 }
20 
21 //创建具有给定核心线程数、线程工厂和拒绝策略的ScheduledThreadPoolExecutor线程池
22 public ScheduledThreadPoolExecutor(int corePoolSize,
23                                    ThreadFactory threadFactory,
24                                    RejectedExecutionHandler handler) {
25     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
26           new DelayedWorkQueue(), threadFactory, handler);
27 }

View Code

ScheduledThreadPoolExecutor有以上四个构造方法，分别只能对核心线程数corePoolSize、线程工厂和拒绝策略handler进行指定，其它参数都是内置的，maximumPoolSize都是Integer.MAX_VALUE，keepAliveTime都是0，任务队列都是DelayedWorkQueue。由于这里的任务队列DelayedWorkQueue是一个无界队列，因此对maximumPoolSize、keepAliveTime的任何设置与更新都是无任何意义的。

任务的提交

  1 // Override AbstractExecutorService methods
  2 //普通非延迟非周期任务的提交----execute、submit
  3 public void execute(Runnable command) {
  4     schedule(command, 0, NANOSECONDS);
  5 }
  6 
  7 // Override AbstractExecutorService methods
  8 
  9 /**
 10  * @throws RejectedExecutionException 
 11  * @throws NullPointerException       
 12  */
 13 public Future<?> submit(Runnable task) {
 14     return schedule(task, 0, NANOSECONDS);
 15 }
 16 
 17 /**
 18  * @throws RejectedExecutionException 
 19  * @throws NullPointerException       
 20  */
 21 public <T> Future<T> submit(Runnable task, T result) {
 22     return schedule(Executors.callable(task, result), 0, NANOSECONDS);
 23 }
 24 
 25 /**
 26  * @throws RejectedExecutionException 
 27  * @throws NullPointerException       
 28  */
 29 public <T> Future<T> submit(Callable<T> task) {
 30     return schedule(task, 0, NANOSECONDS);
 31 }
 32 
 33 /**----------------------schedule实现------------------------**/
 34 /**
 35  * @throws RejectedExecutionException 
 36  * @throws NullPointerException      
 37  */
 38 public ScheduledFuture<?> schedule(Runnable command,
 39                                    long delay,
 40                                    TimeUnit unit) {
 41     if (command == null || unit == null)
 42         throw new NullPointerException();
 43     //这里调用decorateTask方法是留给子类自定义RunnableScheduledFuture实现
 44     //默认直接返回ScheduledFutureTask实例
 45     RunnableScheduledFuture<?> t = decorateTask(command, new ScheduledFutureTask<Void>(command, null, triggerTime(delay, unit)));
 46     delayedExecute(t);
 47     return t;
 48 }
 49 
 50 /**
 51  * @throws RejectedExecutionException 
 52  * @throws NullPointerException      
 53  */
 54 public <V> ScheduledFuture<V> schedule(Callable<V> callable,
 55                                        long delay,
 56                                        TimeUnit unit) {
 57     if (callable == null || unit == null)
 58         throw new NullPointerException();
 59     //这里调用decorateTask方法是留给子类自定义RunnableScheduledFuture实现
 60     //默认直接返回ScheduledFutureTask实例    
 61     RunnableScheduledFuture<V> t = decorateTask(callable, new ScheduledFutureTask<V>(callable, triggerTime(delay, unit)));
 62     delayedExecute(t);
 63     return t;
 64 }
 65 
 66 /*********************周期任务的提交***************************/
 67 
 68 /**
 69  * 提交固定频率的周期任务，
 70  * 任务的执行时间分别是：initialDelay、initialDelay+period、initialDelay+2*period、initialDelay+3*period......
 71  * @throws RejectedExecutionException 
 72  * @throws NullPointerException       
 73  * @throws IllegalArgumentException   
 74  */
 75 public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
 76                                               long initialDelay,
 77                                               long period,
 78                                               TimeUnit unit) {
 79     if (command == null || unit == null)
 80         throw new NullPointerException();
 81     if (period <= 0)
 82         throw new IllegalArgumentException();
 83     ScheduledFutureTask<Void> sft =
 84         new ScheduledFutureTask<Void>(command,
 85                                       null,
 86                                       triggerTime(initialDelay, unit),
 87                                       unit.toNanos(period));
 88     //这里调用decorateTask方法是留给子类自定义RunnableScheduledFuture实现
 89     //默认直接返回ScheduledFutureTask实例                                      
 90     RunnableScheduledFuture<Void> t = decorateTask(command, sft);
 91     sft.outerTask = t;
 92     delayedExecute(t);
 93     return t;
 94 }
 95 
 96 /**
 97  * 提交固定延迟的周期任务，
 98  * 即每一次周期任务都是上一次执行结束再延迟delay。
 99  * @throws RejectedExecutionException 
100  * @throws NullPointerException       
101  * @throws IllegalArgumentException 
102  */
103 public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
104                                                  long initialDelay,
105                                                  long delay,
106                                                  TimeUnit unit) {
107     if (command == null || unit == null)
108         throw new NullPointerException();
109     if (delay <= 0)
110         throw new IllegalArgumentException();
111     ScheduledFutureTask<Void> sft =
112         new ScheduledFutureTask<Void>(command,
113                                       null,
114                                       triggerTime(initialDelay, unit),
115                                       unit.toNanos(-delay));
116     //这里调用decorateTask方法是留给子类自定义RunnableScheduledFuture实现
117     //默认直接返回ScheduledFutureTask实例                                      
118     RunnableScheduledFuture<Void> t = decorateTask(command, sft);
119     sft.outerTask = t;
120     delayedExecute(t);
121     return t;
122 }
123 
124 //入队一个任务，除非当前线程池状态不能接受该任务。
125 //与reExecutePeriodic不同的地方，若不接受任务，这里是走拒绝策略
126 private void delayedExecute(RunnableScheduledFuture<?> task) {
127     if (isShutdown()) //线程池状态至少是SHUTDOWN
128         reject(task); //走拒绝策略
129     else {
130         super.getQueue().add(task); //将任务加入队列
131         //线程池状态二次检查，如果不接受新任务，从任务队列中移除任务
132         if (isShutdown() &&
133             !canRunInCurrentRunState(task.isPeriodic()) &&
134             remove(task))
135             task.cancel(false); //执行ScheduledFutureTask的取消任务
136         else        //否则确保线程池中至少启动有一个线程
137             ensurePrestart();
138     }
139 }

View Code

对于普通非延迟非周期任务通过execute、submit方法提交，这是重写了AbstractExecutorService的方法，统一由schedule方法调度执行，通过调用schedule时传递的delay参数可见，对于这类非延迟非周期的常规任务，延迟时间都是0，表示立即执行。

对于非周期延迟任务通过schedule方法实现，针对Callable、Runnable提供了两个重载方法。对于周期任务通过scheduleAtFixedRate、scheduleWithFixedDelay提交，scheduleAtFixedRate与scheduleWithFixedDelay的区别在线程池概述中已经交代了，简单来说，初始延迟为initialDelay，周期为了period的任务通过scheduleAtFixedRate提交的话，那么其以后的执行时间分别是：initialDelay、initialDelay + period、initialDelay + 2*period、initialDelay + 3*period 以此类推；而通过scheduleWithFixedDelay提交的话，那么每一次任务的执行都是根据上一次执行结束之后间隔period后开始执行。就算某一次的执行时间太长，以至于到达了通过scheduleAtFixedRate提交的任务的下一次的执行时间，线程池也不会让同一个任务同时执行的，因为任务的每一次执行都回将执行它的线程设置到它的runner字段中，一旦发现该字段不为null，就表示有线程在执行，从而跳过执行。

上面的任务提交过程中，都调用了一个decorateTask方法，该方法默认什么也不做，原样返回创建的ScheduledFutureTask实例，这是留给子类重写decorateTask可以创建自定义的RunnableScheduledFuture实现类。创建ScheduledFutureTask实例的时候会根据延迟时间用triggerTime方法计算出任务第一次被调度执行的剩余延迟时间。

从提交任务的源码可以发现，它们的真正核心代码是调用delayedExecute方法实现的，该方法的逻辑很简单：

如果线程池已经shutdown，则走拒绝策略；否则将任务加入任务队列，然后再次确认线程池的状态，如果线程池已经不接受任务的话需要将任务从队列中移除，并取消任务；成功加入队列然后创建一个核心线程，除非线程数量达到了corePoolSize限制。

任务的提交总的来说就是创建任务实体ScheduledFutureTask，并在线程池处于活动情况下加入到任务队列。

任务的执行

通过上一章ThreadPoolExecutor的分析，我们知道任务的执行，其实就是由线程池调度执行任务的run方法，这里就是ScheduledFutureTask的run方法。

 1 //重写 FutureTask的run方法，是为了将需要周期执行的任务重置/再次入队。
 2 public void run() {
 3     boolean periodic = isPeriodic();
 4     //如果线程池的状态至少是STOP，或者是SHUTDOWN但是参数指定不在shutdown状态继续执行任务，则取消任务
 5     if (!canRunInCurrentRunState(periodic))
 6         cancel(false);
 7     //不是周期任务，直接执行FutureTask.run
 8     else if (!periodic)        
 9         ScheduledFutureTask.super.run();
10     //是周期执行的任务，
11     else if (ScheduledFutureTask.super.runAndReset()) { //任务正常结束
12         setNextRunTime();    //设置周期任务的下次执行时间
13         reExecutePeriodic(outerTask); //将当前任务再次加入线程池队列
14     }
15 }
16 
17 //设置周期任务下一次的运行时间。 
18 private void setNextRunTime() {
19     long p = period;
20     if (p > 0) //固定速率 fixed-rate：time,time + p,time +2*p,time + 3*p......
21         time += p;
22     else       //固定延迟 fixed-delay: now + |p| 即当前系统时间 加周期时间
23         time = triggerTime(-p); //这是p是负数，所以取反传递的是正数
24 }
25 
26 
27 //确定是否可以在当前线程池的状态继续执行周期/延迟任务
28 boolean canRunInCurrentRunState(boolean periodic) {
29     return isRunningOrShutdown(periodic ?
30                                continueExistingPeriodicTasksAfterShutdown :
31                                executeExistingDelayedTasksAfterShutdown);
32 }
33 
34 //ScheduledThreadPoolExecutor需要的状态检查，以便在线程池shutdown期间继续执行周期/延迟任务
35 //一旦线程池的状态超过SHUTDOWN，即STOP，TIDYING，TERMINATED则无论如何都返回false。
36 final boolean isRunningOrShutdown(boolean shutdownOK) {
37     int rs = runStateOf(ctl.get());
38     return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
39 }
40 
41 //重新人队一个周期性执行的任务，除非当前线程池状态不能接受该任务。
42 //与delayedExecute的思想相同，不同的是如果不接受任务，这里是直接删除任务，而delayedExecute是拒绝任务
43 void reExecutePeriodic(RunnableScheduledFuture<?> task) {
44     //如果线程池的当前状态或者相关参数指示可以继续接收任务
45     if (canRunInCurrentRunState(true)) {
46         super.getQueue().add(task);    //将任务加入队列
47         //线程池状态二次检查，如果不接受任务，从任务队列中移除任务
48         if (!canRunInCurrentRunState(true) && remove(task))
49             task.cancel(false); //执行ScheduledFutureTask的取消任务
50         else    //否则确保线程池中至少启动有一个线程
51             ensurePrestart();
52     }
53 }
54 
55 
56 /**
57  * 返回一个延迟任务的下一次触发时间
58  */
59 private long triggerTime(long delay, TimeUnit unit) {
60     //转换成纳秒调用triggerTime(long)
61     return triggerTime(unit.toNanos((delay < 0) ? 0 : delay)); 
62 }
63 
64 //返回一个延迟任务的下一次触发时间,delay是纳秒
65 long triggerTime(long delay) {
66     return now() +
67         ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
68 }
69 
70 //限制队列中的所有任务的延迟都在Long.MAX_VALUE范围内，避免 compareTo 方法溢出。
71 //如果一个任务满足出队的条件，但是还没有出队，并且添加了其它一些延迟为Long.MAX_VALUE的任务，则会发生这种情况
72 private long overflowFree(long delay) {
73     Delayed head = (Delayed) super.getQueue().peek(); //拿到当前队列中剩余延迟最小的
74     if (head != null) { //队列不为空
75         long headDelay = head.getDelay(NANOSECONDS);
76         //这里delay是一个正数，为什么减一个负数还小于0呢，只能是溢出了Long.MAX_VALUE范围
77         if (headDelay < 0 && (delay - headDelay < 0))
78             delay = Long.MAX_VALUE + headDelay; //实际上就是 Long.MAX_VALUE 减了一个正数，因此保证delay小于Long.MAX_VALUE
79     }
80     return delay;
81 }

View Code

这里的run、setNextRunTime都是ScheduledFutureTask的方法，其它方法都是ScheduledThreadPoolExecutor的方法，为了方法阅读将它们全部贴出来了。ScheduledFutureTask的run方法也很简单：

如果线程池的状态不能安排任务执行了，则调用cancel(false)取消任务。

如果不是周期任务，即延迟任务（包括延迟为0的普通任务），直接调用FutureTask的run方法。该方法在线程池概述中已经分析了，就不再详述，它会执行任务并设置任务的结果。

如果是周期执行的任务，则执行FutureTask的runAndReset方法，该方法不设置任务的执行结果，因为周期任务本来就没有返回结果一说，如果任务执行过程中没有抛出异常，则通过setNextRunTime设置任务下一次执行的延迟剩余时间，然后通过reExecutePeriodic再次将任务加入任务队列，当然如果线程池此时已经shutdown即不能安排任务执行的话，只有取消任务了，往后的周期也就不会再执行了。

通过任务被调度的执行过程，可见非周期任务的执行与ThreadPoolExecutor一样都是调用FutureTask的run方法，因此可以通过提交任务时返回的Future获取异步任务的结果，而周期任务是通过不设置执行结果的runAndReset方法执行的，因此并没有结果返回，如果通过提交周期任务返回的ScheduledFuture调用get想获取结果将会永久阻塞。

另外对于周期任务，如果任务的执行过程中抛出了异常，将取消以后的周期执行，因此提交的周期任务的run方法中应该将所有的异常都捕获，否则一旦抛出异常，周期任务就停止了。

任务的取消

cancel方法也是由ScheduledFutureTask中实现的，其是Future的接口方法，为了方便任务提交者可以通过返回的Future实例取消任务的执行：

1 //取消任务，重写FutureTask是为了实现实时将被取消的任务从工作队列中移除
2 public boolean cancel(boolean mayInterruptIfRunning) {
3     //直接使用FutureTask的cancel
4     boolean cancelled = super.cancel(mayInterruptIfRunning);
5     //取消任务之后，如果removeOnCancel为true，并且任务还存在延迟队列中(heapIndex >=0)
6     if (cancelled && removeOnCancel && heapIndex >= 0)
7         remove(this);    //将被取消的任务从工作队列中移除
8     return cancelled;
9 }

View Code

可见，任务的取消实际还是调用的FutureTask的cancel实现的，ScheduledFutureTask重写该方法是为了实现取消任务的同时立即将任务从工作队列移除的特性。在ScheduledThreadPoolExecutor中调用cancel的时候传入的mayInterruptIfRunning都是false，因此这种线程池在取消任务时并不会中断已经处于执行中的任务。

对于非周期任务，如果任务还没被从任务队列取走执行，那么取消之后，任务将不会执行，如果任务已经在执行，则取消不会成功，更不会中断任务的执行。

对于周期任务，如果任务还没被从任务队列取走执行，那么取消之后，任务也将不会执行，如果任务已经在执行，则取消之后，由于其状态变成了CANCELLED，在runAndReset方法返回之前，发现任务的状态不是NEW，因此返回false，从而不再安排任务的下一次执行。

其它方法---onShutdown

在ThreadPoolExecutor的shutdown的方法中，线程池的状态被改变的SHUTDOWN之后，会回调一个钩子函数onShutdown，该方法在ThreadPoolExecutor中什么也没做，而在ScheduledThreadPoolExecutor中有了重写：

 1 //取消并清除由于关闭策略而不应该运行的任务队列中的所有任务。该方法在shutdown是回调
 2 @Override void onShutdown() {
 3     BlockingQueue<Runnable> q = super.getQueue();
 4     //延迟任务是否继续存活
 5     boolean keepDelayed =
 6         getExecuteExistingDelayedTasksAfterShutdownPolicy();
 7     //周期任务是否继续存活
 8     boolean keepPeriodic =
 9         getContinueExistingPeriodicTasksAfterShutdownPolicy();
10         
11     //两者都不需要继续存活，则取消并清空所有任务
12     if (!keepDelayed && !keepPeriodic) {
13         for (Object e : q.toArray())
14             if (e instanceof RunnableScheduledFuture<?>)
15                 ((RunnableScheduledFuture<?>) e).cancel(false);
16         q.clear();
17     }
18     else {
19         // 遍历快照以避免迭代器异常
20         for (Object e : q.toArray()) {
21             if (e instanceof RunnableScheduledFuture) {
22                 RunnableScheduledFuture<?> t =
23                     (RunnableScheduledFuture<?>)e;
24                 if ((t.isPeriodic() ? !keepPeriodic : !keepDelayed) ||
25                     t.isCancelled()) { // 如果已经取消依然从队列中移除任务
26                     if (q.remove(t))
27                         t.cancel(false); //没有取消的，则取消任务，但不中断正在执行的任务。
28                 }
29             }
30         }
31     }
32     tryTerminate(); //尝试看能不能进一步将线程池的状态转到TIDYING、TERMINATED。
33 }

View Code

可见，在线程池状态转到SHUTDOWN之后，在调用tryTerminate使线程池走向真正终结之前，ScheduledThreadPoolExecutor会确保将已经入队的任务取消并移除。

总结

总的来说，ScheduledThreadPoolExecutor线程池的实现是在ThreadPoolExecutor的基础上进行的扩展，并借助了一个平衡二叉堆实现的延迟队列作为任务队列，因为如果理解了ThreadPoolExecutor和前面的Java同步数据结构之DelayQueue/DelayedWorkQueue的话，理解起来将没有任何困难。

ScheduledThreadPoolExecutor就是一种支持延迟和周期任务的线程池，当然它也支持提交普通的任务，把他当成一个ThreadPoolExecutor来使用，不过一般来说还是专门用于延迟和周期任务。需要注意的是，周期任务不支持返回结果，因此不要调用提交任务时返回的Future的get方法，而且周期任务在执行任务体时，如果抛出异常将导致周期性结束，之后不再会被执行，因此必要时需要将任务执行体的异常全部捕获。