任务提交
之前在分析线程池的时候，提到过 AbstractExecutorService 的实现：
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
对于 submit 提交的任务，不管是 Runnable 还是 Callable，最终都会统一为 FutureTask 并传给 execute 方法。
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}对于 Runnable 还会创建一个适配器 ：
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;
}
}任务状态
FutureTask 有下面几种状态：
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;初次创建的时候构造器中赋值 state = NEW，后面状态可能有下面几种演化：
- NEW -> COMPLETING -> NORMAL （正常完成的过程）
- NEW -> COMPLETING -> EXCEPTIONAL （执行过程中遇到异常）
- NEW -> CANCELLED （执行前被取消）
- NEW -> INTERRUPTING -> INTERRUPTED （取消时被中断）
任务执行
当线程池执行任务的时候，最终都会执行 FutureTask 的 run 方法：
public void run() {
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 {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
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;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}对于 Callable 直接执行其 call 方法。执行成功则调用 set 方法设置结果，如果遇到异常则调用 setException 设置异常：
protected void set(V v) {
// 首先 CAS 设置 state 为中间状态 COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
// 设置为正常状态
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
// 设置为异常状态
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}这两个方法都是对全局变量 outcome 的赋值。当我们通过 get 方法获取结果时，往往是在另一个线程：
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}如果任务还没有完成则等待任务完成：
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
// 通过 for 循环来阻塞当前线程
for (;;) {
// 响应中断
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
// 任务已完成或者已抛出异常 直接返回
if (s > COMPLETING) {
// WaitNode已创建此时也没用了
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
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);
}
}如果任务已完成或者等待任务直到完成后，调用 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);
}如果 state == NORMAL，标识任务正常完成，返回实际结果。如果 state >= CANCELLED， 则返回 CancellationException，否则返回 ExecutionException，这样在线程池中执行的任务不管是异常还是正常返回了结果，都能被感知。
Treiber Stack
/**
* Simple linked list nodes to record waiting threads in a Treiber
* stack. See other classes such as Phaser and SynchronousQueue
* for more detailed explanation.
*/
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}在 awaitDone 方法中 WaitNode q = null，第一次会创建一个 WaitNode，这时即使有多个线程在等待结果，都会创建各自的 WaitNode：
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);然后在for循环中会跳到第二个 else if，由于没有入队，这时会通过 CAS 将新建的 WaitNode 类型的 q 赋值给 waiters，这个时候同一时刻只有一个线程能赋值成功，后一个在失败后又经历一次循环，最终成功地将当前 WaitNode 插入到 waiters 的头部。
任务取消
FutureTask 有一个 cancel 方法，包含一个 boolean 类型的参数（在执行中的任务是否可以中断）：
public boolean cancel(boolean mayInterruptIfRunning) {
// 如果任务不是刚创建或者是刚创建但是更改为指定状态失败则返回 false
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}最终都会调用 finishCompletion() ，在 set 方法和 setException 方法中也调用了这个 finishCompletion 方法：
private void finishCompletion() {
// assert state > COMPLETING;
// 如果任务执行完或者存在异常的话 这个waiters已经为null了
for (WaitNode q; (q = waiters) != null;) {
// 首先不断尝试把 waiters 设置为 null，如果很多线程调用 task.cancel()，也只有一个能成功
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
}当在 finishCompletion 方法中唤醒线程后，被唤醒的线程在 awaitDone 方法中继续循环，发现状态已完成：
int s = state;
// 任务已完成或者已抛出异常 直接返回
if (s > COMPLETING) {
// WaitNode已创建此时也没用了
if (q != null)
q.thread = null;
return s;
}接着调用 report 方法，发现状态为异常的话将包装成 ExecutionException((Throwable)x); 这个异常就是我们在使用 get 的时候需要捕获的异常。
最近比较忙，这块东西已经很久没有看了， FutureTask 感觉没有彻底弄明白，也没有一个好的结尾，现在这里标记下，后面继续更新。