FutureTask源码阅读
前言
在学习ThreadPoolExecutor咱们提到过submit提交的任务会被封装成FutureTask类型以后再放到线程池中执行，FutureTask类表明了异步执行的结果对象，用户可使用它来获取、查看和取消异步请求。java
public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
FutureTask类和Future、Callable、Runnable接口的对象关系以下图，RunnableFuture接口继承Runnable和Future接口，同时FutureTask实现了RunnableFuture接口，为了可以同时处理Runnable和Callable任务对象，FutureTask内部会将Callable对象做为成员封装起来。
web
简单使用
FutureTask内部封装了须要异步执行的代码和获取异步执行结果的逻辑，这些逻辑主要经过状态机模式来管理，状态之间的相互转化以下图所示，在详细解读这些状态转换过程以前咱们先来简单试用一下这个实现类。
下面的代码在第一个异步请求线程中先睡眠3秒，以后在返回一个字符串模拟网络异步请求，后面的两个线程都是用Future对象等待获取请求到的字符串数据。网络
// 获取异步数据的实现
Callable<String> callable = new Callable<String>() {
@Override
public String call() throws Exception {
Thread.sleep(3000);
return "Hello World";
}
};
// 将异步获取
FutureTask<String> task = new FutureTask<>(callable);
// 线程异步获取
new Thread(task).start();
// 等待结果线程1
Thread thread1 = new Thread(() -> { try { System.out.println(task.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }); // 等待结果线程2 Thread thread2 = new Thread(() -> { try { System.out.println(task.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }); thread1.start(); thread2.start();
若是只执行测试这些代码，最终会在控制台输出两行“Hello World”完成执行，这就是正常执行的结果。若是在后面加上取消前面的异步数据请求，两个等待请求结果的线程都会抛出CancellationException异常而且当即退出程序；若是使用task.cancel(false);程序不会当即退出而是会等到3秒过去以后才会彻底退出。多线程
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
task.cancel(true);
Exception in thread "Thread-2" java.util.concurrent.CancellationException
at java.util.concurrent.FutureTask.report(FutureTask.java:121)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at com.example.Main.lambda$main$1(Main.java:29)
at java.lang.Thread.run(Thread.java:745)
Exception in thread "Thread-1" java.util.concurrent.CancellationException
at java.util.concurrent.FutureTask.report(FutureTask.java:121)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at com.example.Main.lambda$main$0(Main.java:19)
at java.lang.Thread.run(Thread.java:745)
若是在最后加入thread1.interupt()打断第一个在等待结果的线程，会发现第一个等待线程抛出了异常，第二个线程正常等到异步返回的结果。异步
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
thread1.interrupt();
java.lang.InterruptedException
at java.util.concurrent.FutureTask.awaitDone(FutureTask.java:404)
at java.util.concurrent.FutureTask.get(FutureTask.java:191)
at com.example.Main.lambda$main$0(Main.java:19)
at java.lang.Thread.run(Thread.java:745)
Hello World
代码分析
为了可以搞清楚这些状态切换在什么状况下发生，须要查看FutureTask的内部实现源码，它的构造函数会将传递进来的Callable对象记录在成员变量里同时将当前状态设置为NEW。ide
// 保证在多线程状况的可见性
private volatile int state;
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
在配置好异步任务以后就要开始执行run方法，记住这个方法是在另一个线程中调用的。svg
public void run() {
// 若是状态不是NEW，或者当前任务已经有了其余执行线程
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);
}
}
查看正常返回结果和异常致使的执行逻辑，它们会先将状态设置成COMPLETING再设置FutureTask的成员属性outcome为返回结果，最后在把状态设置成NORMAL或者EXCEPTIONAL状态，finishCompletion执行的收尾动做，这两个方法的执行代码解释了前面NORMAL和EXCEPTIONAL状态转换过程。函数
protected void set(V v) {
// 先设置状态为COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
// 再设置状态为NORMAL
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
// 执行收尾动做
finishCompletion();
}
}
protected void setException(Throwable t) {
// 先设置状态为COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
// 再设置状态为EXCEPTIONAL
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
// 执行收尾动做
finishCompletion();
}
}
private void finishCompletion() {
// 遍历全部等待当前FutureTask执行结果的线程
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;
}
}
// 最后执行FutureTask的done方法
done();
callable = null; // to reduce footprint
}
前面执行的都是在请求数据的线程中，如今开始查看等待结果线程中执行的代码逻辑，主要是get方法的实现，能够看到在没有到达EXCEPTIONAL或者NORMAL状态都会一直执行awaitDone操做直到它返回才会调用report方法。学习
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);
}
接着查看awaitDone方法的实现逻辑会首先判断等待线程是否被中断，若是是就再也不等待抛出中断异常，若是等待的结果已经放置到outcome对象里再也不等待，其余将当前等待线程放到Future的等待线程队列中，前面的finishCompletion就是在执行完成异步请求后将这些等待线程所有唤醒。测试
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
// 若是正在等待结果的线程被终端，抛出终端异常
// 这里就解释了前面threadA.interrupt执行时第一个等待线程抛出异常，第二个等待线程正常执行完成
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
// 若是已经处于CANCELLED EXCEPTIONAL或者NORMAL
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s; // 没必要等待直接返回
}
else if (s == COMPLETING) // 若是设置值还没完成，继续等待
Thread.yield();
else if (q == null) // 若是刚开始等待，建立WaitNode
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方法，它作的就是返回异步请求中最终返回的结果，若是异步请求正常结束就返回结果，若是当前请求被取消就抛出CancellationException，不然就是异步请求发生异常抛出ExecutionException。
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的状况，它的参数代表若是任务正在执行是否中断，若是为false表示不执行中断，不然就会对异步请求线程执行interrupt方法，cancel方法最后一样会执行finishCompletion方法。
public boolean cancel(boolean mayInterruptIfRunning) {
// 若是正处于NEW状态，但愿请求正在运行也但愿中断就设置为INTERRUTPTING，不然直接设置CANCELLED
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
// 但愿正执行异步请求也打断，中断执行线程
// 这就解释了直接cancel（true）前面两个等待线程直接退出，由于sleep方法执行时
// 发生中断会当即退出sleep，因此后面的请求就当即结束了，而cancel为false是只是设置
// futureTask为cancel状态，执行线程还在继续执行，最后返回结果时产生
// CancellationException
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
// 设置状态为INTERRUPTED
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
// 执行唤醒全部等待线程操做
finishCompletion();
}
return true;
}