CompletableFuture使用详解

Posted 2023-04-10 sermonlizhi

一、简介

1.1 概述

在上一篇文章《CompletionService使用与源码分析》中，已经介绍过了Future的局限性，它没法直接对多个任务进行链式、组合等处理，需要借助并发工具类才能完成，实现逻辑比较复杂。

而CompletableFuture是对Future的扩展和增强。CompletableFuture实现了Future接口，并在此基础上进行了丰富的扩展，完美弥补了Future的局限性，同时CompletableFuture实现了对任务编排的能力。借助这项能力，可以轻松地组织不同任务的运行顺序、规则以及方式。从某种程度上说，这项能力是它的核心能力。而在以往，虽然通过CountDownLatch等工具类也可以实现任务的编排，但需要复杂的逻辑处理，不仅耗费精力且难以维护。

CompletableFuture的继承结构如下：

CompletionStage接口定义了任务编排的方法，执行某一阶段，可以向下执行后续阶段。异步执行的，默认线程池是ForkJoinPool.commonPool()，但为了业务之间互不影响，且便于定位问题，强烈推荐使用自定义线程池。

CompletableFuture中默认线程池如下：

// 根据commonPool的并行度来选择,而并行度的计算是在ForkJoinPool的静态代码段完成的
private static final boolean useCommonPool =
    (ForkJoinPool.getCommonPoolParallelism() > 1);

private static final Executor asyncPool = useCommonPool ?
    ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

ForkJoinPool中初始化commonPool的参数

static 
    // initialize field offsets for CAS etc
    try 
        U = sun.misc.Unsafe.getUnsafe();
        Class<?> k = ForkJoinPool.class;
        CTL = U.objectFieldOffset
            (k.getDeclaredField("ctl"));
        RUNSTATE = U.objectFieldOffset
            (k.getDeclaredField("runState"));
        STEALCOUNTER = U.objectFieldOffset
            (k.getDeclaredField("stealCounter"));
        Class<?> tk = Thread.class;
        ……
     catch (Exception e) 
        throw new Error(e);
    

    commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;
    defaultForkJoinWorkerThreadFactory =
        new DefaultForkJoinWorkerThreadFactory();
    modifyThreadPermission = new RuntimePermission("modifyThread");

    // 调用makeCommonPool方法创建commonPool,其中并行度为逻辑核数-1
    common = java.security.AccessController.doPrivileged
        (new java.security.PrivilegedAction<ForkJoinPool>() 
            public ForkJoinPool run()  return makeCommonPool(); );
    int par = common.config & SMASK; // report 1 even if threads disabled
    commonParallelism = par > 0 ? par : 1;

1.2 功能

1.2.1 常用方法

依赖关系

• thenApply()：把前面任务的执行结果，交给后面的Function
• thenCompose()：用来连接两个有依赖关系的任务，结果由第二个任务返回

and集合关系

• thenCombine()：合并任务，有返回值
• thenAccepetBoth()：两个任务执行完成后，将结果交给thenAccepetBoth处理，无返回值
• runAfterBoth()：两个任务都执行完成后，执行下一步操作(Runnable类型任务)

or聚合关系

• applyToEither()：两个任务哪个执行的快，就使用哪一个结果，有返回值
• acceptEither()：两个任务哪个执行的快，就消费哪一个结果，无返回值
• runAfterEither()：任意一个任务执行完成，进行下一步操作(Runnable类型任务)

并行执行

• allOf()：当所有给定的 CompletableFuture 完成时，返回一个新的 CompletableFuture
• anyOf()：当任何一个给定的CompletablFuture完成时，返回一个新的CompletableFuture

结果处理

• whenComplete：当任务完成时，将使用结果(或 null)和此阶段的异常(或 null如果没有)执行给定操作
• exceptionally：返回一个新的CompletableFuture，当前面的CompletableFuture完成时，它也完成，当它异常完成时，给定函数的异常触发这个CompletableFuture的完成

1.2.2 异步操作

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)

