FutureTask简介
Future接口和实现Future接口的FutureTask类,代表异步计算的结果
。 FutureTask除了实现Future
接口外,还实现了Runnable
接口。因此,FutureTask可以交给Executor执行,也可以由调用线程直接执行(FutureTask.run()) 。 根据FutureTask.run()方法被执行的时机,FutureTask可以处于下面3种状态。 1)未启动
。FutureTask.run()方法还没有被执行之前,FutureTask处于未启动状态。当创建一个FutureTask,且没有执行FutureTask.run()方法之前,这个FutureTask处于未启动状态。 2)已启动
。FutureTask.run()方法被执行的过程中,FutureTask处于已启动状态。 3)已完成
。FutureTask.run()方法执行完后正常结束
,或被取消
(FutureTask.cancel(…)),或执行FutureTask.run()方法时抛出异常而异常结束
,FutureTask处于已完成状态。 下图是FutureTask的状态迁移的示意图。
FutureTask的状态迁移示意图解释 当FutureTask处于未启动
或已启动状态时,执行FutureTask.get()方法将导致调用线程阻塞
; 当FutureTask处于已完成
状态时,执行FutureTask.get()方法将导致调用线程立即返回结果
或抛出异常
。 当FutureTask处于未启动
状态时,执行FutureTask.cancel()方法将导致此任务永远不会被执行
; 当FutureTask处于已启动
状态时,执行FutureTask.cancel(true
)方法将以中断执行此任务线程的方式
来试图停止任务; 当FutureTask处于已启动
状态时,执行FutureTask.cancel(false
)方法将不会对正在执行此任务的线程产生影响(让正在执行的任务运行完成
); 当FutureTask处于已完成
状态时,执行FutureTask.cancel(…)方法将返回false
。 下图是get方法和cancel方法的执行示意图。
FutureTask的使用
可以把FutureTask交给Executor
执行; 也可以通过ExecutorService.submit
(…)方法返回一个FutureTask,然后执行FutureTask.get()方法或FutureTask.cancel(…)方法。 除此以外,还可以单独使用FutureTask
。 当一个线程需要等待
另一个线程把某个任务执行完后它才能继续执行,此时可以使用FutureTask。 假设有多个线程执行若干任务,每个任务最多只能被执行一次
。当多个线程试图同时执行同一个
任务时,只允许一个
线程执行任务,其他线程需要等待
这个任务执行完后才能继续执行。下面是对应的示例代码。
public class FutureTaskTest {
private final ConcurrentMap < Object , Future < String > > taskCache = new ConcurrentHashMap < > ( ) ;
private String executionTask ( final String taskName) throws ExecutionException , InterruptedException {
while ( true ) {
Future < String > future = taskCache. get ( taskName) ;
if ( future == null ) {
Callable < String > task = ( ) - > taskName;
FutureTask < String > futureTask = new FutureTask < > ( task) ;
future = taskCache. putIfAbsent ( taskName, futureTask) ;
if ( future == null ) {
future = futureTask;
futureTask. run ( ) ;
}
}
try {
return future. get ( ) ;
} catch ( CancellationException e) {
taskCache. remove ( taskName, future) ;
}
}
}
}
当两个线程试图同时执行同一个任务时,如果Thread 1执行1.3后Thread 2执行2.1,那么接下来Thread 2将在2.2等待,直到Thread 1执行完1.4后Thread 2才能从2.2(FutureTask.get())返回。
FutureTask的实现
FutureTask的实现基于AbstractQueuedSynchronizer(以下简称为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所有公有方法的调用都会委托
给这个内部子类。 AQS被作为“模板方法模式
”的基础类提供给FutureTask的内部子类Sync,这个内部子类只需要实现状态检查
和状态更新
的方法即可,这些方法将控制FutureTask的获取和释放操作。具体来说,Sync实现了AQS的tryAcquireShared(int)
方法和tryReleaseShared(int)
方法,Sync通过这两个方法来检查和更新同步状态。 FutureTask的设计示意图如下图所示
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的线程等待队列中等待(见下图的线程A、B、C和D)。当某个线程执行FutureTask.run()方法或FutureTask.cancel(…)方法时,会唤醒线程等待队列的第一个线程(见下图所示的线程E唤醒线程A)。
针对上图FutureTask的级联唤醒示意图 假设开始时FutureTask处于未启动
状态或已启动
状态,等待队列中已经有3个线程(A、B和C)在等待。此时,线程D执行get()方法将导致线程D也到等待队列中去等待。 当线程E执行run()方法时,会唤醒队列中的第一个
线程A。线程A被唤醒后,首先把自己从队列中删除
,然后唤醒
它的后继线程B,最后线程A从get()方法返回。线程B、C和D重复A线程的处理流程。最终,在队列中等待的所有线程都被级联唤醒并从get()方法返回。
-----------------------------------------------------------------------------读书笔记摘自 书名:Java并发编程的艺术 作者:方腾飞;魏鹏;程晓明