FutureTask

FutureTask

Future 接口和实现 Future 接口的 FutureTask 类，代表异步计算的结果。FutureTask 的实现基于 AbstractQueuedSynchronizer（以下简称为 AQS）。java.util.concurrent 中的很多可阻塞类（比如 ReentrantLock）都是基于 AQS 来实现的。AQS 是一个同步框架，它提供通用机制来原子性管理同步状态、阻塞和唤醒线程，以及维护被阻塞线程的队列。JDK 6 中 AQS 被广泛使用，基于 AQS 实现的同步器包括：ReentrantLock、Semaphore、ReentrantReadWriteLock、CountDownLatch 和 FutureTask。每一个基于 AQS 实现的同步器都会包含两种类型的操作，如下。

• 至少一个 acquire 操作。这个操作阻塞调用线程，除非/直到 AQS 的状态允许这个线程继续执行。FutureTask 的 acquire 操作为 get()/get（long timeout，TimeUnit unit）方法调用。

• 至少一个 release 操作。这个操作改变 AQS 的状态，改变后的状态可允许一个或多个阻塞线程被解除阻塞。FutureTask 的 release 操作包括 run()方法和 cancel（…）方法。基于“复合优先于继承”的原则，FutureTask 声明了一个内部私有的继承于 AQS 的子类 Sync，对 FutureTask 所有公有方法的调用都会委托给这个内部子类。Sync 是 FutureTask 的内部私有类，它继承自 AQS。创建 FutureTask 时会创建内部私有的成员对象 Sync，FutureTask 所有的的公有方法都直接委托给了内部私有的 Sync。FutureTask.get()方法会调用 AQS.acquireSharedInterruptibly（int arg）方法，这个方法的执行过程如下。

1）调用 AQS.acquireSharedInterruptibly（int arg）方法，这个方法首先会回调在子类 Sync 中实现的 tryAcquireShared()方法来判断 acquire 操作是否可以成功。acquire 操作可以成功的条件为：state 为执行完成状态 RAN 或已取消状态 CANCELLED，且 runner 不为 null。

2）如果成功则 get()方法立即返回。如果失败则到线程等待队列中去等待其他线程执行 release 操作。

3）当其他线程执行 release 操作（比如 FutureTask.run()或 FutureTask.cancel（…））唤醒当前线程后，当前线程再次执行 tryAcquireShared()将返回正值 1，当前线程将离开线程等待队列并唤醒它的后继线程（这里会产生级联唤醒的效果，后面会介绍）。

4）最后返回计算的结果或抛出异常。

FutureTask.run()的执行过程如下。

1）执行在构造函数中指定的任务（Callable.call()）。

2）以原子方式来更新同步状态（调用 AQS.compareAndSetState（int expect，int update），设置 state 为执行完成状态 RAN）。如果这个原子操作成功，就设置代表计算结果的变量 result 的值为 Callable.call()的返回值，然后调用 AQS.releaseShared（int arg）。

3）AQS.releaseShared（int arg）首先会回调在子类 Sync 中实现的 tryReleaseShared（arg）来执行 release 操作（设置运行任务的线程 runner 为 null，然会返回 true）；AQS.releaseShared（int arg），然后唤醒线程等待队列中的第一个线程。

4）调用 FutureTask.done()。当执行 FutureTask.get()方法时，如果 FutureTask 不是处于执行完成状态 RAN 或已取消状态 CANCELLED，当前执行线程将到 AQS 的线程等待队列中等待。当某个线程执行 FutureTask.run()方法或 FutureTask.cancel（...）方法时，会唤醒线程等待队列的第一个线程

当线程 E 执行 run()方法时，会唤醒队列中的第一个线程 A。线程 A 被唤醒后，首先把自己从队列中删除，然后唤醒它的后继线程 B，最后线程 A 从 get()方法返回。线程 B、C 和 D 重复 A 线程的处理流程。最终，在队列中等待的所有线程都被级联唤醒并从 get()方法返回。