王者并发课-钻石3：琳琅满目-细数CompletableFuture的那些花式玩法

欢迎来到《王者并发课》，本文是该系列文章中的第26篇，砖石中的第3篇。

从Java8开始，JDK引入了很多新的特性，包括lambda表达式、流式计算等，以及本文所要详述的CompletableFuture. 关于CompletableFuture，你可能首先会联想到Future接口，对于它我们并不陌生，在ThreadPoolExecutor和ForkJoinPool中都见过它的身影。如果你对此感到困惑的话，不妨先阅读我们的前两篇文章。

Future的接口定义本身并不复杂，使用起来也较为简单，它的核心是

get()

isDone()

在这篇文章中，我们将结合Future和线程池，探讨CompletableFuture与Future的不同之处，以及它的核心能力和最佳实践。

一、理解CompletableFuture

1. Future的局限性

从本质上说，Future表示一个异步计算的结果。它提供了

isDone()

get()

• 并发执行多任务：Future只提供了get()方法来获取结果，并且是阻塞的。所以，除了等待你别无他法；
• 无法对多个任务进行链式调用：如果你希望在计算任务完成后执行特定动作，比如发邮件，但Future却没有提供这样的能力；
• 无法组合多个任务：如果你运行了10个任务，并期望在它们全部执行结束后执行特定动作，那么在Future中这是无能为力的；
• 没有异常处理：Future接口中没有关于异常处理的方法；

2. CompletableFuture与Future的不同

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

3. CompletableFuture初体验

当然，百闻不如一见，既然CompletableFuture如此神乎其神，我们不妨通过一个特定的场景来体验CompletableFuture的用法。

众所周知，王者中有注明的“草丛三杰（B）”，妲己就是其中之一，她蹲草丛的本领可谓一绝。话说这天，妲己远远看见地方小鲁班蹦蹦跳跳地走来，对付这样的脆皮最适合在草丛中来一波操作。于是，妲己侧身躲进了草丛，在小鲁班欢快地路过时，妲己一套熟练的231连招秒杀了小鲁班。小鲁班死不瞑目，因为他甚至还没看清对手的模样，很快啊！

在这个过程中，包含几组动作：捉拿鲁班、打出技能2、打出技能3以及打出技能1. 我们可以通过CompletableFuture的链式调用来表达这些动作：

CompletableFuture.supplyAsync(CompletableFutureDemo::活捉鲁班)
.thenAccept(player -> note(player.getName())) // 接收supplyAsync的结果，获得对方名字
.thenRun(() -> attack("2技能-偶像魅力：鲁班受到妲己285点法术伤害，并眩晕1.5秒..."))
.thenRun(() -> attack("3技能-女王崇拜：妲己放出5团狐火，鲁班受到325++点法术伤害..."))
.thenRun(() -> attack("1技能-灵魂冲击：鲁班受到妲己520点点法术伤害..."))
.thenRunAsync(() -> note("鲁班，卒...")); // 使用线程池的其他线程

你看，使用CompletableFuture编排动作是不是很容易？在这段只有6行的代码中，我们已经使用了supplyAsync()和thenAccept()等4中不同的方法，并且同时使用了同步和异步。在以往，如果手工实现的话，至少需要洋洋洒洒几十行代码。那CompletableFuture是如何做到如此神功的呢？接着往下看。

二、CompletableFuture的核心设计

总体而言，CompletableFuture实现了Future和CompletionStage两个接口，并且只有少量的属性。但是，它有近2400余行的代码，并且关系复杂。所以，在核心设计方面，我们不会展开讨论。

现在，你已经知道，Future接口仅提供了

get()

isDone

顾名思义，根据CompletionStage名字中的“Stage”，你可以把它理解为任务编排中的步骤。所谓步骤，即任务编排的基本单元，它可以是一次纯粹的计算或者是一个特定的动作。在一次编排中，会包含多个步骤，这些步骤之间会存在依赖、链式和组合等不同的关系，也存在并行和串行的关系。这种关系，类似于Pipeline或者流式计算。

