Java8 增强的Future:CompletableFuture

Posted 2020-09-05 Darcy_wang

     CompletableFuture是Java8新增的一个超大型工具类,为什么说她大呢?因为一方面它实现了Future接口,更重要的是,它实现了CompletionStage接口.这个接口也是Java8新增加的,而CompletionStage拥有多达约40种方法,

	* 完成了通知我
	* 异步执行任务

通过CompletableFuture提供进一步封装,我们很容易实现Future模式那样的异步调用,例如:
public static Integer cale(Integer para) {
    try {
        Thread.sleep(1000);

    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return para * para;
}
public static void main(String[] args) throws ExecutionException, InterruptedException {

    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> cale(50));
    System.out.println(future.get());
}

上述代码中CompletableFuture.supplyAsync()方法构造了一个CompletableFuture实例,在supplyAsync()函数中,他会在一个新的线程中,执行传入的参数.在这里,,他会执行calc()方法,而calc()方法的执行可能是比较慢的,但是不影响CompletableFuture实例的构造速度,因此supplyAsync()会立即返回,他返回的CompletableFuture对象实例就可以作为这次调用的契约,在将来任何场合,用于获得最终的计算结果.
在CompletableFuture中,类似的工厂方法有以下几个:
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) 
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
                                                   Executor executor)
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable,
                                               Executor executor)

     其中supplyAsync()方法用于那些需要返回值的场景,比如计算某个数据,而runAsync()方法用于没有返回值的场景,比如,仅仅是简单地执行某一个异步动作.
     在这两对方法中,都有一个方法可以接手一个Executor参数,这使我们可以让Supplier<U>或者Runnable在指定的线程池工作,如果不指定,则在默认的系统公共的ForkJoinPool.common线程池中执行.

	* 流式调用

在前文中我已经简单的提到,CompletionStage的约40个接口为函数式编程做准备的,在这里,就让我们看一下,如果使用这些接口进行函数式的流式API调用:
CompletableFuture<Void> future = CompletableFuture
        .supplyAsync(() -> cale(50))
        .thenApply(i -> Integer.toString(i))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
future.get();

     上述代码中,使用supplyAsync()函数执行了一个异步任务,接着连续使用流式调用对任务处理结果进行在加工,直到最后的结果输出:

	* CompletableFuture中的异常处理

CompletableFuture<Void> future = CompletableFuture
        .supplyAsync(() -> cale(50))
        .exceptionally(ex -> {
            System.out.println("ex.toString() = " + ex.toString());
            return 0;
        })
        .thenApply(i -> Integer.toString(i))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
future.get();


	* 组合多个CompletableFuture

CompletableFuture还允许你将多个CompletableFuture进行组合,一种方法是使用thenCompose(),它的签名如下:
public <U> CompletableFuture<U> thenCompose(
    Function<? super T, ? extends CompletionStage<U>> fn)

一个CompletableFuture可以执行完成后,将执行结果通过Function传递给下一个CompletionStage进行处理:

CompletableFuture<Void> future = CompletableFuture
        .supplyAsync(() -> cale(50))
        .thenCompose(i -> CompletableFuture
                .supplyAsync(() -> cale(i)))
        .thenApply(i -> Integer.toString(i))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
future.get();


另外一种组和多个CompletableFuture的方法是thenCombine(),它的签名如下:
public <U,V> CompletableFuture<V> thenCombine(
    CompletionStage<? extends U> other,
    BiFunction<? super T,? super U,? extends V> fn)

     方法thenCombine()首先完成当前CompletableFuture和other的执行,接着,将这两者的执行结果传递给BiFunction(该接口接受两个参数,并有一个返回值),并返回代表BiFuntion实例的CompletableFuture对象:
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> cale(50));
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> cale(25));

CompletableFuture<Void> fu = future1.thenCombine(future2, (i, j) -> (i + j))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
fu.get();