1.概述
CompletableFuture是jdk1.8引入的实现类。扩展了Future和CompletionStage，是一个可以在任务完成阶段触发一些操作Future。简单的来讲就是可以实现异步回调。
2.为什么引入CompletableFuture
对于jdk1.5的Future，虽然提供了异步处理任务的能力，但是获取结果的方式很不优雅，还是需要通过阻塞（或者轮训）的方式。如何避免阻塞呢？其实就是注册回调。
业界结合观察者模式实现异步回调。也就是当任务执行完成后去通知观察者。比如Netty的ChannelFuture，可以通过注册监听实现异步结果的处理。
Netty的ChannelFuture
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) { checkNotNull(listener, "listener"); synchronized (this) { addListener0(listener); } if (isDone()) { notifyListeners(); } return this; } private boolean setValue0(Object objResult) { if (RESULT_UPDATER.compareAndSet(this, null, objResult) || RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) { if (checkNotifyWaiters()) { notifyListeners(); } return true; } return false; }
通过addListener方法注册监听。如果任务完成，会调用notifyListeners通知。
CompletableFuture通过扩展Future，引入函数式编程，通过回调的方式去处理结果。
3.功能
CompletableFuture的功能主要体现在他的CompletionStage。
可以实现如下等功能
- 转换（thenCompose）
- 组合（thenCombine）
- 消费（thenAccept）
- 运行（thenRun）。
- 带返回的消费（thenApply）
费和运行的区别：
消费使用执行结果。运行则只是运行特定任务。具体其他功能大家可以根据需求自行查看。
CompletableFuture借助CompletionStage的方法可以实现链式调用。并且可以选择同步或者异步两种方式。
这里举个简单的例子来体验一下他的功能。
public static void thenApply() { ExecutorService executorService = Executors.newFixedThreadPool(2); CompletableFuture cf = CompletableFuture.supplyAsync(() -> { try { // Thread.sleep(2000); } catch (Exception e) { e.printStackTrace(); } System.out.println("supplyAsync " + Thread.currentThread().getName()); return "hello"; }, executorService).thenApplyAsync(s -> { System.out.println(s + "world"); return "hhh"; }, executorService); cf.thenRunAsync(() -> { System.out.println("ddddd"); }); cf.thenRun(() -> { System.out.println("ddddsd"); }); cf.thenRun(() -> { System.out.println(Thread.currentThread()); System.out.println("dddaewdd"); }); }
执行结果
supplyAsync pool-1-thread-1 helloworld ddddd ddddsd Thread[main,5,main] dddaewdd
根据结果我们可以看到会有序执行对应任务。
注意：
如果是同步执行cf.thenRun。他的执行线程可能main线程，也可能是执行源任务的线程。如果执行源任务的线程在main调用之前执行完了任务。那么cf.thenRun方法会由main线程调用。
这里说明一下，如果是同一任务的依赖任务有多个：
- 如果这些依赖任务都是同步执行。那么假如这些任务被当前调用线程（main）执行，则是有序执行，假如被执行源任务的线程执行，那么会是倒序执行。因为内部任务数据结构为LIFO。
- 如果这些依赖任务都是异步执行，那么他会通过异步线程池去执行任务。不能保证任务的执行顺序。
上面的结论是通过阅读源代码得到的。下面我们深入源代码。
4.源码追踪
创建CompletableFuture
创建的方法有很多，甚至可以直接new一个。我们来看一下supplyAsync异步创建的方法。
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) { return asyncSupplyStage(screenExecutor(executor), supplier); } static Executor screenExecutor(Executor e) { if (!useCommonPool && e == ForkJoinPool.commonPool()) return asyncPool; if (e == null) throw new NullPointerException(); return e; }
入参Supplier，带返回值的函数。如果是异步方法，并且传递了执行器，那么会使用传入的执行器去执行任务。否则采用公共的ForkJoin并行线程池，如果不支持并行，新建一个线程去执行。
这里我们需要注意ForkJoin是通过守护线程去执行任务的。所以必须有非守护线程的存在才行。
asyncSupplyStage方法
static <U> CompletableFuture<U> asyncSupplyStage(Executor e, Supplier<U> f) { if (f == null) throw new NullPointerException(); CompletableFuture<U> d = new CompletableFuture<U>(); e.execute(new AsyncSupply<U>(d, f)); return d; }
这里会创建一个用于返回的CompletableFuture。
然后构造一个AsyncSupply，并将创建的CompletableFuture作为构造参数传入。
那么，任务的执行完全依赖AsyncSupply。
AsyncSupply#run
public void run() { CompletableFuture<T> d; Supplier<T> f; if ((d = dep) != null && (f = fn) != null) { dep = null; fn = null; if (d.result == null) { try { d.completeValue(f.get()); } catch (Throwable ex) { d.completeThrowable(ex); } } d.postComplete(); } }
- 该方法会调用Supplier的get方法。并将结果设置到CompletableFuture中。我们应该清楚这些操作都是在异步线程中调用的。
d.postComplete方法就是通知任务执行完成。触发后续依赖任务的执行，也就是实现CompletionStage的关键点。
在看postComplete方法之前我们先来看一下创建依赖任务的逻辑。
