JDK源码分析之concurrent包（三） -- Future方式的实现

　　上一篇我们基于JDK的源码对线程池ThreadPoolExecutor的实现做了分析，本篇来对Executor框架中另一种典型用法Future方式做源码解读。我们知道Future方式实现了带有返回值的程序的异步调用，关于异步调用的场景大家可以自行脑补Ajax的应用（获取返回结果的方式不同，Future是主动询问获取，Ajax是回调函数），这里不做过多说明。

在进入源码前，首先来看下Future方式相关的API：

接口Callable：有返回结果并且可能抛出异常的任务；
接口Future：表示异步执行的结果；
类FutureTask：实现Future、Runnable等接口，是一个异步执行的任务。可以直接执行，或包装成Callable执行；
接口CompletionService：将生产新的异步任务与使用已完成任务的结果分离开来的服务，用来执行Callable或Runnable，并异步获取执行结果；
类ExecutorCompletionService：实现CompletionService接口，使用构造时传入的Executor来执行Callable或Runnable，

接下来通过一个常规的使用实例来展示这些API之间的关系：

 ExecutorService executor = Executors.newFixedThreadPool(3);

 CompletionService<String> completionService = new ExecutorCompletionService<String>(executor);

 Future<String> future = completionService.submit(new Callable<String>() {

     public String call() throws Exception {

         // do something...

         return "success";

     }

 });

 // 其它的程序逻辑。。。

 // 异步的获取执行结果

 String result = future.get();

常规调用主要是通过CompletionService.submit()方法，那我们就从这个方法开始进入JDK源码，以下是ExecutorCompletionService类的源码：

 public Future<V> submit(Callable<V> task) {

     if (task == null) throw new NullPointerException();

     RunnableFuture<V> f = newTaskFor(task);

     executor.execute(new QueueingFuture(f));

     return f;

 }

 public Future<V> submit(Runnable task, V result) {

     if (task == null) throw new NullPointerException();

     RunnableFuture<V> f = newTaskFor(task, result);

     executor.execute(new QueueingFuture(f));

     return f;

 }

sunmit()方法有两个重载，这里我们只对参数为Callable的方法解读，因为另一个也是间接的封装成了Callable最后调用的。

上述代码的第3行可以看到，Callable转换成了RunnableFuture来交给executor执行，下面来看newTaskFor(task)方法的源码：

 private RunnableFuture<V> newTaskFor(Callable<V> task) {

     if (aes == null)

         return new FutureTask<V>(task);

     else

         return aes.newTaskFor(task);

 }

代码第3行和第5行的结果是一样的，继续来看这个FutureTask如何构造的：

 public FutureTask(Callable<V> callable) {

     if (callable == null)

         throw new NullPointerException();

     sync = new Sync(callable);

 }

还是没什么用，然后我们详细的来看下Sync的源码：

 /**

  * Synchronization control for FutureTask. Note that this must be

  * a non-static inner class in order to invoke the protected

  * <tt>done</tt> method. For clarity, all inner class support

  * methods are same as outer, prefixed with "inner".

  *

  * Uses AQS sync state to represent run status

  */

 private final class Sync extends AbstractQueuedSynchronizer {

     /** The underlying callable */

     private final Callable<V> callable;

     /** The result to return from get() */

     private V result;

     /** The exception to throw from get() */

     private Throwable exception;

     Sync(Callable<V> callable) {

         this.callable = callable;

     }

     V innerGet() throws InterruptedException, ExecutionException {

         acquireSharedInterruptibly(0);

         if (getState() == CANCELLED)

             throw new CancellationException();

         if (exception != null)

             throw new ExecutionException(exception);

         return result;

     }

     void innerSet(V v) {

       for (;;) {

         int s = getState();

         if (s == RAN)

           return;

         if (s == CANCELLED) {

           // aggressively release to set runner to null,

           // in case we are racing with a cancel request

           // that will try to interrupt runner

           releaseShared(0);

           return;

         }

         if (compareAndSetState(s, RAN)) {

           result = v;

           releaseShared(0);

           done();

           return;

         }

       }

     }

     void innerSetException(Throwable t) {

       for (;;) {

         int s = getState();

         if (s == RAN)

           return;

         if (s == CANCELLED) {

             // aggressively release to set runner to null,

             // in case we are racing with a cancel request

             // that will try to interrupt runner

             releaseShared(0);

             return;

         }

         if (compareAndSetState(s, RAN)) {

           exception = t;

           result = null;

           releaseShared(0);

           done();

           return;

         }

       }

     }

     void innerRun() {

         if (!compareAndSetState(0, RUNNING))

             return;

         try {

             runner = Thread.currentThread();

             if (getState() == RUNNING) // recheck after setting thread

                 innerSet(callable.call());

             else

                 releaseShared(0); // cancel

         } catch (Throwable ex) {

             innerSetException(ex);

         }

     }

     // others codes

 }

