深入解析ScheduledThreadPoolExecutor并发执行机制源代码

ScheduledThreadPoolExecutor是一个可以在指定一定延时时间后或者定时进行任务调度的线程池&＃xff0c;ScheduledThreadPoolExecutor继承了ThreadPoolExecutor并实现了ScheduledExecutorService接口。线程池的队列是DelayedWorkQueue(他是ScheduledThreadPoolExecutor的一个内部类)。

还要在看一下ScheduledFutureTask(同样是ScheduledThreadPoolExecutor的一个内部类)&＃xff0c;继承FutureTask&＃xff0c;FutureTask的内部有一个变量state用来表示任务的状态&＃xff0c;一开始状态为NEW。下面是所有状态定义。

private volatile int state;private static final int NEW &＃61; 0;//初始化状态private static final int COMPLETING &＃61; 1;//执行中状态private static final int NORMAL &＃61; 2;//正常运行结束状态private static final int EXCEPTIONAL &＃61; 3;//运行中异常private static final int CANCELLED &＃61; 4;//任务被取消private static final int INTERRUPTING &＃61; 5;//任务正在被中段private static final int INTERRUPTED &＃61; 6;//任务已经被中断


ScheduledFutureTask内部还有一个变量period用来表示任务的类型&＃xff0c;如果&＃61;&＃61;0&＃xff0c;则表示任务是一次性的&＃xff0c;任务执行完毕后退出&＃xff0c;如果为负数&＃xff0c;说明当前任务是以固定延时的定时可重复执行任务&＃xff0c;如果为正数&＃xff0c;说明任务是以固定频率的定时定时可重复执行任务。

一、chedule(Runnable command, long delay, TimeUnit unit)方法解析

他的作用是提交一个延时执行的任务&＃xff0c;任务从提交时间算起延时单位为unit的delay时间后开始执行&＃xff0c;任务只会执行一次。

public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {//判断参数是否为空&＃xff0c;空则跑出异常if (command &＃61;&＃61; null || unit &＃61;&＃61; null)throw new NullPointerException();//任务转换RunnableScheduledFuture<?> t &＃61; decorateTask(command,new ScheduledFutureTask<Void>(command, null,triggerTime(delay, unit)));//添加任务到延时队列delayedExecute(t);return t;}


首先是参数判断&＃xff0c;为空则抛出异常&＃xff0c;接着是装饰任务&＃xff0c;把提交的command(Runnable对象)转换为ScheduledFutureTask&＃xff0c;ScheduledFutureTask是具体放入延时队列里面的东西&＃xff0c;由于是延时任务&＃xff0c;所以ScheduledFutureTask实现了getDelay和compareTo方法&＃xff0c;triggerTime方法将延时时间转换为绝对时间&＃xff0c;也就是把当前时间的纳秒加上延迟的纳秒后的值。

ScheduledFutureTask的构造如下&＃xff0c;设period为0&＃xff0c;表示该任务是一次性任务。

ScheduledFutureTask(Callable<V> callable, long ns) {super(callable);this.time &＃61; ns;this.period &＃61; 0;this.sequenceNumber &＃61; sequencer.getAndIncrement();}


然后通过delayedExecute将任务添加到延时队列。

private void delayedExecute(RunnableScheduledFuture<?> task) {//如果线程池管理&＃xff0c;则执行线程拒绝策略if (isShutdown())reject(task);else {//添加任务到延时队列super.getQueue().add(task);//再次检查线程池状态if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))task.cancel(false);else//确保至少一个线程在处理任务ensurePrestart();}}


上述代码首先确保线程池没关闭&＃xff0c;关闭则执行拒绝策略&＃xff0c;没关闭将任务添加到延时队列&＃xff0c;添加后再重新检查线程池是否关闭&＃xff0c;如果关闭则从延时队列里面删除刚才添加的任务。

再看ensurePrestart方法。

void ensurePrestart() {int wc &＃61; workerCountOf(ctl.get());//增加核心线程数if (wc < corePoolSize)addWorker(null, true);else if (wc &＃61;&＃61; 0)addWorker(null, false);}


首先获取到了线程池中的线程数&＃xff0c;如果个数小于核心线程池则新增一个线程&＃xff0c;否则如果当前线程数为0个&＃xff0c;则同样新增一个线程。

我们知道ThreadPoolExecutor在具体执行任务的线程是Worker线程&＃xff0c;Worker线程调用具体任务的run方法来执行&＃xff0c;在这里的任务是ScheduledFutureTask&＃xff0c;所以来看看ScheduledFutureTask的run方法。

public void run() {//是否执行一次boolean periodic &＃61; isPeriodic();//取消任务if (!canRunInCurrentRunState(periodic))cancel(false);else if (!periodic)ScheduledFutureTask.super.run();//定时任务else if (ScheduledFutureTask.super.runAndReset()) {setNextRunTime();reExecutePeriodic(outerTask);}
}


首先判断任务是一次性的还是可重复执行的任务&＃xff0c;ScheduledFutureTask在构造方法中已经设置了period为0&＃xff0c;所以&＃xff0c;这里会返回false。

public boolean isPeriodic() {return period !&＃61; 0;}


然后判断当前任务是否应该被取消。为true则取消任务。

由于periodic为false&＃xff0c;则会执行代码ScheduledFutureTask.super.run();&＃xff0c;调用父类FutureTask的run方法。

public void run() {if (state !&＃61; NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return;try {Callable<V> c &＃61; callable;if (c !&＃61; null && state &＃61;&＃61; NEW) {V result;boolean ran;try {result &＃61; c.call();ran &＃61; true;} catch (Throwable ex) {result &＃61; null;ran &＃61; false;setException(ex);}if (ran)set(result);}} finally {runner &＃61; null;int s &＃61; state;if (s >&＃61; INTERRUPTING)handlePossibleCancellationInterrupt(s);}}


FutureTask.run()首先判断任务状态&＃xff0c;如果不是NEW则直接返回&＃xff0c;或者如果任务状态为NEW&＃xff0c;但是使用CAS设置当前任务的持有者为当前线程失败则直接返回。

然后具体调用callable的call方法执行任务&＃xff0c;如果任务执行成功则修改任务状态&＃xff0c;也就是set方法。

protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome &＃61; v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}
}


使用CAS将当前任务的状态从NEW转换的COMPLETING。这里当有多个线程调用时只有一个线程会成功&＃xff0c;成功的线程在通过 UNSAFE.putOrderedInt设置任务的状态为正常结束状态。

还有在任务执行失败后&＃xff0c;执行setException方法&＃xff0c;和set方法类似了。

protected void setException(Throwable t) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome &＃61; t;UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final statefinishCompletion();}}


