）

Posted 2022-02-17 LL.LEBRON

文章目录

• 《Java并发编程的艺术》读后笔记-FutureTask详解（第十章）
• • 1.FutureTask简介
  • 2.FutureTask的使用
  • 3.FutureTask的实现

《Java并发编程的艺术》读后笔记-FutureTask详解（第十章）

1.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.get()方法：

• 当FutureTask处于未启动或已启动状态时，执行FutureTask.get()方法将导致调用线程阻塞
• 当FutureTask处于已完成状态时，执行FutureTask.get()方法将导致调用线程立即返回结果或抛出异常。

对于FutureTask.cancel()方法：

• 当FutureTask处于未启动状态时，执行FutureTask.cancel()方法将导致此任务永远不会被执行
• 当FutureTask处于已启动状态时，执行FutureTask.cancel(true)方法将以中断执行此任务线程的方式来试图停止任务
• 当FutureTask处于已启动状态时，执行FutureTask.cancel(false)方法将不会对正在执行此任务的线程产生影响（让正在执行的任务运行完成）
• 当FutureTask处于已完成状态时，执行FutureTask.cancel(…)方法将返回false。

get方法和cancel方法的执行示意图：

2.FutureTask的使用

可以把FutureTask交给Executor执行；也可以通过ExecutorService.submit(…)方法返回一个FutureTask，然后执行FutureTask.get()方法或FutureTask.cancel(…)方法。除此以外，还可以单独使用FutureTask

当一个线程需要等待另一个线程把某个任务执行完后它才能继续执行，此时可以使用 FutureTask

举个例子：假设有多个线程执行若干任务，每个任务最多只能被执行一次。当多个线程试图同时执行同一个任务时，只允许一个线程执行任务，其他线程需要等待这个任务执行完后才 能继续执行。

代码如下：

/**
 * @author xppll
 * @date 2022/1/18 16:21
 */
public class FutureTaskTest 
    private final ConcurrentMap<Object, Future<String>> taskCache
            = new ConcurrentHashMap<Object, Future<String>>();

    private String executionTask(final String taskName) 
        while (true) 
            Future<String> future = taskCache.get(taskName);
            if (future == null) 
                Callable<String> task = new Callable<String>() 
                    @Override
                    public String call() throws Exception 
                        return taskName;
                    
                ;
                FutureTask<String> futureTask = new FutureTask<String>(task);
                future = taskCache.putIfAbsent(taskName, futureTask);
                if (future == null) 
                    future = futureTask;
                    futureTask.run();
                
                try 
                    return future.get();
                 catch (InterruptedException | ExecutionException e) 
                    taskCache.remove(taskName, future);
                    e.printStackTrace();
                

            
        
    

    public static void main(String[] args) 

    

上述代码的执行示意图如图：

当两个线程试图同时执行同一个任务时，如果Thread 1执行1.3后Thread 2执行2.1，那么接下来Thread 2将在2.2等待，直到Thread 1执行完1.4后Thread 2才能从2.2（FutureTask.get()）返回

3.FutureTask的实现

FutureTask的实现基于AbstractQueuedSynchronizer（AQS）。AQS是一个同步框架，它提供通用机制来原子性管理同步状态、阻塞和唤醒线程，以及维护被阻塞线程的队列。

详细的AQS可以参考：《Java并发编程的艺术》读后笔记-第五章 Java中的锁

每一个基于AQS实现的同步器都会包含两种类型的操作：

1. 至少一个acquire操作。
   • 这个操作阻塞调用线程，除非/直到AQS的状态允许这个线程继续执行。
   • FutureTask的acquire操作为get()/get(long timeout，TimeUnit unit)方法调用。
2. 至少一个release操作。
   • 这个操作改变AQS的状态，改变后的状态可允许一个或多个阻塞线程被解除阻塞。
   • FutureTask的release操作包括run()方法和cancel(…)方法。

基于复合优先于继承的原则，FutureTask声明了一个内部私有的继承于AQS的子类 Sync，对FutureTask所有公有方法的调用都会委托给这个内部子类。

AQS被作为“模板方法模式”的基础类提供给FutureTask的内部子类Sync，这个内部子类只需要实现状态检查和状态更新的方法即可，这些方法将控制FutureTask的获取和释放操作。具体来说，Sync实现了AQS的tryAcquireShared(int)方法和tryReleaseShared(int)方法，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的线程等待队列中等待（见下图的线程A、B、C和D）。当某个线程执行FutureTask.run()方法或FutureTask.cancel(…)方法时，会唤醒线程等待队列的第一个线程（下图所示的线程E唤醒线程A）。

假设开始时FutureTask处于未启动状态或已启动状态，等待队列中已经有3个线程（A、B和C）在等待。此时，线程D执行get()方法将导致线程D也到等待队列中去等待。

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