这四个方法的区别：

• runAsync() 以Runnable函数式接口类型为参数，没有返回结果，supplyAsync() 以Supplier函数式接口类型为参数，返回结果类型为U；Supplier接口的 get()是有返回值的(会阻塞)
• 使用没有指定Executor的方法时，内部使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。
• 默认情况下CompletableFuture会使用公共的ForkJoinPool线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置ForkJoinPool线程池的线程数）。如果所有CompletableFuture共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰

异步操作

Runnable runnable = () -> System.out.println("无返回结果异步任务");
CompletableFuture.runAsync(runnable);

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> 
    System.out.println("有返回值的异步任务");
    try 
        Thread.sleep(5000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    return "Hello World";
);
String result = future.get();

获取结果(join&get)

join()和get()方法都是用来获取CompletableFuture异步之后的返回值。join()方法抛出的是uncheck异常（即未经检查的异常),不会强制开发者抛出。get()方法抛出的是经过检查的异常，ExecutionException, InterruptedException 需要用户手动处理（抛出或者 try catch）

结果处理

当CompletableFuture的计算结果完成，或者抛出异常的时候，我们可以执行特定的 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)

• Action的类型是BiConsumer<? super T,? super Throwable>，它可以处理正常的计算结果，或者异常情况。
• 方法不以Async结尾，意味着Action使用相同的线程执行，而Async可能会使用其它的线程去执行(如果使用相同的线程池，也可能会被同一个线程选中执行)。
• 这几个方法都会返回CompletableFuture，当Action执行完毕后它的结果返回原始的CompletableFuture的计算结果或者返回异常

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> 
    try 
        TimeUnit.SECONDS.sleep(1);
     catch (InterruptedException e) 
    
    if (new Random().nextInt(10) % 2 == 0) 
        int i = 12 / 0;
    
    System.out.println("执行结束！");
    return "test";
);
// 任务完成或异常方法完成时执行该方法
// 如果出现了异常,任务结果为null
future.whenComplete(new BiConsumer<String, Throwable>() 
    @Override
    public void accept(String t, Throwable action) 
        System.out.println(t+" 执行完成！");
    
);
// 出现异常时先执行该方法
future.exceptionally(new Function<Throwable, String>() 
    @Override
    public String apply(Throwable t) 
        System.out.println("执行失败：" + t.getMessage());
        return "异常xxxx";
    
);

future.get();

上面的代码当出现异常时，输出结果如下

执行失败：java.lang.ArithmeticException: / by zero
null 执行完成！

二、应用场景

2.1 结果转换

将上一段任务的执行结果作为下一阶段任务的入参参与重新计算，产生新的结果。

thenApply

thenApply接收一个函数作为参数，使用该函数处理上一个CompletableFuture调用的结果，并返回一个具有处理结果的Future对象。

常用使用：

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

具体使用：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 
    int result = 100;
    System.out.println("第一次运算：" + result);
    return result;
).thenApply(number -> 
    int result = number * 3;
    System.out.println("第二次运算：" + result);
    return result;
);

thenCompose

thenCompose的参数为一个返回CompletableFuture实例的函数，该函数的参数是先前计算步骤的结果。

常用方法：

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

具体使用：

CompletableFuture<Integer> future = CompletableFuture
    .supplyAsync(new Supplier<Integer>() 
        @Override
        public Integer get() 
            int number = new Random().nextInt(30);
            System.out.println("第一次运算：" + number);
            return number;
        
    )
    .thenCompose(new Function<Integer, CompletionStage<Integer>>() 
        @Override
        public CompletionStage<Integer> apply(Integer param) 
            return CompletableFuture.supplyAsync(new Supplier<Integer>() 
                @Override
                public Integer get() 
                    int number = param * 2;
                    System.out.println("第二次运算：" + number);
                    return number;
                
            );
        
    );

thenApply 和 thenCompose的区别：

