把Future扔进CompletableFuture的封装里

关于异步编程，之前一直在项目中使用Guava的ListenableFuture，对于JDK8的CompletableFuture使用较少。

注：这篇文章绝大部分内容都来自colobu.com中的《Java CompletableFuture 详解》的介绍，感谢作者。后续我会统一改成自己的内容。

Future In JDK5

Future是JDK5添加的类，用来描述一个异步计算的结果。可以用isDone方法来检查计算是否完成，或者使用get阻塞住调用线程，直至计算完成返回结果，也可以用cancel方法来停止任务的执行。

public class BasicFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(10);
        Future<Integer> f = es.submit(() -> {
            // 长时间的异步计算
            // ...
            // 然后返回结果
            return 100;
        });
        f.get();
    }
}

Future以及相关使用方法提供了异步执行任务的能力，但对于结果的获取确实不方便，只能通过阻塞或轮询的方式得到任务结果。阻塞的方式与我们理解的异步编程其实是相违背的，而轮询又会耗CPU 资源，而且还不能及时得到计算结果，为什么不能用「观察者设计模式」当计算结果完成即时通知监听者呢？

Future in Netty

很多语言像Node.js，采用回调的方式实现异步编程。Java的一些框架像Netty，自己扩展Java的Future接口，提供了addListener等多个扩展方法：

ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
future.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception{
      if (future.isSuccess()) {
          // SUCCESS
      }
      else {
          // FAILURE
      }
    }
});

CompletableFuture in JDK8

在Java 8中, 新增加了一个包含50个方法左右的类——CompletableFuture，提供了非常强大的Future 的扩展功能，可以帮助我们简化异步编程的复杂性，提供了「函数式编程」的能力，可以通过「回调」的方式处理计算结果，并且提供了「转换」和「组合」CompletableFuture的方法。

CompletionStage

这个是最基础的接口，CompletionFuture是它的一个子接口，来看一下JavaDoc中的介绍：

A stage of a possibly asynchronous computation, that performs an action or computes a value when another 
CompletionStage completes. A stage completes upon termination of its computation, but this may in turn trigger other dependent stages. 

所以每一个执行过程都称为一个stage，它的执行以来前面的stage的完成，并可能影响后面的dependent stages，同时每个stage都可以是异步执行的。

每一个stage的计算都可以以来下面的任何一种方式：Function、Consumer或Runnable，对应的方法依次以apply、accept或run 开头，这也方便我们记忆，主要区别是stage是否需要参数以及是否需要生成新的结果，如：

stage.thenApply(x -> square(x)).thenAccept(x -> System.out.print(x)).thenRun(() -> System.out.println())

主动完成计算

CompletableFuture类实现了CompletionStage和Future接口，所以你还是可以像以前一样通过阻塞或者轮询的方式获得结果，尽管这种方式不推荐使用。

public T get() throws InterruptedException, ExecutionException
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
public T getNow(T valueIfAbsent)
public T join()

getNow有点特殊，如果结果已经计算完则返回结果或者抛出异常，否则返回给定的valueIfAbsent值。

join返回计算的结果或者抛出一个unchecked异常(CompletionException，它包含对应的原始异常)，它和get对抛出的异常的处理有些细微的区别，你可以运行下面的代码进行比较：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    int i = 1/0;
    return 100;
});
//future.join();
future.get();

区别主要在方法声明上，一个是声明了检查异常，一个没有声明会抛出非检查异常。并且抛出的异常的类型也是不同的，一个是ExecutionException，一个是CompletionException，二者均对原始异常进行了包装。

创建对象

CompletableFuture.completedFuture是一个静态辅助方法，用来返回一个已经计算好的CompletableFuture：

public static <U> CompletableFuture<U> completedFuture(U value)

注意，这个返回的结果已经是计算好的，直接获取可以得到结果。

而以下四个静态方法用来执行一段「异步执行」的代码（通过Runnable或者Supplier）来创建CompletableFuture对象：

public static CompletableFuture<Void>   runAsync(Runnable runnable)
public static CompletableFuture<Void>   runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier, Executor executor)