既然是编排，就需要维护任务的创建、建立计算关系。为此，CompletableFuture提供了多达50多个方法，在数量上确实庞大且令人瞠目结舌，想要全部理解显然不太可能，当然也没有必要。虽然CompletableFuture的方法数量众多，但是在理解时仍有规律可循，我们可以通过分类的方式简化对方法的理解，理解了类型和变种，基本上我们也就掌握了CompletableFuture的核心能力。

根据类型，这些方法可以总结为以下四类，其他大部分方法都是基于这四种类型的变种：

Supply

Apply

Accept

Run

关于方法的变种

上述接种类型的方法一般都有三个变种方法：同步、异步和指定线程池。比如，

thenApply()

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

下面这幅类图，展示了CompletableFuture和Future、CompletionStage以及Completion之间的关系。当然，由于方法众多，这幅图中并没有全部呈现，而是仅选取了部分重要的方法。

三、CompletableFuture的核心用法

前面已经说过，CompletableFuture的核心方法总共分为四类，而这四类方法又分为两种模式：同步和异步。所以，我们从这四类方法中选取了部分核心的API，它们都是我们经常用到的API。

• 同步：使用当前线程运行任务；
• 异步：使用CompletableFuture线程池其他线程运行任务，异步方法的名字中带有

  Async

  .

1. runAsync

runAsync()

