FutureTask简介

川长思鸟来 2024-03-16 13:01 4阅读 0赞

### FutureTask简介 ###

<table> 
 <thead> 
  <tr> 
   <th align="left">Future接口和实现Future接口的FutureTask类，代表<code>异步计算的结果</code>。</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">FutureTask除了实现<code>Future</code>接口外，还实现了<code>Runnable</code>接口。因此，<strong>FutureTask可以交给Executor执行，也可以由调用线程直接执行（FutureTask.run()）</strong>。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left"><strong>根据FutureTask.run()方法被执行的时机，FutureTask可以处于下面3种状态。</strong></td> 
  </tr> 
  <tr> 
   <td align="left">1）<code>未启动</code>。FutureTask.run()方法还没有被执行之前，FutureTask处于未启动状态。当创建一个FutureTask，且没有执行FutureTask.run()方法之前，这个FutureTask处于未启动状态。</td> 
  </tr> 
  <tr> 
   <td align="left">2）<code>已启动</code>。FutureTask.run()方法被执行的过程中，FutureTask处于已启动状态。</td> 
  </tr> 
  <tr> 
   <td align="left">3）<code>已完成</code>。FutureTask.run()方法执行完后<code>正常结束</code>，或<code>被取消</code>（FutureTask.cancel（…）），或执行FutureTask.run()方法时抛出异常而<code>异常结束</code>，FutureTask处于已完成状态。</td> 
  </tr> 
  <tr> 
   <td align="left">下图是FutureTask的状态迁移的示意图。</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][be4cc2a0eef44bbeb85879894272daa6.png_pic_center]

<table> 
 <thead> 
  <tr> 
   <th align="left">FutureTask的状态迁移示意图解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">当FutureTask处于<code>未启动</code>或已启动状态时，执行FutureTask.get()方法将导致调用线程<code>阻塞</code>；</td> 
  </tr> 
  <tr> 
   <td align="left">当FutureTask处于<code>已完成</code>状态时，执行FutureTask.get()方法将导致调用线程立即<code>返回结果</code>或<code>抛出异常</code>。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">当FutureTask处于<code>未启动</code>状态时，执行FutureTask.cancel()方法将导致此任务<code>永远不会被执行</code>；</td> 
  </tr> 
  <tr> 
   <td align="left">当FutureTask处于<code>已启动</code>状态时，执行FutureTask.cancel（<code>true</code>）方法将以<code>中断执行此任务线程的方式</code>来试图停止任务；</td> 
  </tr> 
  <tr> 
   <td align="left">当FutureTask处于<code>已启动</code>状态时，执行FutureTask.cancel（<code>false</code>）方法将不会对正在执行此任务的线程产生影响（<code>让正在执行的任务运行完成</code>）；</td> 
  </tr> 
  <tr> 
   <td align="left">当FutureTask处于<code>已完成</code>状态时，执行FutureTask.cancel（…）方法将返回<code>false</code>。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">下图是get方法和cancel方法的执行示意图。</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][c93e49a4b5d54e5d9f39f5d341559727.png]

### FutureTask的使用 ###

<table> 
 <thead> 
  <tr> 
   <th align="left"></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">可以把FutureTask交给<code>Executor</code>执行；</td> 
  </tr> 
  <tr> 
   <td align="left">也可以通过ExecutorService.<code>submit</code>（…）方法返回一个FutureTask，然后执行FutureTask.get()方法或FutureTask.cancel（…）方法。</td> 
  </tr> 
  <tr> 
   <td align="left">除此以外，还可以单独使用<code>FutureTask</code>。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left"><strong>当一个线程需要<code>等待</code>另一个线程把某个任务执行完后它才能继续执行，此时可以使用FutureTask。</strong></td> 
  </tr> 
  <tr> 
   <td align="left">假设有多个线程执行若干任务，每个任务最多只能被<code>执行一次</code>。当多个线程试图同时<code>执行同一个</code>任务时，只<code>允许一个</code>线程执行任务，其他线程<code>需要等待</code>这个任务执行完后才能继续执行。下面是对应的示例代码。</td> 
  </tr> 
 </tbody> 
</table>

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);// 1.1,2.1
                if (future == null) {
        
                    Callable<String> task = () -> taskName; // 1.2创建任务
    
                    FutureTask<String> futureTask = new FutureTask<>(task);
                    future = taskCache.putIfAbsent(taskName, futureTask);// 1.3
                    if (future == null) {
        
                        future = futureTask;
                        futureTask.run();// 1.4执行任务
                    }
                }
                try {
        
                    return future.get();// 1.5,2.2线程在此等待任务执行完成
                } catch (CancellationException e) {
        
                    taskCache.remove(taskName, future);
                }
            }
        }
    }

![在这里插入图片描述][098e0273eaf04299b614bafb65634bcd.png_pic_center]

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

### FutureTask的实现 ###