以Async结尾并且没有指定Executor的方法会使用 ForkJoinPool.commonPool() 作为它的线程池执行异步代码。

runAsync方法也好理解，它以Runnable函数式接口类型为参数，所以CompletableFuture的计算结果为空类型Void。

supplyAsync方法以Supplier函数式接口类型为参数，CompletableFuture的计算结果类型为Supplier的返回类型U。

whenComplete

当CompletableFuture的计算正常完成，或者计算时抛出异常时，我们可以执行whenComplete方法来执行特定的Action。主要是下面的方法：

public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)

可以看到Action的类型是 BiConsumer<? super T,? super Throwable> ，即回调该方法时传入的包括返回结果和可能的异常对象。

方法不以Async结尾，意味着Action使用触发当前计算CompletableFuture执行的相同的线程执行。

注意这几个方法都会返回CompletableFuture，当Action执行完毕后它的结果返回原始的CompletableFuture的计算结果或者返回异常，需要注意的是不要在该方法中对返回的结果进行「变更」，虽然该方法并不会影响最终返回的CompletableFuture对象实例，但是可以改变它持有的结果对象的value内容（引用），如果想变更返回的value的内容，如改变类型，可以使用下面介绍的thenApply方法。

这里单独提一下有个exceptionally方法，可以通过chain的方式调用它，但它只有在CompletableFuture计算出现异常时才会被执行：

public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn)

例子：

CompletableFuture<String> test=new CompletableFuture<>();
test.whenComplete((result, ex) -> System.out.println("stage 2: "+result+"\t"+ex))
    .exceptionally(ex -> { System.out.println("stage 3: "+ex); return ""; });
test.completeExceptionally(new IOException()); // 这个方法设置stage状态为完成并设置返回的异常

输出结果：

stage 2: null   java.io.IOException
stage 3: java.util.concurrent.CompletionException: java.io.IOException

因为采用了链式处理，所以先处理whenComplete后处理exceptionally，并且exceptionally的触发是由于显式调用了completeExceptionally 方法让计算抛出异常才得以被触发，但是whenComplete无论什么结果都会被触发。

如果不用链式的方式，可能在并发环境下由于执行的顺序会不一样导致结果不一样，具体可以参考上面的例子链接。

handle

它和whenComplete的区别主要是它可以有返回值并且可以改变返回的数据类型，它的参数是BiFunction而不是BiConsumer：

public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);

所以，它本质上可以对返回值和异常进行处理并返回新结果，相当于增强版本的whenComplete+exceptionally，因为它还可以改变返回数据类型。这个有点像我们下面介绍的thenApply，只不过后者不能处理异常并且在前面的CompletableFuture计算中出现异常时不会触发后面的thenApply，而 handle并不受异常的影响。

thenApply

CompletableFuture可以作为monad(单子)和functor。由于回调风格的实现，我们不必因为等待一个计算完成而阻塞着调用线程，而是告诉CompletableFuture当计算完成的时候请执行某个Function。而且我们还可以将这些操作串联起来，或者将CompletableFuture组合起来。

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

这一组函数的功能是当原来的CompletableFuture计算完后，将结果传递给函数fn，将fn的结果作为新的CompletableFuture计算结果。因此它的功能相当于将CompletableFuture转换成CompletableFuture并不改变future对象本身的引用。

注意Guava的Futures.transform(..)方法也提供了类似的功能，可以选择当前线程执行或使用独立的线程池执行。

使用例子如下：

// 使用静态方法生成一个CompletableFuture对象
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    return 100;
});
// 在future执行完毕后串联两个新的转换操作
CompletableFuture<String> f =  future.thenApplyAsync(i -> i * 10).thenApply(i -> i.toString());
System.out.println(f.get()); // "1000"

需要注意的是，这些转换操作并不是马上执行的，也不会阻塞（如果使用了Async），而是在前一个stage完成后继续执行。

下面来看一个链式调用多个操作的例子：