当我们想异步运行某个任务时，在以往需要手动实现Thread或者借助Executor实现。而通过runAsync()`就简单多了。比如，我们可以直接传入Runnable类型的任务：

CompletableFuture.runAsync(new Runnable() {
@Override
public void run() {
note("妲己进入草丛蹲点...等待小鲁班出现");
}
});

此外，在Java8及之后的JDK版本中，我们还可以使用lambda表达式进一步简化代码：

CompletableFuture.runAsync(() -> note("妲己进入草丛蹲点...等待小鲁班出现"));

这样看起来是不是很简单？相信很多同学也是采用这样的方式来使用

runAsync()

2. supply与supplyAsync

对于

supply()

supply()

举个例子，在下面的示例代码中，我们通过

supplyAsync()

thenApply()

// 创建nameFuture，返回姓名
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
return "妲己";
});

// 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
return "爱你，" + name;
});

//阻塞获得表白的结果
System.out.println(sayLoveFuture.get()); // 爱你，妲己

你看，一旦理解了

supply()

supplyAsync()

3. thenApply与thenApplyAsync

刚才，在前面我们已经介绍了

supply()

thenApply()

thenApply()

supply()

supply()

// 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
return "爱你，" + name;
});

public <U> CompletableFuture <U> thenApplyAsync(
Function <? super T, ? extends U> fn) {
return uniApplyStage(null, fn);
}

4. thenAccept与thenAcceptAsync

作为

supply()

thenApply()

thenAccept()

thenApply()

thenAccept()

CompletableFuture<Void> sayLoveFuture = nameFuture.thenAccept(name -> {
System.out.println("爱你，" + name);
});

public CompletableFuture < Void > thenAccept(Consumer < ? super T > action) {
return uniAcceptStage(null, action);
}

5. thenRun

thenRun()

public CompletableFuture < Void > thenRun(Runnable action) {
return uniRunStage(null, action);
}

所以，如果你在回调中不想返回任何的结果，只运行特定的逻辑，那么你可以考虑使用

thenAccept

thenRun

6. thenCompose与 thenCombine

以上几种方法都是各玩各的，但

thenCompose()

thenCombine()

编排两个存在依赖关系的任务

在前面的例子中，在接收前面任务的结果时，我们使用的是thenApply(). 也就是说，sayLoveFuture在执行时必须依赖nameFuture的完成，否则执行个锤子。

// 创建Future
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
return "妲己";
});

// 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
return "爱你，" + name;
});

但其实，除了thenApply()之外，我们还可以使用

thenCompose()

// 创建Future
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
return "妲己";
});

CompletableFuture<String> sayLoveFuture2 = nameFuture.thenCompose(name -> {
return CompletableFuture.supplyAsync(() -> "爱你，" + name);
});

可以看到，

thenCompose()

thenApply()

• thenApply()

  ：返回计算结果的原始类型，比如返回String;

• thenCompose()

  ：返回CompletableFuture类型，比如返回CompletableFuture.

组合两个相互独立的任务

考虑一个场景，当我们在执行某个任务时，需要其他任务就绪才可以，应该怎么做？这样的场景并不少见，我们可以使用前面学过的并发工具类实现，也可以使用

thenCombine()

举个例子，当我们计算某个英雄（比如妲己）的胜率时，我们需要获取她参与的总场次（rounds），以及她获胜的场次（winRounds），然后再通过

winRounds / rounds

CompletableFuture < Integer > roundsFuture = CompletableFuture.supplyAsync(() -> 500);
CompletableFuture < Integer > winRoundsFuture = CompletableFuture.supplyAsync(() -> 365);

CompletableFuture < Object > winRateFuture = roundsFuture
.thenCombine(winRoundsFuture, (rounds, winRounds) -> {
if (rounds == 0) {
return 0.0;
}
DecimalFormat df = new DecimalFormat("0.00");
return df.format((float) winRounds / rounds);
});
System.out.println(winRateFuture.get());

thenCombine()

7. allOf与anyOf

allOf()

anyOf()

• allOf()

  ：给定一组任务，等待所有任务执行结束；

• anyOf()

  ：给定一组任务，等待其中任一任务执行结束。

allOf()

anyOf()

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

需要注意的是，

anyOf()

allOf()

allOf()

anyOf()

sleep

CompletableFuture < Integer > roundsFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(200);
return 500;
} catch (InterruptedException e) {
return null;
}
});
CompletableFuture < Integer > winRoundsFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(100);
return 365;
} catch (InterruptedException e) {
return null;
}
});

CompletableFuture < Object > completedFuture = CompletableFuture.anyOf(winRoundsFuture, roundsFuture);
System.out.println(completedFuture.get()); // 返回365

CompletableFuture < Void > completedFutures = CompletableFuture.allOf(winRoundsFuture, roundsFuture);

在CompletableFuture之前，如果要实现所有任务结束后执行特定的动作，我们可以考虑CountDownLatch等工具类。现在，则多了一选项，我们也可以考虑使用

CompletableFuture.allOf

四、CompletableFuture中的异常处理

对于任何框架来说，对异常的处理都是必不可少的，CompletableFuture当然也不会例外。前面，我们已经了解了CompletableFuture的核心方法。现在，我们再来看如何处理计算过程中的异常。

考虑下面的情况，当

rounds=0

CompletableFuture < ? extends Serializable > winRateFuture = roundsFuture
.thenCombine(winRoundsFuture, (rounds, winRounds) -> {
if (rounds == 0) {
throw new RuntimeException("总场次错误");
}
DecimalFormat df = new DecimalFormat("0.00");
return df.format((float) winRounds / rounds);
});
System.out.println(winRateFuture.get());

在CompletableFuture链式调用中，如果某个任务发生了异常，那么后续的任务将都不会再执行。对于异常，我们有两种处理方式：

exceptionally()

handle()

1. 使用exceptionally()回调处理异常

在链式调用的尾部使用

exceptionally()

exceptionally()

CompletableFuture < ? extends Serializable > winRateFuture = roundsFuture
.thenCombine(winRoundsFuture, (rounds, winRounds) -> {
if (rounds == 0) {
throw new RuntimeException("总场次错误");
}
DecimalFormat df = new DecimalFormat("0.00");
return df.format((float) winRounds / rounds);
}).exceptionally(ex -> {
System.out.println("出错：" + ex.getMessage());
return "";
});
System.out.println(winRateFuture.get());

2. 使用handle()处理异常

除了

exceptionally()

handle()

exceptionally()

handle()

handle()

if (ex != null)

CompletableFuture < ? extends Serializable > winRateFuture = roundsFuture
.thenCombine(winRoundsFuture, (rounds, winRounds) -> {
if (rounds == 0) {
throw new RuntimeException("总场次错误");
}
DecimalFormat df = new DecimalFormat("0.00");
return df.format((float) winRounds / rounds);
}).handle((res, ex) -> {
if (ex != null) {
System.out.println("出错：" + ex.getMessage());
return "";
}
return res;
});
System.out.println(winRateFuture.get());

当然，如果我们允许某个任务发生异常而不中断整个调用链路，那么可以在其内部通过

try-catch

五、CompletableFuture中的线程池

在前面我们已经说过CompletableFuture中的任务有同步、异步和指定线程池三个变种。比如，当我们调用

thenAccept()

thenAcceptAsync()

答案是ForkJoinPool.commonPool()，我们熟悉的老朋友又回来了，还是它。当需要新的线程时，CompletableFuture会从commonPool中获取线程，相关源码如下：

public static CompletableFuture<Void> runAsync(Runnable runnable) {
return asyncRunStage(asyncPool, runnable);
}
private static final Executor asyncPool = useCommonPool ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

可问题是，我们已经知道了commonPool的潜在风险，在生产环境中使用无异于给自己挖坑。那怎么办呢？当然是自定义线程池，如此重要的东西务必要掌握在自己的手上。换句话说，当我决定使用CompletableFuture的时候，默认就是我们要创建自己的线程池。不要偷懒，更不要存在侥幸心理。

CompletableFuture中每个核心类型的方法都提供了自定义线程池的重载，使用起来也较为简单：

// supplyAsync中可以指定线程池的方法
public static < U > CompletableFuture < U > supplyAsync(Supplier < U > supplier,
Executor executor) {
return asyncSupplyStage(screenExecutor(executor), supplier);
}

// 自定义线程池示例
Executor executor = Executors.newFixedThreadPool(10);

CompletableFuture < Integer > roundsFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(200);
return 500;
} catch (InterruptedException e) {
return null;
}
}, executor);

小结

至此，关于CompletableFuture的讲解已经全部结束。我们已经知道，CompletableFuture是Future的扩展和增强，并提供了大量强大且好玩的优秀特性。这些特性可以帮助我们优雅地解决一些场景问题，而在此之前我们要实现相同的方案可能要花费很大的代价。

当然，CompletableFuture这朵玫瑰虽然很漂亮，但它的刺也同样尖锐，它并不是天生完美。因此，在使用CompletableFuture时仍要遵循一些必要的约束：

• 自定义线程池：当你决定在生产环境使用CompletableFuture的时候，你应该同时准备好对应的线程池策略，而不是偷懒地使用默认的线程池；
• 团队共识：技术就是这样，好与不好总是会有不同的标准。当你说好的时候，你的队友可能并不这么认为，反之你也可能也会反对某种技术观点。因此，当你决定采用CompletableFuture的时候，最好和团队同步你的策略，让大家都了解它的优点和潜在的风险，各行其是绝对不是好的策略。

最后，CompletableFuture的源码有近2400行，并且有大量的API. 说实话，在王者系列所分析的源码文章中，CompletableFuture的源码是截止目前最难以理解的。如果将源码展开讲解的话，大概需要数万字，这将直接劝退百分之九十以上的读者。所以，我们也不建议你硬啃所有的源码，而是建议在归纳分类的基础上，有针对性地掌握它的重点部分。当然，如果你饶有兴趣地读完了它所有的源码，在此给你点赞。

正文到此结束，恭喜你又上了一颗星✨

夫子的试炼

• 动手：编写代码体验

  runAsync()

  的用法，并指定线程池。

延伸阅读与参考资料

• 《王者并发课》大纲与更新进度总览：https://juejin.cn/post/6967277362455150628
• https://thepracticaldeveloper.com/differences-between-completablefuture-future-and-streams/#conclusions-pros-and-cons-of-completablefuture

关于作者

从业近十年，先后从事敏捷与DevOps咨询、Tech Leader和管理等工作，对分布式高并发架构有丰富的实战经验。热衷于技术分享和特定领域书籍翻译，掘金小册《高并发秒杀的设计精要与实现》作者。

关注公众号【MetaThoughts】，及时获取文章更新和文稿。

如果本文对你有帮助，欢迎点赞、关注、监督，我们一起从青铜到王者。