Java并发包FutureTask详解
我们已经知道了所有提交给jdk线程池的任务都会被封装成一个FutureTask对象。线程池执行的其实是FutureTask中的run方法。
类图
image.png
可以看到FutureTask实现了Future和Runnable两个接口，说明它既可以当做任务提交给线程池，也可作为Future查询任务执行情况或者是取消任务。
成员变量
/* Possible state transitions:
* 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;
这是用于定义任务状态的变量。
- 0-初始
- 1-执行中
- 2-已完成
- 3-异常
- 4-已取消
- 5-中断中
- 6-被中断
/** The underlying callable; nulled out after running */
private Callable<V> callable;
/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes
/** The thread running the callable; CASed during run() */
private volatile Thread runner;
/** Treiber stack of waiting threads */
private volatile WaitNode waiters;
- callable-就是我们传递进来的原始的任务
- outcome-任务执行的结果
- runner-执行该任务的线程
- waiters-等待任务完成的线程
构造函数
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
}
FutureTask一共有两个构造函数，第一个构造函数接收一个Callable对象，第二个构造函数接收一个Runnable对象和一个泛型对象result，这个构造函数中调用Executors的callable方法将Runnable对象包装成一个Callable对象。
Executors的callable方法核心如下：
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;
}
}
这是将Runnable转化为Callable对象的过程，其实很简单，由于Runnable并没有返回值，而Callable需要返回值，因此就直接拿我们传递的result作为返回值了。
注意到，构造函数中将任务的状态置为NEW。
成员函数
public boolean isCancelled() {
return state >= CANCELLED;
}
public boolean isDone() {
return state != NEW;
}
第一个方法判断任务是否已经被取消了。可以看到4，5，6状态均视为已取消。
第二个方法判断任务是否已经完成。只要不是初始状态都视为已完成。
public boolean cancel(boolean mayInterruptIfRunning) {
//如果state==NEW，说明任务还没开始，此时只需要根据传递的参数将其状态置为中断中或者已取消即可
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;
}
该方法用于取消任务。参数的意义是如果任务已经开始，是否尝试中断。
/**
* Removes and signals all waiting threads, invokes done(), and
* nulls out callable.
*/
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
}
唤醒（unpark方法）所有等待该任务的线程，并从等待列表中移除。最后调用了done方法。该方法在FutureTask为空，主要目的是便于子类定制自己的行为。接着将任务置为null。
再来看看最常用的get方法。
public V get() throws InterruptedException, ExecutionException {
int s = state;
//此时任务尚未完成，进入到等待中
if (s <= COMPLETING)
s = awaitDone(false, 0L);
//
return report(s);
}
/**
* @throws CancellationException {@inheritDoc}
*/
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);
}
/**
* 该方法只用于获取已经正常完成的任务的返回值 其他情况都会抛异常
* Returns result or throws exception for completed task.
*
* @param s completed state value
*/
@SuppressWarnings("unchecked")
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);
}
第一个get方法会一直阻塞到任务完成。第二个get方法会至多阻塞指定的时长。来看看两个get中都调用的一个方法。
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()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
// 如果任务已经结束了 则返回当前状态
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
//如果任务在进行中(状态马上就会变)，则让出cpu，等待下次cpu时间片，暂时不要time out
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) {
//如果是设置了时间限制，则计算到时间了没，由于前面让出了CPU时间片，所以有可能再次执行时已经过点了
nanos = deadline - System.nanoTime();
//到点后还没执行完则移除q之后的所有等待线程，并返回当前状态
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
//挂起当前线程
LockSupport.parkNanos(this, nanos);
}
else
//挂起当前线程
LockSupport.park(this);
}
}
再来看看核心方法
public void run() {
// 前面半句是判断任务的状态是不是初始状态，只有初始状态的任务才能执行run
// 后面半句是判断执行该任务的线程是否为空，如果不为空则将当前线程赋值给runner
// 如果任务不为初始状态 或者 已经有指定的执行线程了 就直接return，这说明已经有线程在执行该任务了
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);
}
//正常完成情况下 调用set方法
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);
}
}
protected void setException(Throwable t) {
//先将任务状态置为完成中
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//将任务的返回值置为异常对象
outcome = t;
//将任务置为异常状态
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
//唤醒所有等待线程
finishCompletion();
}
}
/** 保证所有的强制中断只下发到正在执行中或者是重置后的任务中
* Ensures that any interrupt from a possible cancel(true) is only
* delivered to a task while in run or runAndReset.
*/
private void handlePossibleCancellationInterrupt(int s) {
// It is possible for our interrupter to stall before getting a
// chance to interrupt us. Let's spin-wait patiently.
if (s == INTERRUPTING)
//如果任务状态是中断中，则一直等待
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
// assert state == INTERRUPTED;
// We want to clear any interrupt we may have received from
// cancel(true). However, it is permissible to use interrupts
// as an independent mechanism for a task to communicate with
// its caller, and there is no way to clear only the
// cancellation interrupt.
//
// Thread.interrupted();
}
FutureTask还提供了一个可复用自身的方法。我们注意到run方法中会判断任务是否是初始状态，如果不是则不予执行。这意味着一个已经完成的任务是不可能再次被执行的。而FutureTask的runAndReset方法则是让任务实现复用。
protected boolean runAndReset() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return false;
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 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
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;
}
可以看到几乎和run方法一样，但是不同点在于任务执行完毕后并没有调用set方法，即并没有改变任务的状态，也没有将任务的执行结果保存在outcome中。方法仅在任务成功执行且为初始状态时返回true。
最后是该类中的一个内部类
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
其实就是一个单向链表，保存等待中的线程。
FutureTask中最复杂的方法
/**
* Tries to unlink a timed-out or interrupted wait node to avoid
* accumulating garbage. Internal nodes are simply unspliced
* without CAS since it is harmless if they are traversed anyway
* by releasers. To avoid effects of unsplicing from already
* removed nodes, the list is retraversed in case of an apparent
* race. This is slow when there are a lot of nodes, but we don't
* expect lists to be long enough to outweigh higher-overhead
* schemes.
*/
private void removeWaiter(WaitNode node) {
if (node != null) {
node.thread = null;
retry:
for (;;) { // restart on removeWaiter race
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
s = q.next;
if (q.thread != null)
pred = q;
else if (pred != null) {
pred.next = s;
if (pred.thread == null) // check for race
continue retry;
}
else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
q, s))
continue retry;
}
break;
}
}
}
该方法旨在并发情况下能清除掉所有等待某个FutureTask完成的线程节点。
FutureTask全程没有加锁，全部是调用Unsafe类的cas方法来实现的。
总结
FutureTask将原始的任务（Callable或者Runnable）封装了一层，加入了状态位，并且自己维护状态。最终提交到线程池的对象其实是一个FutureTask，由于FutureTask实现了Runnable接口，因此线程池执行的其实是FutureTask的run方法。任务的状态变更都是FutureTask自己完成的，线程池对于FutureTask内部的状态一无所知。一个FutureTask在执行前被取消，并不意味着线程池不再派出线程去执行FutureTask，线程池照样会派出线程去执行该FutureTask，只是在执行FutureTask的run方法时，FutureTask判断自己已经被取消了，就直接return了，不再执行包在其中的原始任务了。