• thenApply转换的是泛型中的类型，返回的是同一个CompletableFuture；
• thenCompose将内部的CompletableFuture调用展开来并使用上一个CompletableFutre调用的结果在下一步的CompletableFuture调用中进行运算，是生成一个新的CompletableFuture。

2.2 结果消费

与结果处理和结果转换系列函数返回一个新的CompletableFuture不同，结果消费系列函数只对结果执行Action，而不返回新的计算值。

根据对结果的处理方式，结果消费函数又可以分为下面三大类：

• thenAccept()：对单个结果进行消费
• thenAcceptBoth()：对两个结果进行消费
• thenRun()：不关心结果，只对结果执行Action

thenAccept

观察该系列函数的参数类型可知，它们是函数式接口Consumer，这个接口只有输入，没有返回值。

常用方法：

public CompletionStage<Void> thenAccept(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);

具体使用：

CompletableFuture<Void> future = CompletableFuture
    .supplyAsync(() -> 
        int number = new Random().nextInt(10);
        System.out.println("第一次运算：" + number);
        return number;
    ).thenAccept(number ->
                  System.out.println("第二次运算：" + number * 5));

thenAcceptBoth

thenAcceptBoth函数的作用是，当两个CompletionStage都正常完成计算的时候，就会执行提供的action消费两个异步的结果。

常用方法：

public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);

具体使用：

CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() 
    @Override
    public Integer get() 
        int number = new Random().nextInt(3) + 1;
        try 
            TimeUnit.SECONDS.sleep(number);
         catch (InterruptedException e) 
            e.printStackTrace();
        
        System.out.println("任务1结果：" + number);
        return number;
    
);

CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() 
    @Override
    public Integer get() 
        int number = new Random().nextInt(3) + 1;
        try 
            TimeUnit.SECONDS.sleep(number);
         catch (InterruptedException e) 
            e.printStackTrace();
        
        System.out.println("任务2结果：" + number);
        return number;
    
);

futrue1.thenAcceptBoth(future2, new BiConsumer<Integer, Integer>() 
    @Override
    public void accept(Integer x, Integer y) 
        System.out.println("最终结果：" + (x + y));
    
);

thenRun

thenRun也是对线程任务结果的一种消费函数，与thenAccept不同的是，thenRun会在上一阶段 CompletableFuture计算完成的时候执行一个Runnable，而Runnable并不使用该CompletableFuture计算的结果。

常用方法：

public CompletionStage<Void> thenRun(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);

具体使用：

CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> 
    int number = new Random().nextInt(10);
    System.out.println("第一阶段：" + number);
    return number;
).thenRun((

}
getNow(null)方法在future完成的情况下会返回结果，就比如上面这个例子，否则返回null (传入的参数)。

2、运行一个简单的异步阶段

这个例子创建一个一个异步执行的阶段：

static void runAsyncExample() {

CompletableFuture cf = CompletableFuture.runAsync(() -> {
    assertTrue(Thread.currentThread().isDaemon());
    randomSleep();
});
assertFalse(cf.isDone());
sleepEnough();
assertTrue(cf.isDone());

}
通过这个例子可以学到两件事情：

CompletableFuture的方法如果以Async结尾，它会异步的执行(没有指定executor的情况下)， 异步执行通过ForkJoinPool实现， 它使用守护线程去执行任务。注意这是CompletableFuture的特性， 其它CompletionStage可以override这个默认的行为。

参考阅读：任务并行执行神器：Fork&Join框架

3、在前一个阶段上应用函数

下面这个例子使用前面 #1 的完成的CompletableFuture， #1返回结果为字符串message,然后应用一个函数把它变成大写字母。

static void thenApplyExample() {

CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> {
    assertFalse(Thread.currentThread().isDaemon());
    return s.toUpperCase();
});
assertEquals("MESSAGE", cf.getNow(null));

}
注意thenApply方法名称代表的行为。

then意味着这个阶段的动作发生当前的阶段正常完成之后。本例中，当前节点完成，返回字符串message。