上述代码列出了几个重要的方法，可以大概的看出，Future方式的玄机，基本都在这个内部类里了。下面就对这个内部类中的几个方法少做解释：

首先，类的注释主要说明了：内部类必须是非静态的，是为了调用外部类的done()方法（这个我们以后再说）。还有内部类的方法都是以“前缀inner+外部类方法名”来命名的。

其次，通过查看外部类的源码可得知：外部类的所有方法都是通过内部类中同名的inner方法来调用的（源码很简单这里没有列出）。

然后，我们来看这个类中的其中3个成员变量及其注释，就可以大概猜到：callable是传入的执行过程，result用来存储callable的返回值，exception存储callable抛出的异常（如果有）。

最后，我们来分别看这个类中的几个关键方法：

上述代码第74行的innerRun()方法：注意外部类是FutureTask，实现了Runnable接口，事实上就是最开始所说的submit()方法中，最终要执行的Runnable任务，此时执行的其实是内部类的innerRun()，通过代码的第80行可以看出，是调用了callable.call()并把返回值通过innerSet(V v)赋值给了成员变量result。如果callable.call()有异常，则通过innerSetException(Throwable t)赋值给成员变量exception。

通过innserGet()方法差不多就知道了异步获取执行结果的原理了，第23行的acquireSharedInterruptibly(0)方法的意义在于：要等Runnable任务执行完或被中断才能执行后面的代码。（注：这块实现虽然被一笔带过，但其实逻辑还是有点复杂，其实现主要是用死循环检查执行Callable线程的状态，类似自旋锁的概念）

最后，在回过头来看CompletionService.submit()方法：

 public Future<V> submit(Callable<V> task) {

     if (task == null) throw new NullPointerException();

     RunnableFuture<V> f = newTaskFor(task);

     executor.execute(new QueueingFuture(f));

     return f;

 }

代码的第4行并非执行的是我们上面说的FutureTask，而是将这个FutureTask由封装成了QueueingFuture才交给executor执行，当我们看了QueueingFuture的源码就会了解到

 /**

  * FutureTask extension to enqueue upon completion

  */

 private class QueueingFuture extends FutureTask<Void> {

     QueueingFuture(RunnableFuture<V> task) {

         super(task, null);

         this.task = task;

     }

     protected void done() { completionQueue.add(task); }

     private final Future<V> task;

 }

这个封装的意义无非是想在callable.call()执行完后调用第9行的completionQueue.add(task)，done()方法是不是很眼熟？

注释其实已经说明了：是为了让完成的任务入列到completionQueue中，以实现本文最开始罗列出来的，CompletionService接口的“将新的异步任务与完成的任务分离开来”的特性。

总结

本文通过ExecutorCompletionService类与FutureTask类及其内部类中部分关键处源码的解读，介绍了Java5中Future方式的原理。其实可以概括为一句话：

在某线程中执行Callable时，将执行结果或抛出的异常存放在临时变量中，其它线程在Callable执行完或中断前，阻塞的获取执行结果。

关于Executor框架的源码就解读到这，下篇文章开始一些工具类的源码解析。

巴特西

JDK源码分析之concurrent包（三） -- Future方式的实现

总结

最新文章

热门文章