《Java并发编程之美》读书笔记
第8章 Java并发包中线程池ThreadPoolExceutor原理探究
介绍
线程池主要解决两个问题：
- 一是当执行大量异步任务时线程池能提供较好的性能，在不适用线程池时，每当需要执行异步任务时直接new一个线程来运行，而线程的创建和销毁都是需要开销的。线程池里面的线程是可以复用的，不需要每次执行异步任务的时候都重新创建和销毁线程。
- 线程池提供了一种资源限制和管理的手段，比如可以限制线程的个数，动态增减线程的等，ThreadPoolExceutor也保留了一些基本的统计参数，比如当前线程池完成的任务数目等
另外，线程池也提供了许多可调的参数个可扩展性接口，以满足不同情况的需要，程序员可以使用更方便的Executors的工程方法，比如newCachedThreadPool(线程池线程个数最多可达Integer.MAX_VALUE，线程自动回收)newFixedThreadPool(固定大小的线程池)和newSingleThreadExecutor（单个线程）等来创建线程池，当然用户还可以自定义
类图介绍
在类图中，Executors其实是个工具类，里面提供了好多静态方法，这些方法更具用户选择返回不同的线程池实例。ThreadPoolExceutor继承了AbstractExecutorService,成员变量ctl是一个Integer的原子变量，用来记录线程池状态和线程池中的线程个数，类似于ReentrantReadWriteLock使用一个变量来保存两种信息。
这里假设Integer类型是32位二进制表示，则其中高3位用来表示线程池状态，后面29位用来记录线程池的线程个数
//原子变量ctl高3位用爱表示线程池状态，低29位用来表示线程个数
//默认RUNNING状态，线程个数为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程个数掩码位数，并不是所有平台的int类型都是32位的，所有的来说，是具体平台下的Integer的二进制位数-3的剩余位数所表示的数才是线程的个数
private static final int COUNT_BITS = Integer.SIZE - 3;
//线程的最大个数（低29位）00011111111111111111111；
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1；
// 线程池的状态
//高3位 运行态111000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
//高3位 000000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//高3位 001
private static final int STOP = 1 << COUNT_BITS;
//高3位 010
privatestatic final int TIDYING = 2 << COUNT_BITS;
//高3位 011
private static final int TERMINATED = 3 << COUNT_BITS;
// 获取高3位
private static int runStateOf(int c) { return c & ~COUNT_MASK; }
//获取低29位（线程个数）
private static int workerCountOf(int c) { return c & COUNT_MASK; }
//计算ctl新值（线程状态与线程个数）
private static int ctlOf(int rs, int wc) { return rs | wc; }
- RUNNING：接受新任务并且处理阻塞队列里面的任务
- SHUTDOWN:拒绝新任务但是处理阻塞队列里的任务
- STOP:拒绝新任务并且抛弃阻塞队列里的任务，同时会中断正在处理的任务
- TIDYING:所有任务都执行完（包含阻塞队列里面的任务）后当前线程池活动线程为0，将要调用terminated方法
- TERMINATED：终止状态，terminated方法调用完成以后的状态
- RUNNING->SHUTDOWN：显示调用shutdown（）方法，或者隐士调用了finalize（）方法里面的shutdown方法；
- RUNNING or SHUTDOWN ->STOP：显示调用shutdownNow()方法时。
- SHUTDOWN->tIDYING:当线程池和任务队列都为空的时候
- STOP->TIDYING:当线程池为空的时候
- TIDYING->TERMINATED:当terminated（）hook方法完成时。
线程池的参数如下： - corePoolSize:线程池核心线程的个数
- workQueue：用于保存等待执行的任务的阻塞队列，比如基于数组的有界的ArrayBlockingQueue、基于链表的无界LinkedBlockingQueue、最多只有一个元素的同步队列SynchronousQueue及优先级队列PriorityBlockingQueue等
- maximumPoolSize:线程池最大的线程数量
- ThreadFactory:创造线程的工厂
- RejectedExecutionHandler:饱和策略，当队列满并且线程个数达到maximunPollSize后采取的策略，比如AbortPolicy(抛出异常)，CallerRunsPolicy(使用调用者线程来运行任务)、DiscardOldestPolicy(调用poll丢弃一个任务，执行当前任务)及DiscardPolicy（默默丢弃，不抛出异常）
- keeyAliveTime：存活时间。如果当前线程池中的线程数量比核心线程数量多，并且是闲置状态，则这些闲置的线程能存活的最大时间。
- TimeUnit：存活时间的时间单位
- newFixedThreadPool:创建一个核心线程个数和最大线程个数都为nThreads的线程池，并且阻塞队列长度为Integer.MAX_VALUE。 keepAliveTime=0说明只要线程个数比核心线程多并且当前空闲则回收。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//使用自定义线程创建工厂
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
- newSingleThreadExecutor:创建一个核心线程个数和最大线程个数都为1的线程池，并且阻塞队列的长度为Integer.NAX_VALUE.keepAliveTime=0说明只要线程个数比核心个数多并且当前空闲则回收。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//使用自己的线程工厂
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
}
- newCachedTheadPool：创建一个按需创建线程的线程池，出事的线程个数为0，最多线程的个数为Integer.MAX_VALUE,并且阻塞队列为同步队列。KeepAliveTime=60,表示当前线程在60s内空闲则回收。这个类型的特殊之处在于，加入同步队列的任务会被马上执行，同步队列里面最多只有一个任务。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//使用自定义的线程工厂
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
}
如上的TheadPoolExecutor类图所示，其中mainLock是独占锁，用来控制新增Worker线程操作的原子性，termination是该锁对应的条件队列，在线程调用awaitTermination时用来存放阻塞的线程。
Worker继承AQS和Runnable接口，是具体承载任务的对象，worker继承了AQS，自己实现了简单不可重入独占锁，其中state=0表示锁未被获取状态，state=1表示所以就备货区，state=-1是创建worker默认的状态，创建时状态设置为-1是为了避免该线程在运行runWorker方法前被中断，其中变量firstTask记录该工作线程执行的第一个任务，thread是具体执行任务的线程。
DefaultThreadFactory是线程工厂，newThread方法时对线程的一个修饰，其中poolNumber是个静态的原子变量，用来统计线程工程的个数，threadNumber用来记录每个线程工厂创建了多少的线程，这两个值也作为线程池和线程的名称的一部分。
源码分析
public void execute(Runnable command)
execute方法的作用是提交任务command到线程池进行执行，用户线程提交任务到线程池的模型图如下图所示：
从该图可以看出，ThreadPoolExecutor的实现实际上是一个生产消费模型，当用户添加任务到线程池到相当于生产者生产元素，workers线程工作集中的线程直接执行任务或者从任务队列里面获取任务时则相当于消费者消费元素。
public void execute(Runnable command) {
//1.如果任务为null，则抛出NPE异常
if (command == null)
throw new NullPointerException();
//2获取当前线程池的状态+线程个数变量的组合值
int c = ctl.get();
//3当前线程池中的线程个数是否小于corePoolSize，小于的话则开启线程运行
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//4如果线程池处于Running状态，则添加任务到阻塞队列
if (isRunning(c) && workQueue.offer(command)) {
//4，1二次检查
int recheck = ctl.get();
//4.2如果当前的线程池状态不是RUNNING则从队列中删除任务，并且执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
//4.3否则如果当前线程池为空，则添加一个线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//5如果队列满，则新增线程，新增失败则执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
代码3判断如果当前线程池中线程个数小于corePoolSize，会向workers里面新增一个核心线程（core线程）执行该任务
如果当前线程池中线程个数大于等于corePoolSize则执行代码4.如果当前线程池处于RUNNING状态则添加当前任务到任务队列。这里需要判断线程池状态是因为线程池已经处于非RUNNING，而非RUNNING状态下是要抛弃新任务的。
如果向任务队列添加任务成功，则代码4.2对线程池状态进行二次校验，这是因为添加任务到任务队列后，执行代码4.2之前有可能线程池的状态已经发生了变化了，这里进行二次检验，如果当前线程池状态不是RUNNINGLE则把任务从任务队列里面移除，移除后执行拒绝策略；如果二次校验通过，则执行待吗4.3重新判断当前线程池里面是否还有线程，如果没有则新增一个线程。
如果代码4添加任务失败，则说明任务队列已满，那么执行代码5尝试新开启线程如图中中thread3，thread4来执行该任务，如果当前线程池中线程个数>maximunPoolSize则执行拒绝策略。
下面分析新增线程addWorker方法：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
// 6检查队列是否只在必要时为空
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
//7 循环CAS增加线程个数
for (;;) {
//7.1如果线程个数则返回false
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
//7.2CAS增加线程个数，同时只有一个线程成功
if (compareAndIncrementWorkerCount(c))
break retry;
//7.3到这里说明CAS失败了，则看线程池状态是否变化了，变化则调到外层循环重新尝试获取线程池状态，否则内层循环重新CAS
c = ctl.get(); // Re-read ctl
if (runStateAtLeast(c, SHUTDOWN))
continue retry;
}
}
//8到这里说明CAS成功了
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//8.1创建worker
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//8.2加独占锁，为了实现workers同步，因为可能多个线程调用了线程池的execute方法
mainLock.lock();
try {
// 重新检查线程池状态，以避免在获取锁调用了shutdown接口
int c = ctl.get();
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//添加任务
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//添加成功后则启动任务
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
代码比较长主要分为两个部分：
- 第一部分双重循环的目的是通过CAS操作添加线程数
- 第二部分主要是把并发安全的任务添加到workers里面，并且启动任务执行**
首先来分析第一部分代码
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)//1
|| firstTask != null//2
|| workQueue.isEmpty()))//3
return false;
也就是说代码6在下面几种情况下会返回false
- 当前线程池状态为STOP TIDYING TERMINATED
- 当前线程池状态为SHUTDOWN并且已经有第一个任务
- 当前线程池状态为SHUTDOWN并且任务队列为空
内层循环的作用是使用CAS操作增加线程数，代码7.1如果线程个数超限则返回false，否则执行代码7.2CAS操作设置线程个数，CAS成功则退出双循环，CAS失败则执行代码7.3看当前线程池的状态是否变化了，如果变了，则再次进入外层循环重新获取线程池状态，否则进入内存循环继续进行CAS尝试。
执行到了第二部分的代码8是说明使用CAS成功的增加了线程个数，但是现在任务还没有开始执行。这里要使用全局的独占锁把新增的Worker添加到工作集workers中。代码8.1创建了一个工作线程Worker。
代码8.2获取了独占锁，代码8.3重新检查线程池状态，这是为了避免在获取锁之前其他线程调用了shutdown关闭了线程池，如果线程池已经被关闭，则释放锁，新增线程失败，否则执行代码8.4天假工作线程到线程工作集，然后释放锁，代码8.5判断如果新增工作线程成功，则启动工作线程。
工作线程Worker的执行
用户线程提交任务到线程池后，由worker来执行。先看下worker的构造函数。
Worker(Runnable firstTask) {
setState(-1); // 在调用runworker前禁止中断
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);//创建一个线程
}
在构造函数内首先设置worker的状态为1，这是为了避免当前worker在调用runworker方法前被中断（当其他线程调用了线程池的shutdownNow时候，如果worker代码中状态>=0则会中断该线程。这里设置线程的状态为-1，所以该线程就不会中断了，咋子如下的runworker代码中，运行代码9时会调用unlock方法，该方法把status设置了为0，所以这时候调用shutDownNow会中断worker线程。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts9将state设置为0，允许中断
boolean completedAbruptly = true;
try {
//10
while (task != null || (task = getTask()) != null) { //10.1
w.lock();
...
try {
//10.2执行任务前干一些事情
beforeExecute(wt, task);
try {
task.run();//10.3执行任务
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
task = null;
//10.5 统计当前worker完成了多少任务
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
//11执行清理工作
processWorkerExit(w, completedAbruptly);
}
}
在如上的代码10中，如果当前task==null或者调用getTask()从任务队列获取的任务返回null，则跳转到代码11执行，如果task不为null则执行任务10.1获取线程内部持有的独占锁，然后执行扩展接口代码10.2在具体任务之前做些事情，代码10.3具体执行任务，代码10。5统计当前worker完成了多少个任务，并释放锁。
这里在执行具体的任务期间加锁，是为了避免在任务运行的期间，其他线程调用了shutdown后正在执行的任务被中断（shutdown只会中断当前被阻塞挂起的线程）
代码11执行清理任务，其代码如下：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
//11.1统计整个线程池完成的任务个数，并从工作集里面删除当前的worker
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
//11.2尝试设置线程池的状态为Terminated,如果当前是shutdown状态并工作队列为空
//或者当前是stop状态，当前线程池里没有活动线程。
tryTerminate()；
int c = ctl.get();
//11.3如果当前线程个数小于核心个数，则增加
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
在如上的代码中，代码11.1统计线程池完成任务的个数，并且在统计前加了全局所，把在当前工作线程中完成的任务累加到全局计数器，然后从工作集中删除当前的worker。
代码11.2判断如果线程池的状态是SHUTDOWN并且工作队列为空，或者当前线程池状态是STOP并且当前线程池里面没有活动线程，则设置线程池状态为TERMINATED。如果设置为了TERMINATED状态，则还需要调用条件变量termination的signalAll()方法激活所有因为调用线程池的awaitTermination方法而被阻塞的线程。
代码11.3则判断当前线程池个数是否小于核心线程个数，如果是则在新增一个线程。
shutdown操作
调用shutdown方法之后，线程系就不会在接受新的任务了，但是工作队列里面的任务还是需要执行的。该方法会立刻返回，并不等待队列任务完成在返回。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//12 权限检查
checkShutdownAccess();
//13 设置当前线程池状态为SHUTDOWN,如果已经是了SHUTDOWN则直接返回。
advanceRunState(SHUTDOWN);
//设置中断标志
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
//15 尝试将状态变为TERMINATED
tryTerminate();
}
在如上的代码中，代码12检查看是否设置了安全管理器，是则看点前调用shutdown命令的线程是否有关闭线程的权限，如果有则还要看调用线程是否有中断线程工作线程的权限，如果没有权限则抛出SecurityException或者NullPointerException
其中代码13的内从如下，如果当前线程池状态>=
SHUTDOWN则直接返回，否则是指为SHUTDOWN状态。
private void advanceRunState(int targetState) {
// assert targetState == SHUTDOWN || targetState == STOP;
for (;;) {
int c = ctl.get();
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
代码14的内容如下，其设置所有空闲线程的中断标志。这里首先加了全局锁，同时只有一个线程可以调用shutdown方法设置中断标志，然后尝试获取worker自己的锁，获取成功则设置中断标志。由于正在执行的任务已经获取了锁，所以正在执行的任务没有被中断。这里中断的是阻塞到getTask()方法并企图从队列里面获取任务的线程，也就是空闲线程。
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
//如果工作线程没有被中断，并且没有正在运行则设置中断标志
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
final void tryTerminate() {
for (;;) {
..
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {//设置当前线程池状态为TIDYING
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
//设置当前线程池状态为TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
//激活因调用条件变量termination的await方法而被阻塞的所有线程
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
}
}
在如上代码中，首先使用CAS设置当前线程池状态为TIDYING，如果设置成功则执行扩展接口terminated在线程池状态变为TERMINATED前做一些事情，然后设置当前线程值得状态为TERMINATED。最后调用termination.signalAll激活因调用条件变量termination的await方法而被阻塞的所有线程。
shutdownNow操作
调用shutdownNow方法后，线程池就不会再接受新的任务了，并且会丢弃工作队列里面的任务，正在执行的任务会被中断，该方法会立刻返回，并不等待激活的任务执行完成。返回值为这时候队列里面被丢弃的任务列表。
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();//16权限检查
advanceRunState(STOP);//17设置线程池的状态为stop
interruptWorkers();//18中断所有线程
tasks = drainQueue();//19将队列任务全部移动到tasks里面
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
在如上的代码中，首先调用代码16检查权限，然后调用代码17设置当前线程池的状态为stop，随后执行代码18中断所有工作线程，这里需要注意的是，中断的所有线程包含空闲线程和正在执行任务的线程。
private void interruptWorkers() {
// assert mainLock.isHeldByCurrentThread();
for (Worker w : workers)
w.interruptIfStarted();
}
然后执行代码19将任务队列里面的任务移动到tasks列表
awaitTermination操作
当线程调用awaitTermination方法后，当前线程会阻塞，直到线程池状态变为Termination才返回，或者等待时间超时才返回。整个过程中独占锁的代码：
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
while (runStateLessThan(ctl.get(), TERMINATED)) {
if (nanos <= 0L)
return false;
nanos = termination.awaitNanos(nanos);
}
return true;
} finally {
mainLock.unlock();
}
}
如上代码首先获取独占锁，然后再无限循环内部判断当前状态池是否至少是Termination状态，如果是则直接返回，否则说明当前线程池还有线程在执行，则看设置的超时时间nanos是否小于0，小于0则说明不需要等待，那就直接返回，如果大于0则调用条件变量termination的awaitNanos方法等待nanos时间，期望在这段时间内线程池的状态变为TERMINATED
在讲shutdown方法提到过，当线程池状态变为TERMINATED的时候，会调用termination.signalAll激活因调用条件变量termination的await方法而被阻塞的所有线程，所以如果在调用awaitTermination方法之后调用shutdown方法，并且在shutdown内部将线程池状态设置为TERMINATED，则termination.awaitNanos方法会返回。
另外在工作线程中worker的runworker方法内，当工作线程运行结束后，会低啊用processWorkerExit方法，在processWorkerExit方法内部也会调用trytREMINATE方法测试当前时候应该把线程池状态设置为TERMINATED，如果是，则也会调用termination.signalAll激活因调用条件变量termination的await方法而被阻塞的所有线程。
而且当等待时间超时后，terminate.awaitNanos也会返回，这时候会重现检查当前线程池状态是否为TERMINATED;如果是则世界返回，否则继续阻塞挂起自己。
总结
线程池巧妙的使用一个Integer类型的原子变量来记录线程池状态和线程池中线程个数，通过线程池状态来控制任务的执行，每个Worker线程可以处理多个任务，线程池通过线程的复用减少了线程的创建和销毁的开销。
Java并发包中ScheduledThreadPoolExecutor原理探究
ThreadPollExecutor只是Executors工具类的一部分功能，而ScheduledThreadPoolExecutor是一个可以在指定一定延迟时间后或者定时进行任务调度执行的线程池。
类图介绍
Executors其实是个工具类，它提供了好多静态方法，可根据用户的选择返回不同的线程池实例。ScheduledThreadPoolExecutor继承了ThreadPoolExecutor并实现了ScheduledExecutorService接口，线程池队列是DelayedWorkQueue,其和DelayedQueue类似，是一个延迟队列。ScheduledFutureTask是具有返回值得任务，继承自FutureTask。FutureTask的内部用一个变量state用来表示任务的状态，一开始状态为new，所有状态为
private volatile int state;
private static final int NEW = 0;//初始状态
private static final int COMPLETING = 1;//执行中状态
private static final int NORMAL = 2;//正常运行结束状态
private static final int EXCEPTIONAL = 3;//运行中异常
private static final int CANCELLED = 4;//任务被取消
private static final int INTERRUPTING = 5;//任务正在被中断
private static final int INTERRUPTED = 6;//任务已经被中断
可能的任务状态的转换路径为
new->completing-normal 初始态->执行中-正常结束
new->completing->exception 初始态->执行中->执行异常
new->cancelled 初始态->任务取消
new->interrupting-interrupted初始状态->被中断中->被中断
ScheduledFutureTask内部还有一个变量period用来表示任务的类型，任务的类型如下：
- period=0说明当前任务时一次性的，执行完毕后就退出了
- period为附属，说明当前任务为fixed-delay任务，是固定延迟的定时可重复执行的任务
- period为正数，说明当前任务为fixed-rate任务，是固定频率的定时可重复执行任务
ScheduledThreadPoolExecutor的一个构造函数如下，由构造函数可知线程池队列是DelayedWorkQueue
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
new DelayedWorkQueue());
}
原理剖析
- schedule(Runnable command,long delay,TimeUnit unit)
- scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit)
- scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit)
schedule(Runnable command,long delay,TimeUnit unit)
这份方法的作用是提交一个延迟执行的任务，任务从提交时间算起延迟单位为unit的delay的时间开始执行。提交的任务不是周期性任务，任务只会执行一次
public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
//1参数校验
if (command == null || unit == null)
throw new NullPointerException();
//2任务转换
RunnableScheduledFuture<Void> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit),
sequencer.getAndIncrement()));
//3添加任务到延迟队列
delayedExecute(t);
return t;
}
1.如上代码1进行参数校验，如果command或者unit为null，则抛出空指针异常
2。代码2执行装饰任务，把提交的command任务转换为ScheduledFutureTask。ScheduledFutureTask是具体放入延迟队列的东西，由于是延迟任务，所以ScheduledFutureTask实现了long getDelay（TimeUnit unit）和int compareTo(Dealyed other)方法。triggerTime方法将延迟时间转为绝对时间，也就是把当前时间的那描述加上延迟的纳秒数的long值，ScheduledFutureTask的构造函数如下。
ScheduledFutureTask(Runnable r, V result, long triggerTime,
long sequenceNumber) {
//调用父类FutureTask构造函数
super(r, result);
this.time = triggerTime;
this.period = 0;//period为0，说明为一次性任务
this.sequenceNumber = sequenceNumber;
}
在构造函数内部首先调用了父类FutureTask的构造函数，父类FutureTask的构造函数的代码如下
//通过适配器将runnable转换为callable
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // 设置当前任务状态为new
}
FutureTask中的任务被转换为Callable类型后，被保存到了变量this.callable里面，并设置FutureTask的任务为NEW
然后再ScheduledFutureTask构造函数内部设置time为上面说的绝对时间，需要注意，这里的period的值为0，这说明当前任务为一次性的任务，不是定时反复执行任务。其中 long getDealy（TimeUnit unit）方法的代码如下（刚方法是用来计算当前任务还有多少时间就过期了）
public long getDelay(TimeUnit unit) {
return unit.convert(time - System.nanoTime(), NANOSECONDS);
}
public int compareTo(Delayed other) {
if (other == this) // compare zero if same object
return 0;
if (other instanceof ScheduledFutureTask) {
ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
long diff = time - x.time;
if (diff < 0)
return -1;
else if (diff > 0)
return 1;
else if (sequenceNumber < x.sequenceNumber)
return -1;
else
return 1;
}
long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
}
compareTo的作用是假如元素到延迟队列后，在内部建立或者调整堆的时候回使用钙元素的compareTo方法与队列里面其他元素进行比较，让最快要过期的元素放到队首。所以无论什么时候往队列里面添加元素，队首的元素都是最快要过期的元素
private void delayedExecute(RunnableScheduledFuture<?> task) {
//4 如果线程池关闭了，则执行线程池拒绝策略
if (isShutdown())
reject(task);
else {
//5添加任务到延迟队列
super.getQueue().add(task);
//6再次检查线程池状态
if (!canRunInCurrentRunState(task) && remove(task))
task.cancel(false);
else
//7确保至少一个线程处理任务
ensurePrestart();
}
}
4.代码4首先判断当前线程池是否已经关闭了，如果已经关闭则中线程池的拒绝策略没否则实行代码5将任务添加到延迟队列。添加完毕后还要重新检查线程池是否被关闭了，如果已经关闭了则从延迟队列里面删除刚才添加的任务，但是此时有可能线程池中的线程已经从任务队列里面移除了该任务，也就是该任务已经在执行了，所以还需要调用任务的cancel方法取消任务
5如果代码6判断的结果为false，则会执行代码7确保至少有一个线程在处理任务，即使核心线程数corePoolSize被设置为0，ensureOrestart的代码如下
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
//增加核心线程数
addWorker(null, true);
//如果初始化corePoolsize==0，则也增加一个线程
else if (wc == 0)
addWorker(null, false);
}
如上代码首先获取线程池中的线程个数，如果线程个数小于核心线程数则核心线程数新增一个线程，否则如果当前线程数为0则新增一个线程。
上面我们分析了如何向延迟队列里面添加任务，接下来我们来见线程池里面的线程如何获取并执行任务，在前面讲解ThreadPoolExecutor时有提及，具体执行任务的线程是worker线程，worker线程调用具体任务的run方法来执行，由于这里的任务是ScheduledFutureTask，所以我们具体来看看ScheduledFutureTask的run方法
public void run() {
//8是否只执行一次
boolean periodic=isPeriodic()
//9取消任务
if (!canRunInCurrentRunState(this))
cancel(false);
//10 只执行一次，调用schedule方法时候
else if (!isPeriodic())
super.run();
//11定时执行
else if (super.runAndReset()) {
//11.1设置time=time+period
setNextRunTime();
//11.2重新加入该任务到delay队列
reExecutePeriodic(outerTask);
}
}
代码8中isPeriodic的作用是判断当前任务是一次性任务还是可重复执行的任务
public boolean isPeriodic() {
return period != 0;
}
其内部是通过period的值来判断，由于转换任务在创建ScheduledFutureTask时传递的period的值为0，所以这里isPeriodic
返回false.
6.代码9判断当前任务是否应该被取消，
boolean canRunInCurrentRunState(RunnableScheduledFuture<?> task) {
return isRunningOrShutdown（preiodic？continueExistingPeriodicTasksAfterShutdown：
executeExistingDelayedTasksAfterShutdown）；
}
这里传递的preiodic的值为false，所以isRunningOrShutdown的参数为executeExistingDelayedTasksAfterShutdown。executeExistingDelayedTasksAfterShutdown默认为true，表示当其他线程调用了shutdown命令关闭了线程后，当前任务还是要执行，否则如果为false，则当前任务要取消。
7.由于periodic的值为false，所以执行代码10父类FutureTask的方法执行具体执行任务，FutureTask的run方法代码如下。
public void run() {
//12
if (state != NEW ||
!RUNNER.compareAndSet(this, null, Thread.currentThread()))
return;
//13
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
//13.1
setException(ex);
}
//13.2
if (ran)
set(result);
}
} finally {
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
代码12判断如果任务状态不是new则直接返回，或者如果当前任务状态为new但是使用CAS设置当前任务的持有者为当前线程失败则直接返回，代码13具体调用callable的call方法执行任务。这里在调用前又判断了任务的状态是否为new，是为了避免在执行代码12后其他线程不该了任务的状态（比如取消了该任务）
如果任务执行成功则执行代码13.2修改任务的状态，set方法的代码如下
protected void set(V v) {
//如果当前任务的状态为new，则设置为COMPLETING
if (STATE.compareAndSet(this, NEW, COMPLETING)) {
outcome = v;
//设置当前任务的状态为normal，也就是任务正常结束
STATE.setRelease(this, NORMAL); // final state
finishCompletion();
}
}
如上的代码首先使用CAS将当前任务的状态从NEW转换为COMPLETING。这里当有多个线程调用时只有一个线程会成功。成功的线程在此通过unsafe.putOrderInt设置任务的状态为正常结束状态，这里没有使用CAS时因为对于同一个任务只可能有一个线程运行到这里。在这里使用putOrderInt比使用CAS或者putLongvolatile效率更高，并且这里的场景不要求其他线程马上对设置的状态值可见。
思考个问题，在什么时候多个线程会同时执行CAS将当前任务的状态从NEW转换到COMPLETING？其实当同一个command被多次提交到线程池就会存在这样的情况，因为痛一个任务共享一个状态值state。
如果任务执行失败，则执行代码13.1，setException的代码如下，可见与set函数类似。
protected void setException(Throwable t) {
//如果当前任务的状态为new，则设置为COMPLETING
if (STATE.compareAndSet(this, NEW, COMPLETING)) {
outcome = t;
//设置当前任务的状态为EXCEPTIONAL，也就是任务非正常结束
STATE.setRelease(this, EXCEPTIONAL); // final state
finishCompletion();
}
}
到这里代码10的逻辑执行完毕，一次性任务也就执行完毕了
scheduleWithFixedDelay
**** 该方法的作用是，当任务执行完毕后，让其延迟固定时间后再次运行（fixed-delay）。其中initialDelay表示提交任务后延迟多少时间可以执行command任务，delay表示当任务执行完毕后延长多少时间后再次运行command任务，unit是delay和initialDelay时间单位，任务会一直重复运行中直到运行中抛出异常，被取消了，或者关闭了线程池。
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
//14参数校验 TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (delay <= 0L)
throw new IllegalArgumentException();
//15任务转换，注意这里是period=-delay<0
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
-unit.toNanos(delay),
sequencer.getAndIncrement());
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
sft.outerTask = t;
//16 添加任务到队列
delayedExecute(t);
return t;
}
代码14进行参数校验，校验失败则抛出异常，代码15将command任务转化为ScheduledFutureTask，这里需要注意的是，传递给ScheduledFutureTask的period变量的值为-delay，period<0说明该任务时可重复执行的任务。然后代码16添加任务到延迟队列后返回。
将任务添加到延迟队列后线程池线程会从队列里面获取任务，然后调用ScheduledFutureTask的run方法执行，优质这里的period<0,所以isPeriodic返回true，所以执行代码11。
protected boolean runAndReset() {
//17
if (state != NEW ||
!RUNNER.compareAndSet(this, null, Thread.currentThread()))
return false;
//18
boolean ran = false;
int s = state;
try {
Callable<V> c = callable;
if (c != null && s == NEW) {
try {
c.call(); // don't set result
ran = true;
} catch (Throwable ex) {
setException(ex);
}
}
} finally {
runner = null;
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;//19
}
这个代码和FutureTask的run方法类似，只是任务正常执行完毕后不会设置任务的状态，这样做是为了让任务成为可重复执行的任务，这里多了代码19，这段代码判断当前任务正常执行完毕并且任务状态为NEW则返回true，否则返回false。如果true则执行代码11.1的setNextRunTime方法设置该任务下次一的执行时间。setNextRunTime的代码如下
private void setNextRunTime() {
long p = period;
if (p > 0)//fixed-rate类型任务
time += p;
else//fixed-delay类型任务
time = triggerTime(-p);
}
这里p<0说明当前任务为fixed-delay类型任务。然后设置time为当前时间加上-p的时间，也就是延迟-p时间后再次执行
总结：fixed-delay类型的任务的执行原理为：当添加一个人任务到延迟队列后，等待initialDelay时间，任务就会过期，过去的任务就会被从队列移除，并执行，执行完毕后，会重新设置任务的延迟时间，然后再把任务放入延迟队列，循环往复，需要注意的是，如果一个任务在执行中抛出了异常，那么这个任务就结束了，但是不影响其他任务的执行
scheduleAtFixedRate
这个方法相对其实时间点以固定频率调用指定的任务(fixed-rate)。当把任务提交到线程池并延迟initialDelay时间，时间单位为（unit）后开始执行任务command。然后从initialDelay+period时间点再次执行，而后在initialDelay+2*period时间点再次执行，循环往复，直到抛出异常或者调用了任务的cancel取消了任务，或者关闭了线程池，scheduleAtFixedRate和scheduleWithFixedDelay类似。
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit) {
...
//装饰任务类，注意period=period》0，不是负的
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(period),
return t;
}
在如上代码中，在fixed-rate类型的任务command转换为ScheduledFutureTask是设置period=period，不再是-period
所以当前任务执行完毕后，调用setNextRunTime设置任务下次执行的时间时执行的是time+=p而不再是time=triggerTime（-p）
总结：相对于fixed-delay任务来说，fixed-rate方法执行规则为，时间为initdelday+nperiod时启动任务，但是如果当前任务还没有执行完，下一次要执行任务的时间到了，则不会并发执行，下次要执行的任务会延迟执行，要等到当前任务执行完毕后在执行*
总结
ScheduledThreadPoolExecutor的内部使用了DelayedQueue来存放具体任务。任务分为三种，其中一次性执行任务执行完毕后就结束了，fixed-delay任务保证同一个任务在多次执行期间间隔固定时间，fixed-rate任务保证按照固定的频率执行。任务类型使用period的值来划分。
参考资料：
《Java并发编程之美》