Apply意味着返回的阶段将会对结果前一阶段的结果应用一个函数。

函数的执行会被阻塞，这意味着getNow()只有打斜操作被完成后才返回。

另外，关注公众号Java技术栈，在后台回复：面试，可以获取我整理的 Java 并发多线程系列面试题和答案，非常齐全。

4、在前一个阶段上异步应用函数

通过调用异步方法(方法后边加Async后缀)，串联起来的CompletableFuture可以异步地执行（使用ForkJoinPool.commonPool()）。

static void thenApplyAsyncExample() {

CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
    assertTrue(Thread.currentThread().isDaemon());
    randomSleep();
    return s.toUpperCase();
});
assertNull(cf.getNow(null));
assertEquals("MESSAGE", cf.join());

}
5、使用定制的Executor在前一个阶段上异步应用函数

异步方法一个非常有用的特性就是能够提供一个Executor来异步地执行CompletableFuture。《线程池全面解析》推荐看下。

这个例子演示了如何使用一个固定大小的线程池来应用大写函数。

static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() {

int count = 1;

@Override
public Thread newThread(Runnable runnable) {
    return new Thread(runnable, "custom-executor-" + count++);
}

});

static void thenApplyAsyncWithExecutorExample() {

CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
    assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));
    assertFalse(Thread.currentThread().isDaemon());
    randomSleep();
    return s.toUpperCase();
}, executor);

assertNull(cf.getNow(null));
assertEquals("MESSAGE", cf.join());

}
6、消费前一阶段的结果

如果下一阶段接收了当前阶段的结果，但是在计算的时候不需要返回值(它的返回类型是void)， 那么它可以不应用一个函数，而是一个消费者， 调用方法也变成了thenAccept:

static void thenAcceptExample() {

StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture("thenAccept message")
        .thenAccept(s -> result.append(s));
assertTrue("Result was empty", result.length() > 0);

}
本例中消费者同步地执行，所以我们不需要在CompletableFuture调用join方法。

7、异步地消费迁移阶段的结果

同样，可以使用thenAcceptAsync方法， 串联的CompletableFuture可以异步地执行。

static void thenAcceptAsyncExample() {

StringBuilder result = new StringBuilder();
CompletableFuture cf = CompletableFuture.completedFuture("thenAcceptAsync message")
        .thenAcceptAsync(s -> result.append(s));
cf.join();
assertTrue("Result was empty", result.length() > 0);

}
8、完成计算异常

现在我们来看一下异步操作如何显式地返回异常，用来指示计算失败。我们简化这个例子，操作处理一个字符串，把它转换成答谢，我们模拟延迟一秒。

我们使用thenApplyAsync(Function, Executor)方法，第一个参数传入大写函数， executor是一个delayed executor,在执行前会延迟一秒。

static void completeExceptionallyExample() {

CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
        CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
CompletableFuture exceptionHandler = cf.handle((s, th) -> { return (th != null) ? "message upon cancel" : ""; });
cf.completeExceptionally(new RuntimeException("completed exceptionally"));

assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());

try {
    cf.join();
    fail("Should have thrown an exception");
} catch(CompletionException ex) { // just for testing
    assertEquals("completed exceptionally", ex.getCause().getMessage());
}

assertEquals("message upon cancel", exceptionHandler.join());

}
让我们看一下细节。

首先我们创建了一个CompletableFuture, 完成后返回一个字符串message,接着我们调用thenApplyAsync方法，它返回一个CompletableFuture。这个方法在第一个函数完成后，异步地应用转大写字母函数。

这个例子还演示了如何通过delayedExecutor(timeout, timeUnit)延迟执行一个异步任务。

我们创建了一个分离的handler阶段：exceptionHandler， 它处理异常异常，在异常情况下返回message upon cancel。

下一步我们显式地用异常完成第二个阶段。在阶段上调用join方法，它会执行大写转换，然后抛出CompletionException（正常的join会等待1秒，然后得到大写的字符串。不过我们的例子还没等它执行就完成了异常）， 然后它触发了handler阶段。