二、scheduleWithFixedDelay(Runnable command,long initialDelay, long delay,TimeUnit unit)方法解析

他的作用是&＃xff0c;当任务执行完毕后&＃xff0c;让其延迟固定时间后再次运行&＃xff0c;initialDelay表示提交任务后延迟多少时间开始执行任务command&＃xff0c;delay表示当任务执行完毕后延长多少时间后再次运行command&＃xff0c;unit是时间单位。

这个任务会一直重复运行下去&＃xff0c;直到任务中抛出异常、被取消、线程池关闭。

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit) {//参数判断if (command &＃61;&＃61; null || unit &＃61;&＃61; null)throw new NullPointerException();if (delay <&＃61; 0)throw new IllegalArgumentException();//任务转换ScheduledFutureTask<Void> sft &＃61;new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(-delay));RunnableScheduledFuture<Void> t &＃61; decorateTask(command, sft);sft.outerTask &＃61; t;//添加任务到队列delayedExecute(t);return t;}


首先也是参数判断&＃xff0c;为空则抛出异常&＃xff0c;然后将command任务转换为ScheduledFutureTask&＃xff0c;然后添加延迟到队列。

将任务添加到队列后线程池线程会从队列中获取任务&＃xff0c;然后调用ScheduledFutureTask的run方法&＃xff0c;由于这里period<0&＃xff0c;所以isPeriodic返回true&＃xff0c;则会执行方法runAndReset()。

protected boolean runAndReset() {if (state !&＃61; NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return false;boolean ran &＃61; false;int s &＃61; state;try {Callable<V> c &＃61; callable;if (c !&＃61; null && s &＃61;&＃61; NEW) {try {c.call(); ran &＃61; true;} catch (Throwable ex) {setException(ex);}}} finally {runner &＃61; null;s &＃61; state;if (s >&＃61; INTERRUPTING)handlePossibleCancellationInterrupt(s);}return ran && s &＃61;&＃61; NEW;
}


他在任务执行完毕后不会设置任务的状态&＃xff0c;是为了让任务可重复执行&＃xff0c;看最后一句&＃xff0c;判断如果当前任务正常执行完毕并且任务状态为NEW则返回true&＃xff0c;如果返回true则执行方法setNextRunTime()&＃xff0c;用于设置任务下一次的执行时间。

这里p是<0的&＃xff0c;然后设置timer为当前时间加上-p&＃xff0c;也就是延迟-p时间后再次执行。

private void setNextRunTime() {long p &＃61; period;if (p > 0)time &＃43;&＃61; p;elsetime &＃61; triggerTime(-p);}


三、scheduleAtFixedRate(Runnable command, long initialDelay, long period,TimeUnit unit)方法解析

该方法相对起始时间点以固定频率调用指定任务&＃xff0c;当把任务提交到线程池并延迟initialDelay时间后开始执行任务command&＃xff0c;然后从initialDelay&＃43;period时间点再次执行&＃xff0c;而后在initialDelay&＃43;2*period时间点再次执行&＃xff0c;直到抛出异常或者取消、关闭线程池。

原理和scheduleWithFixedDelay类似&＃xff0c;我们就看几个不同点&＃xff0c;

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


首先是period&＃61;period&＃xff0c;不再是-period。所以当前任务执行完毕后调用setNextRunTime设置任务下次执行的时间是 time &＃43;&＃61; p。

private void setNextRunTime() {long p &＃61; period;if (p > 0)time &＃43;&＃61; p;elsetime &＃61; triggerTime(-p);}


如果当前任务还没有执行完&＃xff0c;下一次执行任务的时间到了&＃xff0c;则不会并发执行&＃xff0c;下次要执行的任务会延迟&＃xff0c;要等到当前任务执行完毕后再次执行