写爽了但我要先提醒一件事本文所有代码示例基于Java 8及以上版本部分特性如orTimeout需要Java 9。如果你还在用Java 8请直接跳过我标注Java 9的部分不影响主流程。先聊个我最近被问烂的问题很多人写异步代码上来就是new Thread(() - {...}).start()或者用线程池Future然后future.get()卡在那干等。这不算真正的异步顶多算“换了个线程同步”。真正的异步编程应该让任务之间形成依赖链能并行就并行能回调就回调主线程绝不傻等。Java 8之后CompletableFuture就是干这件事的标准答案也是目前Java并发编程里使用频率最高的异步编排工具。无论是你自己写业务代码、看开源项目源码还是准备Java面试特别是“八股文”环节它都是绕不过去的考点。这篇文章我打算彻底拆一遍CompletableFuture先讲清楚它到底解决了什么再把常用API按场景分类最后给一个接近真实业务的订单编排案例以及我实际踩过的坑和排查经验。我不是那种只讲概念的博主所以下面的内容全部是能直接抄到项目里的东西。1. 为什么选择CompletableFuture异步编程的痛点与解决方案1.1 Future的局限从get()阻塞说起Java 5引入Future接口当时看起来很美提交一个任务到线程池拿到一个Future对象后续调用get()获取结果。但用过的都知道这玩意有两个让人抓狂的设计。第一get()是同步阻塞的。任务如果跑了2秒调用线程就傻等2秒。你当然可以设置超时比如get(2, TimeUnit.SECONDS)但超时之后你拿到的是异常不是结果。换句话说Future无法让你优雅地处理“任务还没完成我先去做别的事”这种场景。第二Future无法直接表达任务之间的依赖关系。比如任务A的结果要传给任务B你需要先get()拿到A的结果再手动提交B。如果B还要等C那你只能在代码里写一串串的阻塞调用本质还是同步串行性能和体验都谈不上好。FutureString future executor.submit(() - doRequest()); String result future.get(); // 这里卡住了这种写法在单一异步任务下还能应付一旦遇到“查库存、扣减库存、生成订单、发通知”这种业务链路代码就会变成一堆get和嵌套丑不说还特别难排查。我在上一个项目里接手过一个用Future写的聚合查询接口一个请求内部有6个串行阻塞点接口延迟高得离谱后来全部用CompletableFuture重写耗时直接砍掉三分之一。1.2 CompletableFuture能做什么从单任务到任务编排CompletableFuture是Java 8在CompletionStage基础上实现的类它解决了上面两个核心痛点。首先它本身就是Future所以你可以继续拿它获取结果但更重要的它是CompletionStage意思是“一个任务完成之后可以自动触发下一个任务”。你可以像搭积木一样把多个异步任务串成流水线也可以把多个并行任务的结果合并成一个最终结果。整个过程无需手动阻塞等待任务完成时会自动通知监听者执行后续动作。我习惯把它的核心价值总结成三个词编排、回调、聚合。编排任务之间可以串行依赖也可以并行执行。回调任务完成后自动触发后续动作不用主线程等。聚合多个异步任务并行跑完后统一合并结果。实际项目里90%的异步场景都能用这三个词覆盖。比如你做一个商品详情页需要同时查商品基本信息、库存、评论这三个请求互不依赖用allOf聚合但如果你需要先拿用户信息再拿用户订单最后计算订单金额这就是串行依赖交给thenCompose。CompletableFutureProductInfo infoFuture CompletableFuture.supplyAsync(() - queryProduct()); CompletableFutureStock stockFuture CompletableFuture.supplyAsync(() - queryStock()); CompletableFutureComment commentFuture CompletableFuture.supplyAsync(() - queryComment()); CompletableFutureVoid all CompletableFuture.allOf(infoFuture, stockFuture, commentFuture);熟悉这种思维之后你会发现原来写同步顺序的代码现在只需要考虑“哪些任务必须等待前一个哪些任务可以并行”这是本质区别。2. 核心API拆解构建异步链路的基础积木2.1 创建异步任务runAsync和supplyAsync创建CompletableFuture的方式有两类不返回结果的和返回结果的。// 不返回结果执行Runnable CompletableFutureVoid f1 CompletableFuture.runAsync(() - System.out.println(执行任务)); // 返回结果执行Supplier CompletableFutureString f2 CompletableFuture.supplyAsync(() - 返回结果);它们都支持传入线程池比如CompletableFuture.supplyAsync(() - ..., executor)。不传的话默认使用ForkJoinPool.commonPool()。这里有个重要细节我建议生产环境一律显式传线程池。默认的commonPool是全局共享的会被其他并行流任务影响任务变多时容易互相阻塞而且它默认的并行度是CPU核数减1对于IO密集的异步任务来说太小了后面章节我会专门说线程池。2.2 结果转换与消费thenApply、thenAccept、thenRun任务创建之后就要开始编排下一步了。thenApply是对结果进行转换返回新的CompletableFuturethenAccept是消费结果不返回新结果thenRun是执行一段不关心结果的代码。CompletableFutureString future CompletableFuture.supplyAsync(() - hello) .thenApply(s - s world) // 结果转换hello world .thenApply(String::toUpperCase); // 结果转换HELLO WORLD future.thenAccept(System.out::println); // 消费结果打印 future.thenRun(() - System.out.println(全部完成)); // 执行额外动作很多人分不清thenApply和thenCompose我简单说一下。thenApply接收的是一个普通函数返回的是普通值CompletableFuture会自动帮你包装成新的CompletableFuture。thenCompose接收的函数返回的是一个CompletableFuture用于展开嵌套的异步任务。如果你在thenApply里又创建了异步任务就会出现CompletableFuture嵌套代码会变成CompletableFutureCompletableFutureT非常恶心。这种场景就应该用thenCompose。// 错误示范嵌套 CompletableFutureCompletableFutureString bad CompletableFuture .supplyAsync(() - token) .thenApply(token - CompletableFuture.supplyAsync(() - callApi(token))); // 正确姿势展开 CompletableFutureString good CompletableFuture .supplyAsync(() - token) .thenCompose(token - CompletableFuture.supplyAsync(() - callApi(token)));2.3 任务编排thenCompose与thenCombinethenCompose解决的是串行依赖thenCombine解决的是并行合并。打个比方先登录拿到token再用token查用户信息这是串行用thenCompose同时查用户基本信息和用户订单最后合并展示这是并行用thenCombine。thenCombine会把两个任务的结果作为参数传入函数然后返回一个新结果。CompletableFutureString userFuture CompletableFuture.supplyAsync(() - getUserInfo()); CompletableFutureListOrder orderFuture CompletableFuture.supplyAsync(() - getOrders()); CompletableFutureUserDetail detailFuture userFuture.thenCombine(orderFuture, (user, orders) - { UserDetail detail new UserDetail(); detail.setUser(user); detail.setOrders(orders); return detail; });注意thenCombine的两个任务会被同时触发吗不一定。如果在已有的future上调用thenCombine第一个任务已经在跑了第二个任务会在调用thenCombine时开始执行。如果你想同时启动两个独立任务然后合并更好的做法是像上面这样分别创建两个CompletableFuture再调用thenCombine。2.4 多任务聚合allOf和anyOfallOf等待所有任务完成返回CompletableFutureVoid本身不携带结果。你需要手动在回调里逐个获取结果或者把结果集中到一个容器里。这里有个实战小技巧allOf返回的是Void如果每个子任务都有返回值我会用thenApply遍历任务列表逐个join取出结果。ListCompletableFutureString futures new ArrayList(); for (int i 0; i 10; i) { int idx i; futures.add(CompletableFuture.supplyAsync(() - fetchData(idx), executor)); } CompletableFutureListString allResults CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .thenApply(v - futures.stream().map(CompletableFuture::join).collect(Collectors.toList()));注意这里用join()而不是get()。join()不抛出受检异常适合在Lambda表达式里使用get()会抛InterruptedException和ExecutionException在Lambda里写起来很啰嗦。下面我会提到join()抛的是CompletionException你在异常处理时要注意这个区别。anyOf则是只要有一个任务完成就返回那个任务的结果返回类型是CompletableFutureObject。最典型的场景是“多个服务提供同一接口谁先返回就用谁”这还能顺带做容灾降级。3. 实战一个订单异步编排的完整案例3.1 场景拆分与线程池选择理论说多了容易飘上一个能直接参考的业务场景。假设你有一个下单接口流程如下校验用户身份拿到userId。并行查询商品信息、库存信息、用户优惠券。检查库存是否充足计算优惠后价格。创建订单扣减库存发消息通知。其中第1步是第2步的前置条件第2步内部的三个查询互相独立第3步依赖第2步的所有结果第4步又依赖第3步。如果用同步代码写这个接口的耗时大概等于所有步骤之和。如果第2步的三个查询串行那就更惨。用CompletableFuture的意图很明确第2步并行第3和第4步串行依赖。首先线程池不能用默认的commonPool。下单接口是典型的混合负载涉及IO数据库查询、Redis操作、消息发送线程会大量等待所以线程池要偏大。我一般在服务里单独建一个异步任务线程池根据业务情况配置。private final ThreadPoolExecutor orderAsyncExecutor new ThreadPoolExecutor( 8, // corePoolSize 16, // maximumPoolSize 60L, TimeUnit.SECONDS, // 非核心线程空闲回收时间 new LinkedBlockingQueue(1000), // 等待队列 new ThreadFactoryBuilder().setNameFormat(order-async-%d).build(), new ThreadPoolExecutor.CallerRunsPolicy() );线程数为什么定8到16这是一个估算区间的中间值。如果这台机器是4核8线程IO密集型的典型建议是CPU核心数 * 2到CPU核心数 * 4之间8到16正好。队列1000是为了一时大量流量进来时不至于直接打满线程池。拒绝策略用CallerRunsPolicy意思是队列满了之后新任务由提交任务的线程自己执行这样既能放慢生产速度又不丢任务。3.2 核心代码实现并行查询、串行依赖与异常兜底我先把核心流程写出来注意看异步编排的走向。CompletableFutureLong userFuture CompletableFuture.supplyAsync(() - validateAndGetUserId(request), orderAsyncExecutor); CompletableFutureOrderCheckResult checkFuture userFuture.thenCompose(userId - { CompletableFutureProduct productFuture CompletableFuture.supplyAsync(() - queryProduct(request.getProductId()), orderAsyncExecutor); CompletableFutureStock stockFuture CompletableFuture.supplyAsync(() - queryStock(request.getProductId()), orderAsyncExecutor); CompletableFutureCoupon couponFuture CompletableFuture.supplyAsync(() - queryCoupon(userId), orderAsyncExecutor); return CompletableFuture.allOf(productFuture, stockFuture, couponFuture) .thenApply(v - new OrderCheckResult( productFuture.join(), stockFuture.join(), couponFuture.join() )); }); CompletableFutureOrderResult resultFuture checkFuture.thenCompose(check - { if (check.getStock().getAvailable() request.getCount()) { return CompletableFuture.failedFuture(new BizException(库存不足)); // Java 9 } BigDecimal finalPrice calculatePrice(check.getProduct(), check.getCoupon(), request.getCount()); return CompletableFuture.supplyAsync(() - createOrder(check.getUser(), finalPrice, request.getCount()), orderAsyncExecutor) .thenCompose(order - CompletableFuture.supplyAsync(() - afterCreateOrder(order), orderAsyncExecutor)); });这段代码讲几个点。第一userFuture.thenCompose把userId传给下一阶段内部再创建三个并行任务用allOf等它们全部完成。三个并行查询的关键在于三个supplyAsync是顺序创建的但一旦创建就会立刻提交到线程池并发执行所以总体耗时约等于最慢的一个查询而不是三者之和。第二checkFuture.thenCompose里我先判断库存不够就用failedFuture直接返回一个已失败的CompletableFuture。这个API在Java 9才有如果你还在Java 8可以用CompletableFuture.supplyAsync(() - { throw new BizException(库存不足); })代替。第三创建订单和后续通知是串行依赖我用嵌套的thenCompose把它们串成一条流水线。上面的写法把afterCreateOrder放在第二个thenCompose里实际上就是告诉CompletableFuture等createOrder完成后再执行通知。3.3 超时控制与降级方案异步避不开超时。如果依赖的下游服务挂了你的异步任务可能一直挂起线程池被占满整个接口都拖垮。Java 9提供了orTimeout和completeOnTimeout直接给异步任务设置超时。CompletableFutureProduct productFuture CompletableFuture .supplyAsync(() - queryProduct(request.getProductId()), orderAsyncExecutor) .orTimeout(3, TimeUnit.SECONDS);orTimeout在超时后会让future以TimeoutException异常完成。如果想让超时后走一个默认值使用completeOnTimeout(value, timeout, unit)它会在超时后直接用默认值完成任务不会再抛异常。上面的案例里我会给库存查询单独加一个completeOnTimeout比如默认库存为0这样即使库存服务超时流程也能继续走只是最终判定为无货不会因为一个下游接口拖垮整体。CompletableFutureStock stockFuture CompletableFuture .supplyAsync(() - queryStock(request.getProductId()), orderAsyncExecutor) .completeOnTimeout(Stock.empty(), 3, TimeUnit.SECONDS);如果你是Java 8没有这两个方法也可以用get(long timeout, TimeUnit)在外面包装或者用ScheduledExecutorService在超时后调用completeExceptionally但会比较繁琐。我实际经验是尽量升级到Java 11以上这两个方法能省不少代码也更安全。4. 异常处理与调试异步任务中最容易踩的坑4.1 异常传播机制exceptionally、handle、whenComplete异步任务里一旦抛异常不会立刻把调用线程炸掉而是会封装成CompletionException一直往后传递直到被某个专门处理异常的方法捕获。常用的异常处理API有三个exceptionally可以拿到异常然后返回一个降级结果适合“出错了给默认值”。handle不管成功还是失败都会调用能拿到结果或者异常适合“不管结果如何都要处理”。whenComplete也都能拿到但它返回的CompletableFuture会保留原始结果或异常不会改变结果。看代码区分CompletableFutureString future1 CompletableFuture .supplyAsync(() - { if (true) throw new RuntimeException(boom); return ok; }) .exceptionally(ex - default); CompletableFutureString future2 CompletableFuture .supplyAsync(() - ok) .handle((res, ex) - ex null ? res handled : default);有一个非常隐蔽的坑exceptionally只能捕获它之前的任务链上的异常如果在exceptionally之后还有thenApply而这个thenApply又抛了异常那后面的调用方依然会收到新的异常。很多人忽略这一点导致降级没有完全覆盖整个链路。CompletableFutureString future CompletableFuture .supplyAsync(() - { throw new RuntimeException(boom); }) .exceptionally(ex - default) .thenApply(s - { throw new RuntimeException(another boom); }); // 最终future仍然会异常所以我的习惯是降级处理尽量放在链路末端或者在每个可能抛异常的中间环节都加exceptionally不要指望一个兜底能罩住所有。4.2 一个掩盖异常的真实事故我有个朋友在生产环境遇到一个诡异问题接口偶尔返回成功但数据却没更新。排查了很久最后发现代码里这样写的CompletableFuture.runAsync(() - updateDb(), executor); return success;这个写法的问题在于runAsync返回的CompletableFuture被直接丢弃了没人调用get或join去检查结果。如果updateDb()抛异常线程池会捕捉到异常但因为没有地方接收异常就被静默吞掉了。接口当然返回成功数据没更新。这是CompletableFuture使用中最常见的事故一定要记住异步任务不是“发个消息就完了”如果任务的执行结果会影响业务你必须对返回的future做消费或异常处理。最稳妥的做法是对不关心结果的异步事务至少加一个whenComplete日志兜底CompletableFuture.runAsync(() - updateDb(), executor) .whenComplete((v, ex) - { if (ex ! null) { log.error(updateDb failed, ex); } });如果需要知道最终结果调用方还要在合适的时机join()或get()不能只是“提交完就不管”。4.3 排查技巧线程转储、日志埋点与任务链路追踪异步任务排查比同步麻烦因为你不知道任务到底在哪个线程执行异常又在哪一层被包装。我常用的排查手法有三个。第一线程池线程名一定要设置。默认线程名是pool-1-thread-1这种排障时根本看不出是哪个业务的任务。我在上面的例子用ThreadFactoryBuilder自定义了order-async-%d一旦线程转储一眼就能看出是订单模块的异步线程。第二日志要带链路标识。如果你的服务接入了类似MDC、TraceId的链路追踪异步线程默认不会继承主线程的MDC你需要在提交任务时手动把TraceId传进去否则日志穿插在一起根本对不上。MapString, String contextMap MDC.getCopyOfContextMap(); CompletableFuture.supplyAsync(() - { if (contextMap ! null) { MDC.setContextMap(contextMap); } try { return doSomething(); } finally { MDC.clear(); } }, executor);第三怀疑线程池出问题时直接jstack看线程堆栈重点看哪些线程卡在ForkJoinPool.commonPool或者自定义线程池的AQS队列里。如果大量线程处于等待状态说明线程池被某个慢任务占满这时候检查任务有没有设置超时是不是某个下游调用没有超时时间。5. 性能调优与线程池配置实战5.1 默认线程池的隐患与自定义线程池前面我说过默认commonPool不适合生产再展开细说。ForkJoinPool.commonPool()的线程数通常是Runtime.getRuntime().availableProcessors() - 1。在容器或云服务器上你看到的是4核8线程那线程数就是7对IO任务来说太小。更麻烦的是如果任务内部有parallelStream或者递归ForkJoin任务它们都共享这个池很容易互相拖累。还有一个问题是commonPool不会被独立监控。你无法方便地知道它的队列长度、线程活跃数、拒绝次数。如果线上出了线程池打满的问题日志里几乎看不到任何线索。所以我的结论很直接任何承载真实业务的CompletableFuture都隐式或显式传入自定义线程池。就算你现在只有一个简单场景也建议建一个线程池因为后期加任务编排时再改线程池需要动很多处调用。5.2 线程池参数计算与拒绝策略线程池参数的经典计算方式网上很多我结合异步场景给一个实际思路。先判断任务类型。如果是CPU密集线程数可以是CPU核心数1每个线程占满一个核心再多只会增加上下文切换。如果是IO密集线程数可以估算为CPU核心数 * (1 平均等待时间 / 平均计算时间)。最粗暴的经验值是CPU核心数 * 2到CPU核心数 * 4。订单场景里有大量数据库操作和外部调用等待时间远大于计算时间所以8到16是合理的。如果你的机器核心更多比如16核可以适当上调到32左右。队列长度也很关键。队列太长会让请求延迟变得不可控因为任务先进队列线程池线程忙完才处理。我一般建议队列长度不要超过maximumPoolSize * 4比如上面max是16队列1000其实偏大但好处是突发流量不会立刻触发拒绝坏处是排队任务可能超时。要结合你的超时设置来看如果超时是3秒队列超过几百就可能积压这时要削队列或增加线程数。拒绝策略我再说一下。AbortPolicy会抛异常让上游感受到压力CallerRunsPolicy让提交线程自己执行任务起到“反向背压”的效果DiscardPolicy和DiscardOldestPolicy会丢任务除非你能接受丢数据否则不要用。我在大多数业务场景用CallerRunsPolicy毕竟丢用户请求不可接受。5.3 上下文传递与安全发布异步线程之间的上下文传递是一个非常容易被忽视的性能与正确性问题。常见需要传递的上下文有TraceId、用户信息、地域标识等。如果Context只是存在ThreadLocal里异步线程拿不到整个链路追踪就断了。上面我给了MDC传递的例子再补充一点不要裸手写这段代码太容易漏。建议封装一个静态方法接收一个SupplierT内部负责保存和恢复上下文或者使用现成的TransmittableThreadLocal这类工具它专门解决线程池场景下ThreadLocal值传递问题。public static T SupplierT withContext(SupplierT supplier) { MapString, String contextMap MDC.getCopyOfContextMap(); return () - { if (contextMap ! null) { MDC.setContextMap(contextMap); } try { return supplier.get(); } finally { MDC.clear(); } }; }使用的时候CompletableFuture.supplyAsync(withContext(() - doSomething()), executor);这个封装能保证提交任务时捕获主线程的MDC上下文并在异步线程中恢复。我后续所有案例里只要涉及全局上下文都建议用类似方式包一层否则异步日志会乱到让你怀疑人生。6. 常见问题排查速查表6.1 问题与对策表格我把实战中最常遇到的情况整理成一张速查表方便你出问题时直接对照。现象可能原因解决方案接口偶尔返回成功但数据没更新异步任务异常被吞没有消费future用whenComplete或exceptionally兜底禁止丢弃future任务全部挤在一起响应变慢使用了默认commonPool线程数过少改成自定义线程池按任务类型调整线程数所有异步任务都卡住不执行线程池被慢任务占满或者任务没有超时给每个下游调用设置超时检查线程池活跃线程日志里TraceId错乱或缺失异步线程没有继承MDC上下文提交任务时传递MDC用withContext封装某个任务抛异常后主流程仍然成功异常在链路末端没有被处理在链路末端统一handle或exceptionallyallOf之后获取结果发现是null忘记join或误用了Void返回值在thenApply中遍历future调用join汇总结果使用了get()导致受检异常写起来很啰嗦在Lambda里不能方便处理异常使用join()代替get()或包装成CompletionException处理这个表格里的第一行就是我最想强调的。很多人用异步就是为了“快”结果把正确性丢了。记住一句话异步不等于可以不管结果你只是把等待交给了回调但结果和异常必须有人处理。6.2 面试中CompletableFuture的高频考点既然标题里提到了面试我把Java面试里关于CompletableFuture的高频考点整理一下都是真实会被追问的问题。CompletableFuture和Future的区别是什么不只要说出回调还要说出任务编排和异常处理。thenApply和thenCompose的区别这个几乎是必考回答时要强调嵌套展开与扁平化。allOf和anyOf的区别结合业务举例比纯概念好得多。默认线程池是什么为什么不建议使用能答出commonPool的并行度是CPU核数减1并且共享池容易互相影响就很有说服力。get()和join()的区别get抛受检异常join不抛受检异常但会包装成CompletionException。如何处理异步任务异常至少要列出exceptionally、handle、whenComplete三种并说明区别。如何超时控制Java 9的orTimeout和completeOnTimeoutJava 8的替代方案也要会。面试官问这些的核心目的是考察你有没有真正在项目里用过而不是背概念。如果你能像我前面那样把订单场景、线程池配置、异常吞掉案例讲出来基本就稳了。单纯背API只能拿个及格分真正能加分的是你说出“任务编排时如何避免嵌套”“异步任务异常丢失怎么排查”这类踩坑经验。最后分享一点实际使用体会如果要用一句话总结CompletableFuture我会说它让你终于可以把“并发”当业务逻辑来写而不是当底层技术来抠。我在项目里实践了几年最大的体会是异步编排让代码看起来优雅了但代价是调试难度上升了一个量级。所以每当你用CompletableFuture写出一个漂亮的长链路时一定要额外花时间把异常处理、超时控制、上下文传递、日志埋点这几件事做完整否则线上迟早会还回来。另一个比较隐蔽的体会是并不是所有场景都适合异步编排如果任务本身很快、又高度串行强行拆成异步反而增加线程切换和代码理解成本。我的原则是“并行提升明显、任务有等待”才上CompletableFuture其他情况宁可用同步保持简单。希望这篇实战指南对你有用踩过的坑都帮你提前踩了接下来你可以放心去写。
