FutureTask源码分析

源码分析"/>

FutureTask源码分析

概念

同步和异步怎么理解？

其实同步和异步最本质的区别就是，同步是主动询问结果，而异步是被动通知获取结果。

Future是什么？

其实Future就是异步任务执行后返回的结果，Java创建线程一般是Thread/Runnable，涉及到返回结果的，就需要用Callable，Callable运行需要结合Future一起使用。Future只是一个接口类，定义了获取结果、检查状态和取消任务的接口。

FutureTask是什么？

FutureTask实现了RunnableFuture，而RunnableFuture又继承了Runnable，当线程池执行任务时，执行的是Runnable的run方法，当FutureTask被放入线程池后，可以被作为Runnable任务来执行它的run方法，同时FutureTask的run方法真正执行的是Callable的call方法，所以，FutureTask其实是一个存放任务返回结果的类。

属性

分析FutureTask，先从它的核心属性开始

/*** 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;  // Callable的结果（正常结果，异常结果）正在封装给当前的FutureTask
private static final int NORMAL       = 2;  // NORMAL任务正常结束
private static final int EXCEPTIONAL  = 3;  // 执行任务时，发生了异常
private static final int CANCELLED    = 4;  // 任务被取消了。
private static final int INTERRUPTING = 5;  // 线程的中断状态，被设置为了true（现在还在运行）
private static final int INTERRUPTED  = 6;  // 线程被中断了。// 当前要执行的任务
private Callable<V> callable;
// 存放任务返回结果的属性，也就是futureTask.get需要获取的结果
private Object outcome; 
// 执行任务的线程。
private volatile Thread runner;
// 单向链表，存放通过get方法挂起等待的线程
private volatile WaitNode waiters;

run方法

线程任务start启动后，通过run方法执行Callable的call方法

// run方法的执行流程，最终会执行Callable的call方法
public void run() {// 任务的状态必须是NEW才可以运行，通过CAS方式，成功设置runner为当前执行线程，那么FutureTask就有了操作当前执行线程的引用对象，并为后续cancel方法中对线程执行中断操作提供基础if (state != NEW ||!UNSAFEpareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return;// 开始执行任务try {// 要执行任务 callableCallable<V> c = callable;// 任务不为null，并且任务的状态是否被改变，必须还处于NEW，否则无法执行if (c != null && state == NEW) {// 存放返回结果V result;// 任务执行是否为正常结束boolean ran;try {// 执行call方法（call才是用户定义的真正任务主体方法），拿到返回结果设置到result中result = c.call();// 正常返回，ran设置为trueran = true;} catch (Throwable ex) {//捕捉用户自定义的方法执行异常结果result = null;// 执行失败，异常返回，ran设置为falseran = false;// 设置执行失败的结果（异常信息）setException(ex);}if (ran)// 执行成功，正常结束，设置返回结果set(result);}} finally {// 执行完毕（成功/失败），将执行任务的runner设置空，释放对执行线程的引用，方便gcrunner = null;// 判断状态是否被中断，如果是，则调用handlePossibleCancellationInterrupt处理后续中断操作int s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);}
}// 当任务执行成功时，设置返回结果，与setException的实现一样，只是状态不一样
protected void set(V v) {// 首先要将任务状态从NEW设置为COMPLETINGif (UNSAFEpareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {// 将返回结果设置给outcome。outcome = v;// 将状态修改为NORMAL，代表正常结束UNSAFE.putOrderedInt(this, stateOffset, NORMAL);// 完成后唤醒链表中其他挂起线程finishCompletion();}
}
//当任务执行失败并抛出异常时，调用该方法，将FutureTask状态变为异常
protected void setException(Throwable t) {if (UNSAFEpareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome = t;//保证不发生指令重排序，即上下代码之间执行顺序同编写顺序UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state// 完成后唤醒链表中其他挂起线程finishCompletion();}
}
//当正在执行的线程被其他线程调用cancel方法且标记指定了中断线程时，调用该方法
private void handlePossibleCancellationInterrupt(int s) {//如果状态为INTERRUPTING，将循环等待状态为INTERRUPTEDif (s == INTERRUPTING)while (state == INTERRUPTING)Thread.yield();  //让出cpu
}

get方法

get方法获取返回结果，到挂起的位置

public V get() throws InterruptedException, ExecutionException {// 拿状态int s = state;// 满足找个状态就代表现在可能还没有返回结果if (s <= COMPLETING)// 尝试挂起线程，等待拿结果s = awaitDone(false, 0L);return report(s);
}// 线程要等待任务执行结束，等待任务执行的状态变为大于COMPLETING状态
private int awaitDone(boolean timed, long nanos) throws InterruptedException {// 计算deadline，如果是get()，就是0，  如果是get(time,unit)那就追加当前系统时间final long deadline = timed ? System.nanoTime() + nanos : 0L;// 构建WaitNodeWaitNode q = null;// queued = falseboolean queued = false;// 死循环for (;;) {// 找个get的线程是否中断了。if (Thread.interrupted()) {// 将当前节点从waiters中移除。removeWaiter(q);// 并且抛出中断异常throw new InterruptedException();}// 拿到现在任务的状态int s = state;// 判断任务是否已经执行结束了if (s > COMPLETING) {// 如果设置过WaitNode，直接移除WaitNode的线程if (q != null)q.thread = null;// 返回当前任务的状态return s;}// 如果任务的状态处于 COMPLETING ，else if (s == COMPLETING)// COMPLETING的持续时间非常短，只需要做一手现成的让步即可。Thread.yield();// 现在线程的状态是NEW，（call方法可能还没执行完呢，准备挂起线程）else if (q == null)// 封装WaitNode存放当前线程q = new WaitNode();else if (!queued)// 如果WaitNode还没有排在waiters中，现在就排进来（头插法的效果）queued = UNSAFEpareAndSwapObject(this, waitersOffset, q.next = waiters, q);else if (timed) {// get(time,unit)挂起线程的方式// 计算挂起时间nanos = deadline - System.nanoTime();// 挂起的时间，是否小于等于0if (nanos <= 0L) {// 移除waiters中的当前NoderemoveWaiter(q);// 返回任务状态return state;}// 正常指定挂起时间即可。（线程挂起）LockSupport.parkNanos(this, nanos);}else {// get()挂起线程的方式LockSupport.park(this);}}
}

finishCompletion方法

线程挂起后，如果任务执行完毕，由finishCompletion唤醒线程

// 任务状态已经变为了NORMAL，做一些后续处理
private void finishCompletion() {for (WaitNode q; (q = waiters) != null;) {// 拿到第一个节点后，直接用CAS的方式，将其设置为null，并将链路上的其他等待线程唤醒if (UNSAFEpareAndSwapObject(this, waitersOffset, q, null)) {for (;;) {// 拿到线程信息Thread t = q.thread;// 线程不为nullif (t != null) {// 将WaitNode的thread设置为null，释放资源q.thread = null;// 唤醒这个线程LockSupport.unpark(t);}// 往后遍历，接着唤醒WaitNode next = q.next;if (next == null)break;//将该节点断开链表，方便gcq.next = null;// 指向链表中下一个的WaitNode节点q = next;}break;}}// 钩子函数，留待子类实现done();// 任务处理完了！callable = null;   
}

cancel方法

任务取消方法

//通过调用该方法来取消任务的执行。此时，任务可能处于以下状态：1、new状态；2、执行中；3、已完成
public boolean cancel(boolean mayInterruptIfRunning) {//如果任务已经完成，直接返回false,此时可能完成中，也可能已经完成if (!(state == NEW &&UNSAFEpareAndSwapInt(this, stateOffset, NEW,mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))return false;try {    // 指定了中断线程if (mayInterruptIfRunning) {try {Thread t = runner;if (t != null)t.interrupt();  //直接中断} finally { // 最终完成状态设置为中断结束UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);}}} finally {//最后还要完成任务并唤醒等待线程finishCompletion();}return true;
}

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);
}