JAVA 并发 编程系列 （三）线程池 介绍 、基础原理 Cal
先来看看 线程池框架的 继承结构:
继承图
public interface Executor {
/**
*Executor，任务的执行者，线程池框架中几乎所有类都直接或者间接实 现Executor接口，它是线程池框架的基础。
*Executor提供了一种将“任务提交”与“任务执行”分离开来的机制，
*它仅提供了一个Execute()方法用来执行已经提交的Runnable任务。
*/
void execute(Runnable command);
}
public interface ExecutorService extends Executor {
/**
* 启动一次顺序关闭，执行以前提交的任务，但不接受新任务
*/
void shutdown();
/**
* 试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表
*/
List<Runnable> shutdownNow();
/**
* 试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表
*/
boolean isShutdown();
/**
* 试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表
*/
boolean isTerminated();
/**
*接收人timeout和TimeUnit两个参数，用于设定超时时间及单位。
*当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false。
*一般情况下会和shutdown方法组合使用
*/
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
*/
Future<?> submit(Runnable task);
/**
* 执行给定的任务，返回一个在所有完成时保持其状态和结果的未来列表。
*注意，任务可以正常终止，也可以抛出异常。
*如果在此操作正在进行时对给定的集合进行修改，则该方法的结果是未定义的。
*
*/
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
/**
* 执行给定的任务，当所有任务完成或超时期满时
*（无论哪个首先发生），返回保持任务状态和结果的 Future 列表
*/
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
/**
*提交的任务集合,一旦有1个任务正常完成(没有抛出异常),会终止其他未完成的任务
*/
<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;
}
有关submit 和 invokeAll的方法实例的写在这里
Future 接口继承图
Future
Future接口
作为异步计算的顶层接口，Future对具体的Runnable或者Callable任务提供了三种操作：执行任务的取消、查询任务是否完成、获取任务的执行结果。其接口定义如下：
*/
public interface Future<V> {
/**
* 试图取消对此任务的执行
* 如果任务已完成、或已取消，或者由于某些其他原因而无法取消，则此尝试将失败。
* 当调用 cancel 时，如果调用成功，而此任务尚未启动，则此任务将永不运行。
* 如果任务已经启动，则 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;
}
FutureTask
实现RunnableFuture接口，既可以作为Runnable被执行，也可以作为Future得到Callable的返回值。
构造函数
- 先从FutureTask的构造函数看起，FutureTask有两个构造函数，其中一个如下：
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
这个构造函数会把传入的Callable变量保存在this.callable字段中，该字段定义为private Callable<V> callable;用来保存底层的调用，在被执行完成以后会指向null,接着会初始化state字段为NEW。state字段用来保存FutureTask内部的任务执行状态，一共有7中状态，每种状态及其对应的值如下：
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;
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
为了后面更好的分析FutureTask的实现，这里有必要解释下各个状态。
-
NEW:表示是个新的任务或者还没被执行完的任务。这是初始状态。
-
COMPLETING:任务已经执行完成或者执行任务的时候发生异常，但是任务执行结果或者异常原因还没有保存到outcome字段(outcome字段用来保存任务执行结果，如果发生异常，则用来保存异常原因)的时候，状态会从NEW变更到CO- MPLETING。但是这个状态会时间会比较短，属于中间状态。
-
NORMAL:任务已经执行完成并且任务执行结果已经保存到outcome字段，状态会从COMPLETING转换到NORMAL。这是一个最终态。
-
EXCEPTIONAL:任务执行发生异常并且异常原因已经保存到outcome字段中后，状态会从COMPLETING转换到EXCEPTIONAL。这是一个最终态。
-
CANCELLED:任务还没开始执行或者已经开始执行但是还没有执行完成的时候，用户调用了cancel(false)方法取消任务且不中断任务执行线程，这个时候状态会从NEW转化为CANCELLED状态。这是一个最终态。
-
INTERRUPTING: 任务还没开始执行或者已经执行但是还没有执行完成的时候，用户调用了cancel(true)方法取消任务并且要中断任务执行线程但是还没有中断任务执行线程之前，状态会从NEW转化为INTERRUPTING。这是一个中间状态。
-
INTERRUPTED:调用interrupt()中断任务执行线程之后状态会从INTERRUPTING转换到INTERRUPTED。这是一个最终态。
有一点需要注意的是，所有值大于COMPLETING的状态都表示任务已经执行完成(任务正常执行完成，任务执行异常或者任务被取消)。
状态转换图
另一个 构造函数：
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
//这个构造函数会把传入的Runnable封装成一个Callable对象保存在callable字段中，
//同时如果任务执行成功的话就会返回传入的result。
//这种情况下如果不需要返回值的话可以传入一个null。
//顺带看下Executors.callable()这个方法，
//这个方法的功能是把Runnable转换成Callable，代码如下:
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
//可以看到这里采用的是适配器模式，调用RunnableAdapter<T>(task, result)方法来适配，实现如下:
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;
}
}
这个适配器很简单，就是简单的实现了Callable接口，在call()实现中调用Runnable.run()方法，然后把传入的result作为任务的结果返回。
在new了一个FutureTask对象之后，接下来就是在另一个线程中执行这个Task,无论是通过直接new一个Thread还是通过线程池，执行的都是run()方法，接下来就看看
run()方法的实现。
public void run() {
// 1. 状态如果不是NEW，说明任务或者已经执行过，或者已经被取消，直接返回
// 2. 状态如果是NEW，则尝试把当前执行线程保存在runner字段中
// 如果赋值失败则直接返回
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
// 3. 执行任务
result = c.call();
ran = true;
} catch (Throwable ex) {
//4.执行任务失败
result = null;
ran = false;
//将异常保存至 outcome字段
setException(ex);
}
if (ran)//如果成功保存结果字段
set(result);
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
// 5. 如果任务被中断，执行中断处理
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
run()方法首先会
- 判断当前任务的state是否等于NEW,如果不为NEW则说明任务或者已经执行过，或者已经被取消，直接返回。
- 如果状态为NEW则接着会通过unsafe类把任务执行线程引用CAS的保存在runner字段中，如果保存失败，则直接返回。
- 执行任务。
- 如果任务执行发生异常，则调用setException()方法保存异常信息。
setException()方法如下：
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
setException（）
- 首先会CAS的把当前的状态从NEW变更为COMPLETING状态。
- 把异常原因保存在outcome字段中，outcome字段用来保存任务执行结果或者异常原因。
- CAS的把当前任务状态从COMPLETING变更为EXCEPTIONAL。这个状态转换对应着上图中的
- 调用finishCompletion()。关于这个方法后面在分析。
任务成功调用
set(V v)方法 如下：
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
这个方法跟上面分析的setException()差不多，
- 首先会CAS的把当前的状态从NEW变更为COMPLETING状态。
把任务执行结果保存在outcome字段中。 - CAS的把当前任务状态从COMPLETING变更为NORMAL。这个状态转换对应着上图中的一。
- 调用finishCompletion()。
get()
任务发起线程可以调用get()方法来获取任务执行结果，如果此时任务已经执行完毕则会直接返回任务结果，如果任务还没执行完毕，则调用方会阻塞直到任务执行结束返回结果为止。get()方法实现如下:
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
get()方法实现比较简单，
- 会判断任务当前的state <= COMPLETING是否成立。前面分析过，COMPLETING状态是任务是否执行完成的临界状态。
- 如果成立，表明任务还没有结束(这里的结束包括任务正常执行完毕，任务执行异常，任务被取消)，则会调用awaitDone()进行阻塞等待。
- 如果不成立表明任务已经结束，调用report()返回结果。
awaitDone()
当调用get()获取任务结果但是任务还没执行完成的时候，调用线程会调用awaitDone()方法进行阻塞等待，该方法定义如下:
/**
* Awaits completion or aborts on interrupt or timeout.
*
* @param timed true if use timed waits
* @param nanos time to wait, if timed
* @return state upon completion
*/
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
// 计算等待截止时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
//1.判断当前线程是否阻塞,
//如果被中断则在等待队列中删除该节点并抛出InterruptedException异常
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
// 2. 获取当前状态，如果状态大于COMPLETING
// 说明任务已经结束(要么正常结束，要么异常结束，要么被取消)
// 则把thread显示置空，并返回结果
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
// 3. 如果状态处于中间状态COMPLETING
// 表示任务已经结束但是任务执行线程还没来得及给outcome赋值
// 这个时候让出执行权让其他线程优先执行
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
// 4. 如果等待节点为空，则构造一个等待节点
else if (q == null)
q = new WaitNode();
// 5. 如果还没有入队列，则把当前节点加入waiters首节点并替换原来waiters
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
// 如果需要等待特定时间，则先计算要等待的时间
// 如果已经超时，则删除对应节点并返回对应的状态
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
- awaitDone()中有个死循环，每一次循环都会判断调用get()的线程是否被其他线程中断，如果是的话则在等待队列中删除对应节点然后抛出InterruptedException异常。
- 获取任务当前状态，如果当前任务状态大于COMPLETING则表示任务执行完成，则把thread字段置null并返回结果。
- 如果任务处于COMPLETING状态，则表示任务已经处理完成(正常执行完成或者执行出现异常)，但是执行结果或者异常原因还没有保存到outcome字段中。这个时候调用线程让出执行权让其他线程优先执行。
- 如果等待节点为空，则构造一个等待节点WaitNode。
- 如果第四步中新建的节点还没如队列，则CAS的把该节点加入waiters队列的首节点。
- 阻塞等待。
假设当前state=NEW且waiters为NULL,也就是说还没有任何一个线程调用get()获取执行结果，这个时候有两个线程threadA和threadB先后调用get()来获取执行结果。再假设这两个线程在加入阻塞队列进行阻塞等待之前任务都没有执行完成且threadA和threadB都没有被中断的情况下(因为如果threadA和threadB在进行阻塞等待结果之前任务就执行完成或线程本身被中断的话，awaitDone()就执行结束返回了)，执行过程是这样的，以threadA为例:
- 第一轮for循环，执行的逻辑是q == null,所以这时候会新建一个节点q。第一轮循环结束。
- 第二轮for循环，执行的逻辑是!queue，这个时候会把第一轮循环中生成的节点的netx指针指向waiters，然后CAS的把节点q替换waiters。也就是把新生成的节点添加到waiters链表的首节点。如果替换成功，queued=true。第二轮循环结束。
- 第三轮for循环，进行阻塞等待。要么阻塞特定时间，要么一直阻塞知道被其他线程唤醒。
在threadA和threadB都阻塞等待之后的waiters结果如图
[图片上传失败...(image-d8c828-1513762765353)]
cancel(boolean)
public boolean cancel(boolean mayInterruptIfRunning) {
//1.传入 false 一般意思是 取消还没开始的任务设置成CANCELLED
//2.传入 true 一般意思是取消还没开始的任务设置成INTERRUPTING
if (!(state == NEW &&// 1. 如果任务已经结束，则直接返回false
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
//还没开始 设置CANCELLEN 开始了设置 失败返回false
return false;
try { // in case call to interrupt throws exception
//2.传入true
if (mayInterruptIfRunning) {
try {
Thread t = runner;
// 2.1 中断任务执行线程
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
cancel()方法会做下面几件事:
1 .判断任务当前执行状态，如果任务状态不为NEW，则说明任务或者已经执行完成，或者执行异常，不能被取消，根据传入的情况设置INTERRUPTING，或CANCELLED。
-
判断需要中断任务执行线程，则
修改任务状态为INTERRUPTED。这个转换过程对应上图中的四。 -
调用finishCompletion()。
finishCompletion()
根据前面的分析，不管是任务执行异常还是任务正常执行完毕，或者取消任务，最后都会调用finishCompletion()方法，该方法实现如下:
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
这个方法的实现比较简单，依次遍历waiters链表，唤醒节点中的线程，然后把callable置空。
被唤醒的线程会各自从awaitDone()方法中的LockSupport.park*()阻塞中返回，然后会进行新一轮的循环。在新一轮的循环中会返回执行结果(或者更确切的说是返回任务的状态)。
report()
report()方法用在get()方法中，作用是把不同的任务状态映射成任务执行结果。实现如下：
private V report(int s) throws ExecutionException {
Object x = outcome;
// 1. 任务正常执行完成，返回任务执行结果
if (s == NORMAL)
return (V)x;
// 2. 任务被取消，抛出CancellationException异常
if (s >= CANCELLED)
throw new CancellationException();
// 3. 其他状态，抛出执行异常ExecutionException
throw new ExecutionException((Throwable)x);
}
映射关系如下图所示:
[图片上传失败...(image-a3c07e-1513762765353)]
get(long,TimeUnit)
public V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
if (unit == null)
throw new NullPointerException();
int s = state;
if (s <= COMPLETING &&
// 如果awaitDone()超时返回之后任务还没结束，则抛出异常
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
return report(s);
}
跟get()不同点在于get(long,TimeUnit)会在awaitDone()超时返回之后抛出TimeoutException异常。
isCancelled()和isDone()
vpublic boolean isCancelled() {
return state >= CANCELLED;
}
public boolean isDone() {
return state != NEW;
}
根据前面的分析，这两个方法就很容易理解不用多做解释了。
总结下，其实FutureTask的实现还是比较简单的，当用户实现Callable()接口定义好任务之后，把任务交给其他线程进行执行。FutureTask内部维护一个任务状态，任何操作都是围绕着这个状态进行，并随时更新任务状态。任务发起者调用get*()获取执行结果的时候，如果任务还没有执行完毕，则会把自己放入阻塞队列中然后进行阻塞等待。当任务执行完成之后，任务执行线程会依次唤醒阻塞等待的线程。调用cancel()取消任务的时候也只是简单的修改任务状态，如果需要中断任务执行线程的话则调用Thread.interrupt()中断任务执行线程。
有个值得关注的问题就是当任务还在执行的时候用户调用cancel(true)方法能否真正让任务停止执行呢？
在前面的分析中我们直到，当调用cancel(true)方法的时候，实际执行还是Thread.interrupt()方法，而interrupt()方法只是设置中断标志位，如果被中断的线程处于sleep()、wait()或者join()逻辑中则会抛出InterruptedException异常。
