多线程学习-Executor框架
前言
Executor框架提供了组件来管理Java中的线程，Executor框架将其分为任务，线程执行任务，任务执行结果三部分。本篇文章将对Executor框架中的组件进行学习。
参考：《Java并发编程的艺术》
正文
一. Executor的组件
前言中已经提到，Executor框架将线程的管理分为任务，线程执行任务，任务执行结果三部分。下面以图表形式对这三部分进行说明。
| 项 | 说明 |
|---|---|
| 任务 | Executor框架提供了Runnable接口和Callable接口，任务需要实现这两个接口才能被线程执行。 |
| 线程执行任务 | Executor框架提供了接口Executor和继承于Executor的ExecutorService接口来定义任务执行机制。Executor框架中的线程池类ThreadPoolExecutor和ScheduledThreadPoolExecutor均实现了ExecutorService接口。 |
| 任务执行结果 | Executor框架提供了Future接口和实现了Future接口的FutureTask类来定义任务执行结果。 |
文章图片
二. ThreadPoolExecutor的解析
ThreadPoolExecutor继承于AbstractExecutorService，并实现了ExecutorService接口，是Executor框架的核心类，用于管理线程。在多线程学习-线程池使用中已经对ThreadPoolExecutor的原理，创建，执行和关闭进行了简单学习。在本小节将对ThreadPoolExecutor的具体实现进行学习。ThreadPoolExecutor使用了原子整型ctl来表示线程池状态和Worker数量。ctl是一个原子整型，前3位表示线程池状态，后29位表示Worker数量。ThreadPoolExecutor中这部分的源码如下所示。public class ThreadPoolExecutor extends AbstractExecutorService {private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY= (1 << COUNT_BITS) - 1;
private static final int RUNNING= -1 << COUNT_BITS;
private static final int SHUTDOWN=0 << COUNT_BITS;
private static final int STOP=1 << COUNT_BITS;
private static final int TIDYING=2 << COUNT_BITS;
private static final int TERMINATED =3 << COUNT_BITS;
//取整型前3位，即获取线程池状态
private static int runStateOf(int c){ return c & ~CAPACITY;
}
//取整型后29位，即获取Worker数量
private static int workerCountOf(int c){ return c & CAPACITY;
}
//根据线程池状态和Worker数量拼装ctl
private static int ctlOf(int rs, int wc) { return rs | wc;
}//线程池状态判断
private static boolean runStateLessThan(int c, int s) {
return c < s;
}//线程池状态判断
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}//判断线程池状态是否为RUNNING
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}......}在
ThreadPoolExecutor中规定了线程池的状态如下。- RUNNING，线程池接受新任务，会执行任务阻塞队列中的任务，
ctl前三位表示为111； - SHUTDOWN，线程池拒绝新任务，会执行任务阻塞队列中的任务，
ctl前三位表示为000； - STOP，线程池拒绝新任务，不会执行任务阻塞队列中的任务，尝试中断正在执行的任务，
ctl前三位表示为001； - TIDYING，所有任务被关闭，
Worker数量为0，ctl前三位表示为010； - TERMINATED，
terminated()执行完毕，ctl前三位表示为011。
ctl的结构，所以无论Worker数量是多少，ThreadPoolExecutor中线程池状态存在如下关系。RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
因此
runStateLessThan()，runStateAtLeast()和isRunning()方法可以方便的对线程池状态进行判断。ThreadPoolExecutor中执行任务的入口方法为execute()，其实现如下。public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//如果Worker数量小于核心线程数，则创建Worker并执行任务
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//如果Worker数量大于等于核心线程数，则将任务添加到任务阻塞队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//如果线程池状态突然不再是RUNNING，则尝试将任务从任务阻塞队列中删除，删除成功则为该任务执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
//如果线程池中Worker数量突然为0，则创建一个Worker来执行任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//执行到这里表示线程池状态已经不是RUNNING或者任务阻塞队列已满
//此时尝试新建一个Worker来执行任务
//如果新建一个Worker来执行任务失败，表明线程池状态不再是RUNNING或者Worker数量已经达到最大线程数，此时执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}在
execute()中会根据Worker数量和线程池状态来决定是新建Worker来执行任务还是将任务添加到任务阻塞队列。新建Worker来执行任务的实现如下所示。private boolean addWorker(Runnable firstTask, boolean core) {
//标记外层for循环
retry:
for (;
;
) {
int c = ctl.get();
//获取线程池状态
int rs = runStateOf(c);
//rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())
//线程池状态为RUNNING时，可以创建Worker
//线程池状态为SHUTDOWN，且任务阻塞队列不为空时，可以创建初始任务为null的Worker
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;
;
) {
//获取Worker数量
int wc = workerCountOf(c);
//如果Worker数量大于CAPACITY，拒绝创建Worker
//core为true表示创建核心线程Worker，如果Worker数量此时已经大于等于核心线程数，则拒绝创建Worker，转而应该将任务添加到任务阻塞队列
//core为false表示创建非核心线程Worker，如果Worker数量此时已经大于等于最大线程数，则拒绝创建Worker，转而应该执行拒绝策略
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//以CAS方式将Worker数量加1
//加1成功表明无竞争发生，从外层for循环跳出
if (compareAndIncrementWorkerCount(c))
break retry;
//加1失败表明有竞争发生，此时需要重新获取ctl的值
c = ctl.get();
//重新获取ctl后如果发现线程池状态发生了改变，此时重新执行外层for循环，即需要基于新的线程池状态判断是否允许创建Worker
//重新获取ctl后如果线程池状态未发生改变，则继续执行内层for循环，即尝试再一次以CAS方式将Worker数量加1
if (runStateOf(c) != rs)
continue retry;
}
}boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//创建一个Worker
w = new Worker(firstTask);
//获取Worker的线程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//由于线程池中存储Worker的集合为HashSet，因此将Worker添加到Worker集合时需要获取全局锁保证线程安全
mainLock.lock();
try {
//再一次获取线程池状态
int rs = runStateOf(ctl.get());
//如果线程池状态还是为RUNNING或者线程池状态为SHUTDOWN但创建的Worker的初始任务为null，则允许将创建出来的Worker添加到集合
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//检查一下Worker的线程是否可以启动（处于活动状态的线程无法再启动）
if (t.isAlive())
throw new IllegalThreadStateException();
//将Worker添加到Worker集合
workers.add(w);
int s = workers.size();
//largestPoolSize用于记录线程池最多存在过的Worker数
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//启动Worker线程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//Worker线程没有成功启动起来，此时需要对该Worker的创建执行回滚操作
addWorkerFailed(w);
}
return workerStarted;
}addWorker()方法中只允许两种情况可以创建Worker。- 线程池状态为RUNNING，可以创建
Worker； - 线程池状态为SHUTDOWN，且任务阻塞队列不为空，可以创建初始任务为null的
Worker。
Worker创建成功，就会将Worker的线程启动来，如果Worker创建失败或者Worker的线程启动失败，则会调用addWorkerFailed()方法执行回滚操作，其实现如下所示。private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//如果Worker添加到了Worker集合中，则将Worker从Worker集合中删除
if (w != null)
workers.remove(w);
//以CAS方式将Worker数量减1
decrementWorkerCount();
//尝试终止线程池
tryTerminate();
} finally {
mainLock.unlock();
}
}由于
Worker自身实现了Runnable，因此Worker自身就是一个任务，实际上Worker的线程执行的任务就是Worker本身，因此addWorker()中将Worker的线程启动时，会调用Worker的run()方法，其实现如下。public void run() {
runWorker(this);
}在
Worker的run()方法中调用了ThreadPoolExecutor的runWorker()方法，其实现如下所示。final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
boolean completedAbruptly = true;
try {
//如果task为null，则从任务阻塞队列中获取任务
//通常Worker启动时会先执行初始任务，然后再去任务阻塞队列中获取任务
while (task != null || (task = getTask()) != null) {
w.lock();
//线程池正在停止时，需要确保当前Worker的线程是被中断的
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
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执行任务发生异常或者从getTask()中获取任务为空时会执行这里的逻辑
//processWorkerExit()会将Worker从Worker集合中删除，并尝试终止线程池
processWorkerExit(w, completedAbruptly);
}
}runWorker()方法就是先让Worker将初始任务（如果有的话）执行完，然后循环从任务阻塞队列中获取任务来执行，如果Worker执行任务发生异常或者从任务阻塞队列获取任务失败（获取到的任务为null），则调用processWorkerExit()方法来将自身从Worker集合中删除。下面先看一下getTask()方法的实现。private Runnable getTask() {
boolean timedOut = false;
for (;
;
) {
int c = ctl.get();
int rs = runStateOf(c);
//如果线程池状态为SHUTDOWN，且任务阻塞队列为空，则不再允许从任务阻塞队列中获取任务
//如果线程池状态为STOP，则不再允许从任务阻塞队列中获取任务
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}int wc = workerCountOf(c);
//如果allowCoreThreadTimeOut为true，或者当前线程数大于核心线程数，此时timed为true，表明从任务阻塞队列以超时退出的方式获取任务
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//如果当前线程数大于最大线程数，则当前Worker应该被删除
//如果当前Worker上一次从任务阻塞队列中获取任务时超时退出，且任务阻塞队列现在还是为空，则当前Worker应该被删除
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}try {
//从任务阻塞队列中获取任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
//获取到任务则返回该任务
return r;
//timedOut为true表明Worker上一次从任务阻塞队列中获取任务时超时退出
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}getTask()方法在如下情况不允许Worker从任务阻塞队列中获取任务。- 线程池状态为SHUTDOWN，且任务阻塞队列为空；
- 线程池状态为STOP。
Worker有资格从任务阻塞队列获取任务，那么当allowCoreThreadTimeOut为true，或者当前线程数大于核心线程数时，Worker以超时退出的方式获取任务，否则Worker以一直阻塞的方式获取任务。当
Worker在getTask()方法中获取任务失败时，getTask()方法会返回null，从而导致Worker会执行processWorkerExit()方法来删除自身，其实现如下所示。private void processWorkerExit(Worker w, boolean completedAbruptly) {
//completedAbruptly为true表明是执行任务时发生异常导致Worker需要被删除
if (completedAbruptly)
//修正Worker数量
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
//将Worker从Worker集合中删除
workers.remove(w);
} finally {
mainLock.unlock();
}//尝试终止线程池
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return;
}
addWorker(null, false);
}
}Worker在processWorkerExit()方法中删除自身之后，还会调用tryTerminate()尝试终止线程池，tryTerminate()方法很精髓，后面会对其进行详细分析，这里暂且不谈。至此，Worker的创建，执行任务，获取任务和删除的整个流程已经大体分析完毕。下面对ThreadPoolExecutor中关闭线程池的shutdown()和shutdownNow()方法进行分析。首先分析
shutdown()方法，其实现如下。public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
//循环通过CAS方式将线程池状态置为SHUTDOWN
advanceRunState(SHUTDOWN);
//中断空闲Worker
interruptIdleWorkers();
onShutdown();
} finally {
mainLock.unlock();
}
//尝试终止线程池
tryTerminate();
}在
shutdown()方法中首先会将线程池状态置为SHUTDOWN，然后调用interruptIdleWorkers()方法中断空闲Worker，最后调用tryTerminate()方法来尝试终止线程池。那么这里要解释一下什么是空闲Worker，先看一下interruptIdleWorkers()的实现。private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
//中断线程前需要先尝试获取Worker的锁
//只能获取到空闲Worker的锁，所以shutdown()方法只会中断空闲Worker
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}调用
interruptIdleWorkers()方法中断Worker前首先需要尝试获取Worker的锁，已知Worker除了实现Runnable接口外，还继承于AbstractQueuedSynchronizer，因此Worker本身是一把锁，然后在runWorker()中Worker执行任务前都会先获取Worker的锁，这里看一下Worker的lock()方法的实现。public void lock() {
acquire(1);
}protected boolean tryAcquire(int unused) {
//以CAS方式将state从0设置为1
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}可以发现，
Worker在lock()中调用了acquire()方法，该方法由AbstractQueuedSynchronizer抽象类提供，在acquire()中会调用其子类实现的tryAcquire()方法，tryAcquire()方法会以CAS方式将state从0设置为1，因此这样的设计让Worker是一把不可重入锁。回到interruptIdleWorkers()方法，前面提到该方法中断Worker前会尝试获取Worker的锁，能够获取到锁才会中断Worker，而因为Worker是不可重入锁，所以正在执行任务的Worker是无法获取到锁的，只有那些没有执行任务的Worker的锁才能够被获取，因此所谓的中断空闲Worker，实际就是中断没有执行任务的Worker，那些执行任务的Worker在shutdown()方法被调用时不会被中断，这些Worker执行完任务后会继续从任务阻塞队列中获取任务来执行，直到任务阻塞队列为空，此时没有被中断过的Worker也会被删除掉，等到线程池中没有Worker以及任务阻塞队列没有任务后，线程池才会被终止掉。对于
shutdown()方法，一句话总结就是：将线程池状态置为SHUTDOWN并拒绝接受新任务，等到线程池Worker数量为0，任务阻塞队列为空时，关闭线程池。现在再来分析
shutdownNow()方法。public List shutdownNow() {
List tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
//循环通过CAS方式将线程池状态置为STOP
advanceRunState(STOP);
//中断所有Worker
interruptWorkers();
//将任务阻塞队列中的任务获取出来并返回
tasks = drainQueue();
} finally {
mainLock.unlock();
}
//尝试终止线程池
tryTerminate();
return tasks;
}private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//中断线程池中所有Worker
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
} 在
shutdownNow()方法中首先会将线程池状态置为STOP，然后调用interruptWorkers()方法中断线程池中的所有Worker，接着调用tryTerminate()方法来尝试终止线程池，最后shutdownNow()方法会将任务阻塞队列中还未被执行的任务返回。shutdownNow()方法调用之后，线程池中的所有Worker都会被中断，包括正在执行任务的Worker，等到所有Worker都被删除之后，线程池即被终止，也就是说，shutdownNow()不会保证当前时刻正在执行的任务会被安全的执行完，并且会放弃执行任务阻塞队列中的所有任务。关于线程池的关闭，还有一个重要的方法，那就是前面多次提到的
tryTerminate()方法，该方法能确保线程池可以被正确的关闭，其实现如下所示。final void tryTerminate() {
for (;
;
) {
int c = ctl.get();
//如果线程池状态为RUNNING，则没有资格终止线程池
//如果线程池状态大于等于TIDYING，则没有资格终止线程池
//如果线程池状态为SHUTDOWN但任务阻塞队列不为空，则没有资格终止线程池
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//线程池状态为SHUTDOWN且任务阻塞队列为空会执行到这里
//线程池状态为STOP会执行到这里
//Worker数量不为0，表明当前还有正在执行任务的Worker或者空闲的Worker，此时中断一个空闲的Worker
//在这里被中断的空闲Worker会在getTask()方法中返回null，从而执行processWorkerExit()，最终该Worker会被删除
//processWorkerExit()方法中又会调用tryTerminate()，因此将shutdown信号在空闲Worker之间进行了传播
if (workerCountOf(c) != 0) {
interruptIdleWorkers(ONLY_ONE);
return;
}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.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
}
}在
tryTerminate()方法的官方注释中给出了两种线程池会被终止的情况：- 线程池的状态为SHUTDOWN，
Worker数量为0，任务阻塞队列为空； - 线程池的状态为STOP，
Worker数量为0。
tryTerminate()方法来尝试终止线程池，因此线程池中Worker被删除时和任务阻塞队列中任务被删除时会调用tryTerminate()，以达到在线程池符合终止条件时及时终止线程池。小节：
ThreadPoolExecutor执行任务的入口方法为execute()，如果Worker数量小于核心线程数，则创建Worker并执行任务，如果Worker数量大于等于核心线程数，则将任务添加到任务阻塞队列，如果任务阻塞队列已满但Worker数量小于最大线程数，则创建Worker并执行任务，如果Worker数量已经大于等于最大线程数，此时执行拒绝策略。执行完任务的Worker会调用getTask()方法从任务阻塞队列获取任务，如果Worker从任务阻塞队列获取任务失败那么线程池会删除这个Worker。调用shutdown()方法会置线程池状态为SHUTDOWN，此时会等到线程池Worker数量为0，任务阻塞队列为空时将线程池终止；调用shutdownNow()方法会置线程池状态为STOP，此时会等到线程池Worker数量为0时将线程池终止。三. ScheduledThreadPoolExecutor的解析
ScheduledThreadPoolExecutor继承于ThreadPoolExecutor，扩展实现了延时执行任务和定时执行任务的功能。ScheduledThreadPoolExecutor存储任务的队列为DelayedWorkQueue，是一个基于小根堆实现的延时优先级队列，ScheduledThreadPoolExecutor会将每一个提交到线程池的任务先封装为ScheduledFutureTask，然后再插入到DelayedWorkQueue中。下面将结合源码，对ScheduledThreadPoolExecutor的原理进行解析。1. 任务提交
ScheduledThreadPoolExecutor提供了四个提交任务的方法，如下所示。//提交无返回值的延时任务
public ScheduledFuture schedule(Runnable command,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
RunnableScheduledFuture t = decorateTask(command,
new ScheduledFutureTask(command, null,
triggerTime(delay, unit)));
delayedExecute(t);
return t;
}//提交有返回值的延时任务
public ScheduledFuture schedule(Callable callable,
long delay,
TimeUnit unit) {
if (callable == null || unit == null)
throw new NullPointerException();
RunnableScheduledFuture t = decorateTask(callable,
new ScheduledFutureTask(callable,
triggerTime(delay, unit)));
delayedExecute(t);
return t;
}//提交固定周期的定时任务
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;
}//提交固定延时的定时任务
public ScheduledFuture scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (delay <= 0)
throw new IllegalArgumentException();
ScheduledFutureTask sft =
new ScheduledFutureTask(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(-delay));
RunnableScheduledFuture t = decorateTask(command, sft);
sft.outerTask = t;
delayedExecute(t);
return t;
} 可知
ScheduledThreadPoolExecutor支持执行如下的任务。- 无返回值的延时任务；
- 有返回值的延时任务；
- 固定周期的定时任务；
- 固定延时的定时任务。
在提交任务的四个方法中，均是将提交的任务封装为
ScheduledFutureTask，在将任务封装成ScheduledFutureTask时需要指定任务首次执行的时间点（即初始延时），和任务的执行间隔（为正值表示固定周期，为负值表示固定时延，为0表示仅执行一次），而指定任务首次执行的时间点时，为了防止时间点的值不在长整型的最大值范围内，需要在triggerTime()方法中进行处理，如下所示。//获取延时操作的触发时间
//触发时间 = 当前时间 + 延时时间
private long triggerTime(long delay, TimeUnit unit) {
return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}//如果延时时间大于等于长整型最大值的一半，则执行overflowFree()方法对延时时间进行处理
//执行overflowFree()方法就是为了使得最终得到的触发时间的值在长整型最大值以内，以防止compareTo()时值溢出
long triggerTime(long delay) {
return now() +
((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}封装好的
ScheduledFutureTask会在delayedExecute()方法中添加到延时队列，如下所示。private void delayedExecute(RunnableScheduledFuture task) {
//如果线程池状态不是RUNNING，则执行拒绝策略
if (isShutdown())
reject(task);
else {
//将任务添加到延时队列中
super.getQueue().add(task);
//此时如果线程池状态为非RUNNING，并且线程池策略为非RUNNING状态下延时任务或定时任务不再执行，则将任务从延时队列中删除
//将任务从延时队列中删除后，还需要关闭任务，如果任务尚未执行，那么关闭任务后任务就不会再被执行
//如果任务正在被执行，则不会尝试中断执行任务的Worker来关闭任务
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
//创建初始任务为null的Worker，即使核心线程数为0也需要确保线程池至少有一个Worker在工作
ensurePrestart();
}
}void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(null, false);
}如果线程池状态为RUNNING，那么会将封装好的
ScheduledFutureTask添加到延时队列，同时会创建一个初始任务为null的Worker，之所以创建的Worker初始任务为null，是因为ScheduledThreadPoolExecutor中的任务ScheduledFutureTask都会有一个初始延时，即提交到线程池的任务不会立即被执行，所以Worker均需要到延时队列中去获取任务。2. 任务-ScheduledFutureTask 已知提交到
ScheduledThreadPoolExecutor中的任务均会被封装成ScheduledFutureTask，因此这里对ScheduledFutureTask的原理进行学习。首先查看
ScheduledFutureTask的类图，如下所示。
文章图片
即
ScheduledFutureTask继承了FutureTask的功能，同时实现了RunnableScheduledFuture接口的方法。下面看一下ScheduledFutureTask的字段。private class ScheduledFutureTask
extends FutureTask implements RunnableScheduledFuture {//任务添加到ScheduledThreadPoolExecutor中时分配的一个序列号
private final long sequenceNumber;
//任务下次执行的时间点
private long time;
//任务的执行间隔
//大于0表示任务是以固定周期执行的任务
//小于0表示任务是以固定延时执行的任务
//等于0表示任务是非重复执行的任务
private final long period;
//指向当前任务本身
RunnableScheduledFuture outerTask = this;
//任务在延时队列中的索引
int heapIndex;
......} 因为
ScheduledFutureTask是作为元素存储在基于小根堆实现的延时优先级队列中，所以ScheduledFutureTask提供了sequenceNumber和time这两个字段用于堆中元素之间的比较，然后heapIndex就是ScheduledFutureTask在延时队列中的索引，即元素在堆数组中的索引。ScheduledFutureTask提供了三个构造方法，如下所示。ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}ScheduledFutureTask(Runnable r, V result, long ns, long period) {
super(r, result);
this.time = ns;
this.period = period;
this.sequenceNumber = sequencer.getAndIncrement();
}ScheduledFutureTask(Callable callable, long ns) {
super(callable);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
} 在
ScheduledFutureTask的构造方法中，会先构造父类FutureTask，然后根据构造方法入参设置任务下次的执行时间点和任务的执行间隔，并为任务分配sequenceNumber。下面再看一下
ScheduledFutureTask是如何进行比较的。ScheduledFutureTask实现了Comparable接口，其实现的compareTo()方法如下所示。public int compareTo(Delayed other) {
if (other == this)
return 0;
if (other instanceof ScheduledFutureTask) {
ScheduledFutureTask x = (ScheduledFutureTask)other;
//优先比较任务下一次执行的时间点先后，越先执行time越小
long diff = time - x.time;
if (diff < 0)
return -1;
else if (diff > 0)
return 1;
//如果任务下一次执行的时间点相同，则比较sequenceNumber
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;
}两个
ScheduledFutureTask进行比较时，会先比较下一次任务执行的时间先后，如果下一次任务执行的时间一样，则再根据sequenceNumber进行比较。最后再看一下
ScheduledFutureTask的run()方法，如下所示。public void run() {
boolean periodic = isPeriodic();
if (!canRunInCurrentRunState(periodic))
//如果当前线程池状态下不允许执行任务，则关闭任务
cancel(false);
else if (!periodic)
//执行延时任务
ScheduledFutureTask.super.run();
//执行定时任务，然后为定时任务设置下一次执行的时间点并添加到延时队列中
else if (ScheduledFutureTask.super.runAndReset()) {
setNextRunTime();
reExecutePeriodic(outerTask);
}
}ScheduledFutureTask在执行run()方法时会先根据periodic字段的值判断任务是延时任务（只在延时时间到了后执行一次）还是定时任务（需要根据指定间隔重复执行），如果是定时任务，那么在定时任务执行完之后，会为定时任务设置下一次执行的时间点并重新添加到延时队列中。设置定时任务下一次执行的时间点的setNextRunTime()实现如下。//设置任务的下一次执行的时间点
private void setNextRunTime() {
long p = period;
//如果是以固定周期执行的任务，则下一次执行的时间点就是上一次执行的时间点加上执行间隔
if (p > 0)
time += p;
//如果是以固定延时执行的任务，则下一次执行的时间点就是当前时间加上执行间隔
else
time = triggerTime(-p);
}3. 延时队列-DelayedWorkQueue
DelayedWorkQueue是ScheduledThreadPoolExecutor线程池使用的任务阻塞队列。DelayedWorkQueue是基于小根堆实现的延时优先级队列，队列中的元素就是ScheduledFutureTask，因此DelayedWorkQueue的队列头节点任务总是最优先被执行的任务。先看一下DelayedWorkQueue的字段，如下所示。static class DelayedWorkQueue extends AbstractQueue
implements BlockingQueue {//堆数组的初始大小
private static final int INITIAL_CAPACITY = 16;
//堆数组，数组中的元素实际上是ScheduledFutureTask
private RunnableScheduledFuture[] queue =
new RunnableScheduledFuture[INITIAL_CAPACITY];
private final ReentrantLock lock = new ReentrantLock();
//延时队列元素个数
private int size = 0;
//在延时队列头等待任务的领导线程
private Thread leader = null;
private final Condition available = lock.newCondition();
......} 特别说明一下leader字段和available字段，首先是leader字段，表示在延时队列头等待任务的第一个线程，即如果延时队列头的任务需要被执行时，这个任务会被leader字段指向的线程获得。同时所有在延时队列头等待任务的线程，均会在available上进入等待状态，并且在延时队列头的任务需要被执行时或者延时队列头的任务被更新时唤醒所有在available上等待的线程。
已知
DelayedWorkQueue是一个基于小根堆实现的延时优先级队列，那么往DelayedWorkQueue中插入和删除任务后，均需要保持堆的性质，在DelayedWorkQueue中，主要是siftUp()和siftDown()这两个方法来保持堆的性质，siftUp()是用于往DelayedWorkQueue中插入任务时来保持堆的性质，而siftDown()是用于DelayedWorkQueue弹出任务后保持堆的性质，其实现如下。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);
}siftDown()实现如下所示。private void siftDown(int k, RunnableScheduledFuture key) {
int half = size >>> 1;
while (k < half) {
//计算左子节点索引，并将左子节点索引赋值给child
int child = (k << 1) + 1;
RunnableScheduledFuture c = queue[child];
//计算右子节点索引
int right = child + 1;
//令c表示左子节点和右子节点中元素值更小的元素
//令child表示左子节点和右子节点中元素值更小的节点索引
if (right < size && c.compareTo(queue[right]) > 0)
c = queue[child = right];
//将当前元素值与c的值进行比较，如果当前元素值已经小于等于c的值，则退出循环
if (key.compareTo(c) <= 0)
break;
//如果当前元素值大于c的值，则将当前元素与c互换位置
queue[k] = c;
setIndex(c, k);
k = child;
}
queue[k] = key;
setIndex(key, k);
}理解了
siftUp()和siftDown()这两个方法之后，先来看一下DelayedWorkQueue中添加任务的实现。因为DelayedWorkQueue实现了BlockingQueue接口，因此对外提供了put()，add()，offer()和超时退出的offer()这四个方法来添加任务，但是因为DelayedWorkQueue在容量满时会进行扩容，可以当成一个无界队列来看待，所以DelayedWorkQueue的put()，add()和超时退出的offer()方法均是调用的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)
//扩容后容量为扩容前容量的1.5倍
grow();
size = i + 1;
if (i == 0) {
queue[0] = e;
setIndex(e, 0);
} else {
//向延时队列插入任务
//与向堆插入元素一样，将任务插入到任务堆的末尾节点，并逐步与父节点进行比较和交换，直到满足堆的性质为止
siftUp(i, e);
}
if (queue[0] == e) {
//如果插入的任务最终成为延时队列头节点任务，那么重置领导线程leader并唤醒所有等待获取任务的线程
leader = null;
available.signal();
}
} finally {
lock.unlock();
}
return true;
}同样的，
DelayedWorkQueue对外提供了remove()，poll()，take()和超时退出的poll()这四个方法来移除或获取任务，这里重点分析一下take()和超时退出的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 {
//delay表示延时队列头节点任务的剩余等待时间
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
//如果延时队列头节点任务的剩余等待时间小于等于0，则弹出头节点任务并保持堆性质
return finishPoll(first);
first = null;
if (leader != null)
//如果已经存在领导线程，则进入等待状态
available.await();
else {
//如果不存在领导线程，则将当前Worker的线程置为领导线程
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
//等待delay的时间，即等到延时队列头任务可以执行
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}take()方法中，首先判断延时队列头节点任务是否为空，如果为空则直接在available上进入等待状态，如果不为空则再判断任务是否已经可以执行，若可以执行则直接弹出任务并返回，若还不能执行那么就再判断领导线程是否已经存在，如果存在那么说明当前线程不是在延时队列头等待任务的第一个线程，需要在available上进入等待状态，如果不存在就说明当前线程是在延时队列头等待任务的第一个线程，需要将当前线程置为领导线程，然后在available上进入等待状态直到头节点任务可以执行。超时退出的
poll()和take()方法的大体实现一样，只是超时退出的poll()还需要额外加入对Worker从延时队列获取任务的等待时间的判断，其实现如下所示。public RunnableScheduledFuture poll(long timeout, TimeUnit unit)
throws InterruptedException {
//nanos表示Worker从延时队列获取任务的等待时间
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;
;
) {
RunnableScheduledFuture first = queue[0];
//延时队列为空时
//如果nanos小于等于0，则直接返回null
//如果nanos大于0，则进入等待状态并等待nanos的时间
if (first == null) {
if (nanos <= 0)
return null;
else
nanos = available.awaitNanos(nanos);
} else {
//delay表示延时队列头节点任务的剩余等待时间
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
//如果延时队列头节点任务的剩余等待时间小于等于0，则弹出头节点任务并保持堆性质
return finishPoll(first);
if (nanos <= 0)
//如果Worker从延时队列获取任务的等待时间小于等于0，则返回null
return null;
first = null;
//如下情况会进入等待状态并等待nanos的时间
//Worker从延时队列获取任务的等待时间小于延时队列头节点任务的剩余等待时间
//Worker从延时队列获取任务的等待时间大于等于延时队列头节点任务的剩余等待时间，但领导线程不为空
if (nanos < delay || leader != null)
nanos = available.awaitNanos(nanos);
else {
//如果Worker从延时队列获取任务的等待时间大于等于延时队列头节点任务的剩余等待时间，且领导线程为空
//将领导线程置为当前Worker的线程
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
//等待delay的时间，即等到延时队列头任务可以执行
long timeLeft = available.awaitNanos(delay);
//重新计算Worker从延时队列获取任务的等待时间
nanos -= delay - timeLeft;
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}4. 小节 本节主要是对
ScheduledThreadPoolExecutor的任务提交，任务和延时优先级队列进行了分析，理解了这三个部分，也就会对ScheduledThreadPoolExecutor的大致工作原理有一个较为清晰的认识。而ScheduledThreadPoolExecutor关闭线程池是复用的ThreadPoolExecutor中的shutdown()和shutdownNow方法，故本节不再赘述。四. 任务和任务的执行结果
1. 任务 在
Executor框架中，Runnable接口和Callable接口用于定义任务，Runnable接口的实现类可以被Thread，ThreadPoolExecutor和ScheduledThreadPoolExecutor执行，Callable接口的实现类可以被ThreadPoolExecutor和ScheduledThreadPoolExecutor执行，此外，Runnable接口和Callable接口最大的不同在于：Runnable任务没有返回值，Callable任务有返回值。Executor框架提供的工具类Executors可以将Runnable封装成Callable，方法签名如下所示。public static Callable2. 任务的执行结果
Executor框架提供了Future接口来表示任务的异步计算结果。Future接口定义如下所示。public interface Future {//关闭任务的执行，在任务已经执行完毕或者已经被关闭时，返回false
//在该方法调用时如果任务正在被执行，mayInterruptIfRunning决定是否打断执行任务的线程来停止任务
boolean cancel(boolean mayInterruptIfRunning);
//判断任务是否被关闭过
//如果在任务执行完之前执行过关闭任务的操作，该方法返回true
boolean isCancelled();
//判断任务是否执行完毕
//因为中断，异常和关闭而导致的任务执行完毕，该方法均会返回true
boolean isDone();
//等待任务执行完毕并获取执行结果
V get() throws InterruptedException, ExecutionException;
//在指定时间内等待任务执行完毕并获取执行结果
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
} 此外，
Executor框架中还有一个RunnableFuture接口，该接口继承于Runnable接口和Future接口，即该接口的实现类即可以作为Runnable被执行，也能作为Future获取执行结果。所以
Executor框架提供了一个RunnableFuture接口的实现类FutureTask，而且，在调用ThreadPoolExecutor的submit()方法提交任务（Runnable或者Callable）或者向ScheduledThreadPoolExecutor提交任务（Runnable或者Callable）时，所提交的任务均会被封装为FutureTask，然后封装成FutureTask的任务会作为Runnable被添加到任务阻塞队列中，同时也会作为Future被返回。FutureTask的类图如下所示。
文章图片
3. FutureTask
FutureTask是Executor框架中重要的组件，下面对其原理进行学习。FutureTask的关键字段如下所示。public class FutureTask implements RunnableFuture {//任务状态，会有以下几种变化情况：
//NEW -> COMPLETING -> NORMAL
//NEW -> COMPLETING -> EXCEPTIONAL
//NEW -> CANCELLED
//NEW -> INTERRUPTING -> INTERRUPTED
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;
//需要被执行的任务
private Callable callable;
//任务执行的结果或抛出的异常
private Object outcome;
//正在执行任务的线程
private volatile Thread runner;
//记录调用get()方法等待的线程
private volatile WaitNode waiters;
......} 关于
FutureTask的状态字段稍后再分析，现在先看一下FutureTask中的需要被执行的任务callable字段，可知该字段为Callable接口，而前面已经分析知道无论是ThreadPoolExecutor还是ScheduledThreadPoolExecutor，均可以将Runnable或Callable封装为FutureTask，而FutureTask中却只有Callable，那么肯定是在某个地方将Runnable转换为了Callable，这个地方就是FutureTask的构造函数，如下所示。public FutureTask(Callable callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW;
} 可以看到，
FutureTask的构造函数中，如果传进来的任务为Runnable，那么会使用工具类Executors将Runnable转换为Callable。同时构造函数中还会将状态设置为NEW。下面分析
FutureTask的run()方法，源码如下所示。public void run() {
//判断任务状态是否为NEW，状态为NEW的任务才允许被执行
//如果任务状态为NEW，则以CAS方式将任务的runner字段设置为当前执行任务的线程，设置成功才允许执行任务
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//执行任务
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
//任务执行时抛出异常，将异常赋值给outcome字段
//然后以CAS方式将状态先置为COMPLETING，然后置为EXCEPTIONAL，最后唤醒所有调用get()方法进入等待的线程
setException(ex);
}
if (ran)
//任务执行时未抛出异常，将执行结果赋值给outcome字段
//然后以CAS方式将状态先置为COMPLETING，然后置为NORMAL，最后唤醒所有调用get()方法进入等待的线程
set(result);
}
} finally {
//在任务执行结束后将runner字段置为null
//在任务执行结束以前runner字段不能为null，以防止任务被并发多次执行
runner = null;
int s = state;
if (s >= INTERRUPTING)
//执行到这里，表示任务在执行时（任务状态为NEW）被调用了cancel(true)方法，并且cancel(true)中将任务状态置为了INTERRUPTING或INTERRUPTED
//如果任务状态为INTERRUPTING，则循环调用Thread.yield()来放弃时间片，直到任务状态变为INTERRUPTED
handlePossibleCancellationInterrupt(s);
}
} 在
run()方法中，主要的步骤如下。- 先在任务状态为NEW的情况下以CAS方式将执行任务的线程runner字段置为当前线程，任务状态不为NEW或者CAS失败，都直接退出
run()方法，防止任务被并发多次执行； - 任务执行成功或者任务执行抛出异常，会分别以CAS方式将任务状态先置为COMPLETING，如果CAS成功，之后调用
cancel()方法会直接返回false，但是如果CAS操作之前先调用了cancel()方法，那么会导致CAS失败，此时任务的状态为INTERRUPTING，CANCELLED或者INTERRUPTED； - 在
run()方法最后会判断任务的状态是否为INTERRUPTING或INTERRUPTED，如果满足，表明在任务执行完毕并翻转任务状态之前cancel(true)方法被调用了，此时如果任务状态为INTERRUPTING，则循环调用Thread.yield()来放弃时间片以等待任务状态翻转为INTERRUPTED。
FutureTask还提供了一个runAndReset()方法，该方法与run()方法略有不同，如下所示。runAndReset()不关心任务的执行结果，即不会将任务执行结果赋值给outcome字段；runAndReset()不会主动去翻转任务的状态，即任务正常执行完毕之后状态还是为NEW，以适用于任务需要多次被执行的情况，比如定时任务。
FutureTask的get()方法，FutureTask提供了一直等待直到任务执行完毕的get()方法，和指定等待时间的get()方法，如下所示。public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}public V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
if (unit == null)
throw new NullPointerException();
int s = state;
if (s <= COMPLETING &&
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
return report(s);
}在
get()方法中，会先在awaitDone()方法中获取任务状态，如果任务状态为NEW，则进入等待状态。获取到任务状态之后，在report()方法中根据获取到的任务状态来返回任务执行结果。awaitDone()方法实现如下所示。private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;
;
) {
if (Thread.interrupted()) {
//调用get()方法的线程如果被中断，则将其从等待链表中删除，并抛出中断异常
removeWaiter(q);
throw new InterruptedException();
}int s = state;
if (s > COMPLETING) {
//如果任务状态大于COMPLETING，说明任务执行完毕，或抛出异常，或被调用了cancel()方法
if (q != null)
q.thread = null;
//返回任务的状态
return s;
}
else if (s == COMPLETING)
//如果任务状态为COMPLETING，则放弃时间片，目的是为了调用get()方法的线程再次获得时间片时任务状态翻转为NORMAL或者EXCEPTIONAL
Thread.yield();
else if (q == null)
//执行到这里，说明任务状态为NEW
//基于调用get()方法的线程创建一个WaitNode节点
q = new WaitNode();
else if (!queued)
//如果WaitNode节点没有添加到等待链表，则将其加入到等待链表中
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
//如果调用get()方法时指定了等待时间，则使用LockSupport进入等待状态并指定等待时间
//如果等待时间到，任务状态还是NEW，则移除WaitNode节点并返回任务状态
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
//调用get()方法的线程进入等待状态
LockSupport.park(this);
}
}awaitDone()中，主要步骤如下所示。- 如果任务状态大于COMPLETING，即任务执行完毕，或抛出异常，或被调用了
cancel()方法，此时返回任务状态； - 如果任务状态为COMPLETING，表示任务已经执行完毕（或抛出异常），但是执行结果尚未赋值给outcome字段，此时调用
Thread.yield()放弃时间片，因为任务状态从COMPLETING向NORMAL或EXCEPTIONAL转换的时间非常短，所以Thread.yield()能够让调用get()方法的线程更快的响应任务状态的转换，最终目的是为了让调用get()方法的线程再次获得时间片时任务状态已经翻转为NORMAL或者EXCEPTIONAL； - 如果任务状态为NEW，那么说明任务还未开始执行或者任务正在执行，此时基于调用
get()方法的线程创建一个WaitNode节点并加入等待链表中； - 调用
get()方法的线程进入等待状态，只有在等待时间到（如果指定了的话），或者任务执行完毕，或者任务执行抛出异常，或者cancel()方法被调用时，所有等待线程才会结束等待状态。
report()方法的实现如下所示。private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x;
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}下面最后看一下
cancel()方法的实现，如下所示。public boolean cancel(boolean mayInterruptIfRunning) {
//如果任务状态不是NEW，则返回false
//如果以CAS方式将任务状态置为INTERRUPTING或者CANCELLED失败，则返回false
//即任务已经执行完毕或者已经被关闭时，返回false
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try {
//mayInterruptIfRunning为true表示需要中断正在执行任务的线程，并最终将任务状态置为INTERRUPTED
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally {
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
//唤醒所有调用get()方法进入等待的线程
finishCompletion();
}
return true;
}cancel()方法调用时如果任务已经执行完毕（状态为COMPLETING，NORMAL或EXCEPTIONAL），或者任务已经被关闭（状态为CANCELLED，INTERRUPTING或INTERRUPTED），则直接返回false，并且会根据mayInterruptIfRunning参数来决定是否会中断正在执行任务的线程。4. 小节
Executor框架中，Runnable接口和Callable接口表示需要被执行的任务，不同在于前者无返回值而后者有返回值。Future接口表示异步计算的结果，同时RunnableFuture接口继承了Runnable接口和Future接口，RunnableFuture接口的实现类FutureTask既能作为Runnable任务来执行，也能作为Future来获取计算结果。总结 【多线程学习-Executor框架】本篇文章对
Executor框架的主要组件进行了学习，主要分为任务，线程执行任务和任务执行结果三部分，具体学习了任务相关接口Runnable接口和Callable接口，线程执行任务相关组件ThreadPoolExecutor和ScheduledThreadPoolExecutor线程池，以及执行结果相关组件FutureTask。推荐阅读
- 放屁有这三个特征的，请注意啦！这说明你的身体毒素太多
- 爱就是希望你好好活着
- 昨夜小楼听风
- 知识
- 死结。
- 我从来不做坏事
- 由浅入深理解AOP
- 烦恼和幸福
- 继续努力，自主学习家庭Day135(20181015）
- 关于QueryWrapper|关于QueryWrapper,实现MybatisPlus多表关联查询方式