探索 JUC 之美---Future 与 FutureTask
本文共 12144 字，大约阅读时间需要 40 分钟。
Future
Future 表示一个任务的生命周期，是一个可取消的异步运算。提供了相应的方法来判断任务状态（完成或取消），以及获取任务的结果和取消任务等。
适合具有可取消性和执行时间较长的异步任务。 在并发包中许多异步任务类都继承自Future，其中最典型的就是 FutureTask
介绍
JavaDoc
get方法来获取结果，如有必要，计算完成前可以阻塞此方法。 取消则由 cancel 方法来执行。 还提供了其他方法，以确定任务是正常完成还是被取消了。 一旦计算完成，就不能再取消计算。 如果为了可取消性而使用 Future 但又不提供可用的结果，则可以声明 Future<?> 形式类型、并返回 null 作为底层任务的结果。 也就是说Future具有这样的特性
- 异步执行，可用 get 方法获取执行结果
- 如果计算还没完成，get 方法是会阻塞的，如果完成了，是可以多次获取并立即得到结果的
- 如果计算还没完成，是可以取消计算的
- 可以查询计算的执行状态
FutureTask
FutureTask 为 Future 提供了基础实现，如获取任务执行结果（get）和取消任务（cancel）等。如果任务尚未完成，获取任务执行结果时将会阻塞。一旦执行结束，任务就不能被重启或取消（除非使用
runAndReset执行计算）。FutureTask 常用来封装 Callable 和 Runnable，也可以作为一个任务提交到线程池中执行。除了作为一个独立的类之外，此类也提供了一些功能性函数供我们创建自定义 task 类使用。FutureTask 的线程安全由CAS来保证。
FutureTask 内部维护了一个由volatile修饰的int型变量—state，代表当前任务的运行状态
- NEW：新建
- COMPLETING：完成
- NORMAL：正常运行
- EXCEPTIONAL：异常退出
- CANCELLED：任务取消
- INTERRUPTING：线程中断中
- INTERRUPTED：线程已中断
在这七种状态中，有四种任务终止状态：NORMAL、EXCEPTIONAL、CANCELLED、INTERRUPTED。各种状态的转化如下：
数据结构及核心参数
//内部持有的callable任务，运行完毕后置空private Callablecallable;//从get()中返回的结果或抛出的异常private Object outcome; // non-volatile, protected by state reads/writes//运行callable的线程,在 run 时进行 CAS 操作private volatile Thread runner;//使用Treiber栈保存等待线程private volatile WaitNode waiters;
FutureTask 继承了Runnale和Future，本身也作为一个线程运行,可以提交给线程池执行。
WaitNode，使用简单的Treiber栈（无锁并发栈）实现，用于存储等待线程。 FutureTask 只有一个自定义的同步器 Sync 的属性，所有的方法都是委派给此同步器来实现。这也是JUC里使用AQS的通用模式。 源码解析
FutureTask 的同步器
由于Future在任务完成后，可以多次自由获取结果，因此，用于控制同步的AQS使用共享模式。FutureTask 底层任务的执行状态保存在AQS的状态里。AQS是否允许线程获取（是否阻塞）是取决于任务是否执行完成，而不是具体的状态值。
private final class Sync extends AbstractQueuedSynchronizer { // 定义表示任务执行状态的常量。由于使用了位运算进行判断，所以状态值分别是2的幂。 // 表示任务已经准备好了，可以执行 private static final int READY = 0; // 表示任务正在执行中 private static final int RUNNING = 1; // 表示任务已执行完成 private static final int RAN = 2; // 表示任务已取消 private static final int CANCELLED = 4; // 底层的表示任务的可执行对象 private final Callable callable; // 表示任务执行结果，用于get方法返回。 private V result; // 表示任务执行中的异常，用于get方法调用时抛出。 private Throwable exception; /* * 用于执行任务的线程。在 set/cancel 方法后置为空，表示结果可获取。 * 必须是 volatile的，用于确保完成后（result和exception）的可见性。 * （如果runner不是volatile，则result和exception必须都是volatile的） */ private volatile Thread runner; /** * 已完成或已取消 时成功获取 */ protected int tryAcquireShared( int ignore) { return innerIsDone() ? 1 : -1; } /** * 在设置最终完成状态后让AQS总是通知，通过设置runner线程为空。 * 这个方法并没有更新AQS的state属性， * 所以可见性是通过对volatile的runner的写来保证的。 */ protected boolean tryReleaseShared( int ignore) { runner = null; return true; } // 执行任务的方法 void innerRun() { // 用于确保任务不会重复执行 if (!compareAndSetState(READY, RUNNING)) return; // 由于Future一般是异步执行，所以runner一般是线程池里的线程。 runner = Thread.currentThread(); // 设置执行线程后再次检查，在执行前检查是否被异步取消 // 由于前面的CAS已把状态设置RUNNING， if (getState() == RUNNING) { // recheck after setting thread V result; // try { result = callable.call(); } catch (Throwable ex) { // 捕获任务执行过程中抛出的所有异常 setException(ex); return; } set(result); } else { // 释放等待的线程 releaseShared(0); // cancel } } // 设置结果 void innerSet(V v) { // 放在循环里进行是为了失败后重试。 for (;;) { // AQS初始化时，状态值默认是 0，对应这里也就是 READY 状态。 int s = getState(); // 已完成任务不能设置结果 if (s == RAN) return; // 已取消 的任务不能设置结果 if (s == CANCELLED) { // releaseShared 会设置runner为空， // 这是考虑到与其他的取消请求线程 竞争中断 runner releaseShared(0); return; } // 先设置已完成，免得多次设置 if (compareAndSetState(s, RAN)) { result = v; releaseShared(0); // 此方法会更新 runner，保证result的可见性 done(); return; } } } // 获取异步计算的结果 V innerGet() throws InterruptedException, ExecutionException { acquireSharedInterruptibly(0);// 获取共享，如果没有完成则会阻塞。 // 检查是否被取消 if (getState() == CANCELLED) throw new CancellationException(); // 异步计算过程中出现异常 if (exception != null) throw new ExecutionException(exception); return result; } // 取消执行任务 boolean innerCancel( boolean mayInterruptIfRunning) { for (;;) { int s = getState(); // 已完成或已取消的任务不能再次取消 if (ranOrCancelled(s)) return false; // 任务处于 READY 或 RUNNING if (compareAndSetState(s, CANCELLED)) break; } // 任务取消后，中断执行线程 if (mayInterruptIfRunning) { Thread r = runner; if (r != null) r.interrupt(); } releaseShared(0); // 释放等待的访问结果的线程 done(); return true; } /** * 检查任务是否处于完成或取消状态 */ private boolean ranOrCancelled( int state) { return (state & (RAN | CANCELLED)) != 0; } // 其他方法省略} 从 innerCancel 方法可知，取消操作只是改变了任务对象的状态并可能会中断执行线程。如果任务的逻辑代码没有响应中断，则会一直异步执行直到完成，只是最终的执行结果不会被通过get方法返回，计算资源的开销仍然是存在的。
总的来说，Future 是线程间协调的一种工具。
AbstractExecutorService.submit(Callable<T> task)
FutureTask 内部的实现方法都很简单，这里我们先从线程池的submit方法开始分析，一步一步分析FutureTask。submit方法默认实现在AbstractExecutorService中，几种实现源码如下：
public Future submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture ftask = newTaskFor(task, null); execute(ftask); return ftask;}public Future submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture ftask = newTaskFor(task, result); execute(ftask); return ftask;}public Future submit(Callable task) { if (task == null) throw new NullPointerException(); RunnableFuture ftask = newTaskFor(task); execute(ftask); return ftask;}protected RunnableFuture newTaskFor(Runnable runnable, T value) { return new FutureTask (runnable, value);}public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; // ensure visibility of callable} 首先调用newTaskFor方法构造FutureTask，然后调用execute把任务放进线程池中，返回FutureTask
FutureTask.run()
public void run() { //新建任务，CAS替换runner为当前线程 if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { Callable 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);//处理中断逻辑 }} 运行任务，如果任务状态为NEW状态，则利用CAS修改为当前线程。执行完毕调用set(result)方法设置执行结果。
set(result)源码如下 
cas修改 state状态为 
UNSAFE.putOrderedInt）的方式设置 state状态为 
finishCompletion()唤醒等待线程 
private void finishCompletion() { 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} 回到run方法，如果在 run 期间被中断，此时需要调用handlePossibleCancellationInterrupt处理中断逻辑，确保任何中断（例如cancel(true)）只停留在当前run或runAndReset的任务中
private void handlePossibleCancellationInterrupt(int s) { //在中断者中断线程之前可能会延迟，所以我们只需要让出CPU时间片自旋等待 if (s == INTERRUPTING) while (state == INTERRUPTING) Thread.yield(); // wait out pending interrupt} FutureTask.runAndReset()
runAndReset是 FutureTask另外一个任务执行的方法，它不会返回执行结果，而且在任务执行完之后会重置 stat的状态为 NEW，使任务可以多次执行。 runAndReset的典型应用是在 ScheduledThreadPoolExecutor 中，周期性的执行任务。 FutureTask.get()
get()获取任务执行结果。 如果任务处于未完成的状态（ state <= COMPLETING），就调用 awaitDone等待任务完成。 任务完成后，通过 report获取执行结果或抛出执行期间的异常。 
awaitDone(boolean timed, long nanos)
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);//移除等待WaitNode throw new InterruptedException(); } int s = state; if (s > COMPLETING) { 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) //CAS修改waiter 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用于等待任务完成，或任务因为中断或超时而终止。返回任务的完成状态。
- 如果线程被中断，首先清除中断状态，调用
removeWaiter移除等待节点，然后抛InterruptedException。removeWaiter源码如下：
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)) //cas替换 continue retry; } break; } }} - 如果当前为结束态(
state>COMPLETING),则根据需要置空等待节点的线程，并返回 Future 状态 - 如果当前为正在完成(
COMPLETING)，说明此时 Future 还不能做出超时动作，为任务让出CPU执行时间片 - 如果
state为NEW，先新建一个WaitNode，然后CAS修改当前waiters - 如果等待超时，则调用
removeWaiter移除等待节点，返回任务状态；如果设置了超时时间但是尚未超时，则park阻塞当前线程 - 其他情况直接阻塞当前线程
FutureTask.cancel(boolean mayInterruptIfRunning)
public boolean cancel(boolean mayInterruptIfRunning) { //如果当前Future状态为NEW，根据参数修改Future状态为INTERRUPTING或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 { Thread t = runner; if (t != null) t.interrupt(); } finally { // final state UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); } } } finally { finishCompletion();//移除并唤醒所有等待线程 } return true;} 说明：尝试取消任务。如果任务已经完成或已经被取消，此操作会失败。
如果当前Future状态为NEW，根据参数修改Future状态为INTERRUPTING或CANCELLED。 如果当前状态不为NEW，则根据参数mayInterruptIfRunning决定是否在任务运行中也可以中断。中断操作完成后，调用finishCompletion移除并唤醒所有等待线程。 示例
小结
本章重点：FutureTask 结果返回机制，以及内部运行状态的转变
转载地址：http://yjhkl.baihongyu.com/
你可能感兴趣的文章
Rabbitmq安装
查看>>
2016年3月9日作业
查看>>
tomcat 部署站点时遇到的部分问题以及解决方案
查看>>
excel两个下拉框相互关联
查看>>
HttpURLConnection发送post请求信息
查看>>
mysql出现多线程操作同一个表的情况，应该怎么办？
查看>>
springmvc 将post转换为delete，put
查看>>
第二届清华大学项目管理精英训练营【敏捷个人】分享
查看>>
Centos 安装 Solr
查看>>
Android Toast自己定义Toast例子
查看>>
bash shell实现二进制与十进制数的互转
查看>>
精准测试
查看>>
Oracle11G_逻辑备份
查看>>
Linux正变得无处不在;应用大盘点
查看>>
IEEE期刊
查看>>
pssh使用
查看>>
索引优化系列十四--啥时分区索引性能反而低
查看>>
hive的变量传递设置
查看>>
apache-shiro杂记(三) 用了apache-shiro后，HttpSession.getServletContext() API无法正常工作了...
查看>>
Jquery的$命名冲突
查看>>