9、取消计算

和完成异常类似，我们可以调用cancel(boolean mayInterruptIfRunning)取消计算。对于CompletableFuture类，布尔参数并没有被使用，这是因为它并没有使用中断去取消操作，相反，cancel等价于completeExceptionally(new CancellationException())。

static void cancelExample() {

CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
        CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
CompletableFuture cf2 = cf.exceptionally(throwable -> "canceled message");
assertTrue("Was not canceled", cf.cancel(true));
assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());
assertEquals("canceled message", cf2.join());

}
10、在两个完成的阶段其中之一上应用函数

下面的例子创建了CompletableFuture, applyToEither处理两个阶段， 在其中之一上应用函数(包保证哪一个被执行)。本例中的两个阶段一个是应用大写转换在原始的字符串上， 另一个阶段是应用小些转换。

static void applyToEitherExample() {

String original = "Message";
CompletableFuture cf1 = CompletableFuture.completedFuture(original)
        .thenApplyAsync(s -> delayedUpperCase(s));
CompletableFuture cf2 = cf1.applyToEither(
        CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
        s -> s + " from applyToEither");
assertTrue(cf2.join().endsWith(" from applyToEither"));

}
11、在两个完成的阶段其中之一上调用消费函数

和前一个例子很类似了，只不过我们调用的是消费者函数 (Function变成Consumer):

static void acceptEitherExample() {

String original = "Message";
StringBuilder result = new StringBuilder();
CompletableFuture cf = CompletableFuture.completedFuture(original)
        .thenApplyAsync(s -> delayedUpperCase(s))
        .acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
                s -> result.append(s).append("acceptEither"));
cf.join();
assertTrue("Result was empty", result.toString().endsWith("acceptEither"));

}
12、在两个阶段都执行完后运行一个 Runnable

这个例子演示了依赖的CompletableFuture如果等待两个阶段完成后执行了一个Runnable。

注意下面所有的阶段都是同步执行的，第一个阶段执行大写转换，第二个阶段执行小写转换。

static void runAfterBothExample() {

String original = "Message";
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth(
        CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
        () -> result.append("done"));
assertTrue("Result was empty", result.length() > 0);

}
13、 使用BiConsumer处理两个阶段的结果

上面的例子还可以通过BiConsumer来实现:

static void thenAcceptBothExample() {

String original = "Message";
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth(
        CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
        (s1, s2) -> result.append(s1 + s2));
assertEquals("MESSAGEmessage", result.toString());

}
14、使用BiFunction处理两个阶段的结果

如果CompletableFuture依赖两个前面阶段的结果， 它复合两个阶段的结果再返回一个结果，我们就可以使用thenCombine()函数。整个流水线是同步的，所以getNow()会得到最终的结果，它把大写和小写字符串连接起来。

static void thenCombineExample() {

String original = "Message";
CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
        .thenCombine(CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)),
                (s1, s2) -> s1 + s2);
assertEquals("MESSAGEmessage", cf.getNow(null));

}
15、异步使用BiFunction处理两个阶段的结果

类似上面的例子，但是有一点不同：依赖的前两个阶段异步地执行，所以thenCombine()也异步地执行，即时它没有Async后缀。

Javadoc中有注释：

Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method

所以我们需要join方法等待结果的完成。

static void thenCombineAsyncExample() {

String original = "Message";
CompletableFuture cf = CompletableFuture.completedFuture(original)
        .thenApplyAsync(s -> delayedUpperCase(s))
        .thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
                (s1, s2) -> s1 + s2);
assertEquals("MESSAGEmessage", cf.join());

}
16、组合 CompletableFuture

我们可以使用thenCompose()完成上面两个例子。这个方法等待第一个阶段的完成(大写转换)， 它的结果传给一个指定的返回CompletableFuture函数，它的结果就是返回的CompletableFuture的结果。