CompletableFuture<String> futureA = CompletableFuture
                .supplyAsync(() -> "执行结果:" + (100 / 0))
                .thenApply(s -> "apply result:" + s)
                .whenComplete((s, e) -> {
                    if (s != null) {
                        System.out.println(s);//未执行
                    }
                    if (e == null) {
                        System.out.println(s);//未执行
                    } else {
                        System.out.println(e.getMessage());//java.lang.ArithmeticException: / by zero
                    }
                })
                .exceptionally(e -> {
                    System.out.println("ex"+e.getMessage()); //ex:java.lang.ArithmeticException: / by zero
　　　　　　　　　　　  return "futureA result: 100"; }); 
System.out.println(futureA.join());//futureA result: 100

运行可知，因为Supplier计算方法中出现了异常，导致后续的thenApply并不会被触发，而是直接运行到whenComplete ，即该方法不受异常的影响（毕竟它的参数就是支持可以处理异常的BiConsumer），最后因为发生了异常还会执行到exceptionally方法，它可以设置一些默认返回结果等操作。

需要注意的是链式操作的顺序是可以随意改变的，exceptionally方法也可以放到whenComplete方法之前执行，如果前者因为有异常被执行并返回了新的结果，则whenComplete方法将获取不到异常对象，只会得到前面的方法返回的结果，这点一定要注意！

thenAccept

上面介绍的方法是当计算完成的时候，会生成新的计算结果(如thenApply、 handle) ，或者返回同样的计算结果whenComplete，CompletableFuture还提供了一种处理结果的消费方法，只对结果执行Action而不返回新的计算值，因此计算值为Void:

public CompletableFuture<Void>  thenAccept(Consumer<? super T> action)
public CompletableFuture<Void>  thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void>  thenAcceptAsync(Consumer<? super T> action, Executor executor)

例子：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    return 100;
});
CompletableFuture<Void> f =  future.thenAccept(System.out::println);
System.out.println(f.get());

同样需要注意的是，这个方法在前面的CompletableFuture计算出异常时不会被触发，这一点显然不如用handle方便。

thenAcceptBoth以及相关方法提供了类似的功能，「当两个CompletionStage都正常完成计算的时候」，就会执行提供的 BiConsumer<? super T,? super U>类型的action，它用来将两个CompletionStage的结果进行统一处理。

runAfterBoth是当两个CompletionStage都正常完成计算的时候执行一个Runnable，因为run方法没有参数，所以它得不到计算的结果。

public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action, Executor executor)
public     CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,  Runnable action)

例子如下：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    return 100;
});
CompletableFuture<Void> f =  future.thenAcceptBoth(CompletableFuture.completedFuture(10), (x, y) -> System.out.println(x * y));
System.out.println(f.get());  // null

更彻底地，下面一组方法当计算完成的时候会执行一个Runnable，与thenAccept不同，Runnable并不使用CompletableFuture计算的结果。

public CompletableFuture<Void> thenRun(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

因此先前的CompletableFuture计算的结果被忽略了，这个方法返回CompletableFuture类型的对象。

这个方法在我们不关心CompletableFuture的结果的时候用比较方便，如只想打印完成时间。当然也可以使用whenComplete或handle方法，但有点杀鸡用牛刀的感觉。

compose

public <U> CompletableFuture<U> thenCompose(Function<? super T,? extends CompletionStage<U>> fn)
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T,? extends CompletionStage<U>> fn)
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T,? extends CompletionStage<U>> fn, Executor executor)

这一组方法接受一个Function作为参数，这个Function的输入是当前的CompletableFuture的计算结果值，返回结果将是一个新的CompletableFuture，这个新的CompletableFuture会组合原来的CompletableFuture和函数返回的CompletableFuture。因此它的功能类似：

A +–> B +—> C

注意这里的A和B的执行顺序是固定的，A要等待B的执行完毕后才能组合成一个新的C。

例子：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    return 100;
});
CompletableFuture<String> f = future.thenCompose(i -> {
    return CompletableFuture.supplyAsync(() -> {
        return (i * 10) + "";
    });
});
System.out.println(f.get()); //1000