<table> 
 <thead> 
  <tr> 
   <th align="left">FutureTask的实现基于AbstractQueuedSynchronizer（以下简称为AQS）。</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><strong><code>AQS</code> 是一个同步框架，它提供通用机制来原子性管理同步状态、阻塞和唤醒线程，以及维护被阻塞线程的队列。JDK 6中AQS被广泛使用，基于AQS实现的同步器包括：ReentrantLock、Semaphore、ReentrantReadWriteLock、CountDownLatch和FutureTask。</strong></td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">每一个基于AQS实现的同步器都会包含<code>两种类型的操作</code>，如下。</td> 
  </tr> 
  <tr> 
   <td align="left">·至少一个<code>acquire</code>操作。这个操作<code>阻塞调用线程</code>，除非/直到AQS的状态允许这个线程继续执行。FutureTask的acquire操作为<code>get()</code>/<code>get（long timeout，TimeUnit unit）</code>方法调用。</td> 
  </tr> 
  <tr> 
   <td align="left">·至少一个<code>release</code>操作。这个操作<code>改变AQS的状态</code>，改变后的状态可允许一个或多个阻塞线程被解除阻塞。FutureTask的<code>release</code>操作包括<code>run()</code>方法和<code>cancel（…）</code>方法。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">基于“<code>复合优先于继承</code>”的原则，FutureTask声明了一个内部私有的继承于AQS的子类<code>Sync</code>，对FutureTask所有公有方法的调用都会<code>委托</code>给这个内部子类。</td> 
  </tr> 
  <tr> 
   <td align="left">AQS被作为“<code>模板方法模式</code>”的基础类提供给FutureTask的内部子类Sync，这个内部子类只需要实现<code>状态检查</code>和<code>状态更新</code>的方法即可，这些方法将控制FutureTask的获取和释放操作。具体来说，Sync实现了AQS的<code>tryAcquireShared（int）</code>方法和<code>tryReleaseShared（int）</code>方法，Sync通过这两个方法来检查和更新同步状态。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">FutureTask的设计示意图如下图所示</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][164ce0ff7a8d45d48253fc8663f58914.png_pic_center]

<table> 
 <thead> 
  <tr> 
   <th align="left"></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><strong><code>FutureTask.get()</code></strong> 方法会调用AQS.acquireSharedInterruptibly（int arg）方法，这个方法的执行过程如下。</td> 
  </tr> 
  <tr> 
   <td align="left">1）调用AQS.acquireSharedInterruptibly（int arg）方法，这个方法首先会<code>回调</code>在子类Sync中实现的tryAcquireShared()方法来判断acquire操作是否可以成功。acquire操作可以成功的条件为：state为执行完成状态<code>RAN</code>或已取消状态<code>CANCELLED</code>，且runner不为<code>null</code>。</td> 
  </tr> 
  <tr> 
   <td align="left">2）如果成功则get()方法<code>立即返回</code>。如果失败则到<code>线程等待队列</code>中去等待其他线程执行release操作。</td> 
  </tr> 
  <tr> 
   <td align="left">3）当其他线程执行<code>release</code>操作（比如<code>FutureTask.run()</code>或<code>FutureTask.cancel（…）</code>）唤醒当前线程后，当前线程<code>再次执行</code>tryAcquireShared()将返回正值1，当前线程将离开线程等待队列并<code>唤醒</code>它的后继线程（这里会产生级联唤醒的效果，后面会介绍）。</td> 
  </tr> 
  <tr> 
   <td align="left">4）最后返回计算的结果或抛出异常。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left"><strong><code>FutureTask.run()</code></strong> 的执行过程如下。</td> 
  </tr> 
  <tr> 
   <td align="left">1）执行在构造函数中指定的任务（Callable.call()）。</td> 
  </tr> 
  <tr> 
   <td align="left">2）以原子方式来<code>更新同步状态</code>（调用AQS.compareAndSetState（int expect，int update），设置state为执行完成状态<code>RAN</code>）。如果这个原子操作成功，就设置代表计算结果的变量result的值为Callable.call()的返回值，然后调用AQS.releaseShared（int arg）。</td> 
  </tr> 
  <tr> 
   <td align="left">3）AQS.releaseShared（int arg）首先会<code>回调</code>在子类Sync中实现的tryReleaseShared（arg）来执行release操作（设置运行任务的线程runner为null，然会返回true）；AQS.releaseShared（int arg），然后<code>唤醒</code>线程等待队列中的第一个线程。</td> 
  </tr> 
  <tr> 
   <td align="left">4）调用FutureTask.done()。</td> 
  </tr> 
  <tr> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">当执行FutureTask.get()方法时，如果FutureTask不是处于执行完成状态<code>RAN</code>或已取消状态<code>CANCELLED</code>，当前执行线程将到AQS的线程等待队列中等待（见下图的线程A、B、C和D）。当某个线程执行FutureTask.run()方法或FutureTask.cancel（…）方法时，会唤醒线程等待队列的第一个线程（见下图所示的线程E唤醒线程A）。</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][7318f70113e64068966190cabebd09e6.png_pic_center]

<table> 
 <thead> 
  <tr> 
   <th align="left">针对上图FutureTask的级联唤醒示意图</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">假设开始时FutureTask处于<code>未启动</code>状态或<code>已启动</code>状态，等待队列中已经有3个线程（A、B和C）在等待。此时，线程D执行get()方法将导致线程D也到等待队列中去等待。</td> 
  </tr> 
  <tr> 
   <td align="left">当线程E执行run()方法时，会唤醒队列中的<code>第一个</code>线程A。线程A被唤醒后，首先把自己从队列中<code>删除</code>，然后<code>唤醒</code>它的后继线程B，最后线程A从get()方法返回。线程B、C和D重复A线程的处理流程。最终，在队列中等待的所有线程都被级联唤醒并从get()方法返回。</td> 
  </tr> 
 </tbody> 
</table>

\-----------------------------------------------------------------------------读书笔记摘自 书名：Java并发编程的艺术 作者：方腾飞；魏鹏；程晓明

[be4cc2a0eef44bbeb85879894272daa6.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/72220072e72a420d8bbc652712ca5f74.png
[c93e49a4b5d54e5d9f39f5d341559727.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/2d7aeaa6799141de8a2a9be2e080f942.png
[098e0273eaf04299b614bafb65634bcd.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/041f46c4cf08473ba9455eaf9b3a235c.png
[164ce0ff7a8d45d48253fc8663f58914.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/018bda214c494e488e37a37b65d9c84d.png
[7318f70113e64068966190cabebd09e6.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/6a6cac07f7d040328bc4e8a259a5cf64.png