有点拗口，但是我们看例子来理解。函数需要一个大写字符串做参数，然后返回一个CompletableFuture, 这个CompletableFuture会转换字符串变成小写然后连接在大写字符串的后面。

static void thenComposeExample() {

String original = "Message";
CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
        .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))
                .thenApply(s -> upper + s));
assertEquals("MESSAGEmessage", cf.join());

}
17、当几个阶段中的一个完成，创建一个完成的阶段

下面的例子演示了当任意一个CompletableFuture完成后， 创建一个完成的CompletableFuture.

待处理的阶段首先创建， 每个阶段都是转换一个字符串为大写。因为本例中这些阶段都是同步地执行(thenApply), 从anyOf中创建的CompletableFuture会立即完成，这样所有的阶段都已完成，我们使用whenComplete(BiConsumer<? super Object, ? super Throwable> action)处理完成的结果。

static void anyOfExample() {

StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List<CompletableFuture> futures = messages.stream()
        .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))
        .collect(Collectors.toList());
CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((res, th) -> {
    if(th == null) {
        assertTrue(isUpperCase((String) res));
        result.append(res);
    }
});
assertTrue("Result was empty", result.length() > 0);

}
18、当所有的阶段都完成后创建一个阶段

上一个例子是当任意一个阶段完成后接着处理，接下来的两个例子演示当所有的阶段完成后才继续处理, 同步地方式和异步地方式两种。

static void allOfExample() {

StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List<CompletableFuture> futures = messages.stream()
        .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))
        .collect(Collectors.toList());
CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((v, th) -> {
    futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
    result.append("done");
});
assertTrue("Result was empty", result.length() > 0);

}
19、当所有的阶段都完成后异步地创建一个阶段

使用thenApplyAsync()替换那些单个的CompletableFutures的方法，allOf()会在通用池中的线程中异步地执行。所以我们需要调用join方法等待它完成。

static void allOfAsyncExample() {

StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List<CompletableFuture> futures = messages.stream()
        .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> delayedUpperCase(s)))
        .collect(Collectors.toList());
CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
        .whenComplete((v, th) -> {
            futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
            result.append("done");
        });
allOf.join();
assertTrue("Result was empty", result.length() > 0);

}
20、真实的例子

现在你已经了解了CompletionStage 和 CompletableFuture 的一些函数的功能，下面的例子是一个实践场景：

首先异步调用cars方法获得Car的列表，它返回CompletionStage场景。cars消费一个远程的REST API。

然后我们复合一个CompletionStage填写每个汽车的评分，通过rating(manufacturerId)返回一个CompletionStage, 它会异步地获取汽车的评分(可能又是一个REST API调用)

当所有的汽车填好评分后，我们结束这个列表，所以我们调用allOf得到最终的阶段， 它在前面阶段所有阶段完成后才完成。

在最终的阶段调用whenComplete(),我们打印出每个汽车和它的评分。

cars().thenCompose(cars -> {

List<CompletionStage> updatedCars = cars.stream()
        .map(car -> rating(car.manufacturerId).thenApply(r -> {
            car.setRating(r);
            return car;
        })).collect(Collectors.toList());

CompletableFuture done = CompletableFuture
        .allOf(updatedCars.toArray(new CompletableFuture[updatedCars.size()]));
return done.thenApply(v -> updatedCars.stream().map(CompletionStage::toCompletableFuture)
        .map(CompletableFuture::join).collect(Collectors.toList()));

}).whenComplete((cars, th) -> {

if (th == null) {
    cars.forEach(System.out::println);
} else {
    throw new RuntimeException(th);
}

}).toCompletableFuture().join();
因为每个汽车的实例都是独立的，得到每个汽车的评分都可以异步地执行，这会提高系统的性能(延迟)，而且，等待所有的汽车评分被处理使用的是allOf方法，而不是手工的线程等待(Thread#join() 或 a CountDownLatch)。

最后，关注公众号Java技术栈，在后台回复：面试，可以获取我整理的 Java 并发多线程系列面试题和答案，非常齐全。