注意thenCompose和thenApply的区别，后者只是将CompletableFuture转换成CompletableFuture，改变的是同一个CompletableFuture中的泛型类型。而thenCompose用来把当前CompletableFuture的结果作为参数连接到另外一个CompletableFuture，并返回一个新的CompletableFuture。

combine

而下面的一组方法thenCombine用来复合另外一个CompletionStage的结果。它的功能类似：

A | +——> B +——> C

两个CompletionStage是并行执行的，它们之间并没有先后依赖顺序，other并不会等待先前的CompletableFuture执行完毕后再执行，二者是独立执行的。

public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

其实从功能上来讲它们的功能更类似thenAcceptBoth，只不过thenAcceptBoth是纯消费，它的函数参数没有返回值，而thenCombine的函数参数fn有返回值。如果连两个任务的返回值也不关心，那么可以直接用runAfterBoth。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    return 100;
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return "abc";
});
CompletableFuture<String> f =  future.thenCombine(future2, (x,y) -> y + "-" + x);
System.out.println(f.get()); //abc-100

注意，默认使用Async底层会用ForkJoinPool.commonPool()来执行，如果用thenCombine默认用当前线程执行，如果加了Async的方法会用ForkJoinPool .commonPool()执行。

Either

thenAcceptBoth和runAfterBoth是当两个CompletableFuture都计算完成，而我们下面要了解的方法是当任意一个CompletableFuture计算完成的时候就会执行。

public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)
public <U> CompletableFuture<U>  applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn)
public <U> CompletableFuture<U>  applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn)
public <U> CompletableFuture<U>  applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor)

acceptEither方法是当任意一个CompletionStage完成的时候，action这个消费者就会被执行。这个方法返回CompletableFuture

applyToEither方法是当任意一个CompletionStage完成的时候，fn会被执行，它的返回值会当作新的CompletableFuture的计算结果。

异常处理

如果在设置 CompletableFuture.complete(value)之前出现了异常，那么 get() 或其他回调函数像 whenComplete() 都会无限期的等待下去。

public static void main(string[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Double> futurePrice = new CompletableFuture<>();
    new Thread(() -> {
        if(true) {
            throw new RuntimeExeption("");
        }
        futurePrice.complete(23.5);
    }).start();
 
    System.out.println(futurePrice.get());
}

但是异常并不会在线程间传播，所以futurePrice.get()会一直等待。解决方法1是调用get(timeout)方法，给定一个超时时间，如果指定时间内还没有获得结果则得到TimeoutException。方法2是要在线程中通过completeExceptionally(ex)来传播异常：

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Double> futurePrice = new CompletableFuture<>();
    new Thread(() -> {
        try {
            if (true) {
                throw new RuntimeException("Something wrong");
            }
            futurePrice.complete(23.5);
        } catch (Exception ex) {
            // 捕获的异常还会由ExecutionException包裹一下
            futurePrice.completeExceptionally(ex); 
        }
    }).start();
 
     System.out.println(futurePrice.get());
}

这时候在futurePrice.get()马上就能收到如下异常，注意get方法是返回ExecutionException异常而非CompletionException的：

java.util.concurrent.ExecutionException: java.lang.Exception: Something wrong

这里有一个例子，可以参考下。

辅助方法

前面我们已经介绍了几个静态方法：completedFuture、runAsync、supplyAsync，下面介绍的这两个方法用来组合多个CompletableFuture。

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

allOf方法是当所有的CompletableFuture都执行完后执行计算。
anyOf方法是当任意一个CompletableFuture执行完后就会执行计算，计算的结果相同。

但是anyOf和applyToEither不同： anyOf接受任意多的CompletableFuture但是applyToEither 只是判断两个CompletableFuture，anyOf返回值的计算结果是参数中其中一个CompletableFuture的计算结果， applyToEither返回值的计算结果却是要经过fn处理的。

References

本文首次发布于 LiuShuo’s Blog, 转载请保留原文链接.

CompletableFuture in Java