上一章介绍FutureTask时我们已经知道异步任务执行结束后正常结果或异常会保存在FutureTask中提交任务的线程可以通过get()取得最终结果。例如FutureUseruserFutureexecutorService.submit(()-queryUser());UseruseruserFuture.get();这段代码可以取得异步结果。但如果查询用户后还要根据用户编号查询订单普通Future的写法就会变成FutureUseruserFutureexecutorService.submit(()-queryUser());UseruseruserFuture.get();FutureListOrderorderFutureexecutorService.submit(()-queryOrders(user.getId()));ListOrderordersorderFuture.get();这里的问题不是Future不能保存结果而是它不擅长描述多个异步任务之间的关系。调用线程必须先等待用户查询完成取得User再提交订单查询任务。任务链越长调用线程越容易变成“提交一个、等待一个、再提交下一个”。CompletableFuture要解决的核心问题是能不能提前描述任务之间的依赖关系让前一个阶段完成后自动推动后一个阶段而不是由调用线程反复调用get()来手动推进。一、CompletableFuture 改变了什么普通Future更适合表示单个异步任务的结果。调用线程提交任务后拿到一个Future以后什么时候取结果由调用线程主动调用get()决定。CompletableFuture更适合描述多个异步阶段之间的关系。调用线程不一定要马上等待结果而是可以提前登记后续操作。CompletableFutureStringfutureCompletableFuture.supplyAsync(()-queryUser()).thenApply(user-user.getName());这段代码表达的是先异步查询用户查询完成后把User交给下一阶段下一阶段读取user.getName()得到String。调用线程执行thenApply()时用户查询可能还没有完成thenApply()此时只是登记一个后续阶段。等前一阶段真正完成后CompletableFuture会自动把结果传给下一阶段。普通FutureCompletableFuture调用线程主动等待结果调用线程提前登记后续阶段调用线程取得结果后亲自提交下一步前一阶段完成后自动触发下一阶段更适合单个异步任务更适合多个异步任务编排任务链越长阻塞等待代码越多可以连续描述完整任务链需要注意CompletableFuture没有消除任务依赖。如果任务 B 必须使用任务 A 的结果那么 B 仍然必须等 A 完成。它改变的是“谁来推动下一步”从调用线程手动等待并提交变成由完成事件自动触发后续阶段。二、如何创建任务链的起点任务链必须先有一个起点。CompletableFuture常用两个方法创建第一个阶段方法接收接口是否有业务返回值返回类型runAsync()Runnable否CompletableFutureVoidsupplyAsync()SupplierT是CompletableFutureT如果异步任务只需要执行一段操作不需要产生业务结果可以使用runAsync()CompletableFutureVoidfutureCompletableFuture.runAsync(()-{sendMessage();});runAsync()接收的是Runnable它的run()没有返回值。因此返回的是CompletableFutureVoid。这里的Void只表示没有业务结果不表示这个任务无法被等待、取消或继续编排。如果异步任务执行后需要产生结果可以使用supplyAsync()CompletableFutureUserfutureCompletableFuture.supplyAsync(()-{returnqueryUser();});supplyAsync()接收的是SupplierT它的get()会产生一个结果T。SupplierT本身没有参数但 Lambda 可以使用外部已经存在的变量longuserId1001L;CompletableFutureUserfutureCompletableFuture.supplyAsync(()-queryUser(userId));这里的userId不是Supplier.get()的参数而是 Lambda 捕获的外部变量。三、一个阶段完成后下一步怎么写一个阶段正常完成后下一步通常有三种需求使用上一步结果并产生新结果使用上一步结果但不产生新结果不使用上一步结果只需要等它完成。它们分别对应thenApply()、thenAccept()和thenRun()。方法是否使用上一步结果是否产生新结果接收接口thenApply()是是FunctionT, RthenAccept()是否ConsumerTthenRun()否否RunnablethenApply()适合做结果转换CompletableFutureIntegerfutureCompletableFuture.supplyAsync(()-10).thenApply(value-value*2);第一阶段产生10第二阶段接收10并返回20。任务链中的结果类型也可以继续变化CompletableFutureStringfutureCompletableFuture.supplyAsync(()-10).thenApply(value-value*2).thenApply(value-结果是value);thenAccept()适合消费结果但不再产生新结果CompletableFutureVoidfutureCompletableFuture.supplyAsync(()-10).thenAccept(value-{System.out.println(value);});thenRun()只关心前一阶段已经正常完成不关心前一阶段产生了什么CompletableFutureVoidfutureCompletableFuture.supplyAsync(()-10).thenRun(()-{System.out.println(任务链结束);});选择这三个方法时不需要先背方法名可以先问两个问题下一步是否需要上一步结果下一步是否要产生新结果这两个问题决定了该用thenApply()、thenAccept()还是thenRun()。四、下一步本身也是异步任务时为什么要用 thenCompose假设已经有两个异步方法CompletableFutureUserqueryUserAsync(){returnCompletableFuture.supplyAsync(()-queryUser());}CompletableFutureListOrderqueryOrdersAsync(longuserId){returnCompletableFuture.supplyAsync(()-queryOrders(userId));}订单查询依赖用户编号因此必须先查询用户再根据用户编号查询订单。如果使用thenApply()CompletableFutureCompletableFutureListOrdernestedFuturequeryUserAsync().thenApply(user-queryOrdersAsync(user.getId()));这里会得到两层CompletableFuture。原因是thenApply()会把 Lambda 的返回值当作当前阶段的普通结果而 Lambda 返回的本身就是一个CompletableFutureListOrder。这表示外层任务完成时只能说明queryOrdersAsync()已经被调用并返回了一个新的异步对象并不表示订单查询已经完成。后续如果要取得订单可能需要处理两层结果。这种嵌套不方便继续编排因此应该使用thenCompose()CompletableFutureListOrderordersFuturequeryUserAsync().thenCompose(user-queryOrdersAsync(user.getId()));thenCompose()的作用是连接两个前后依赖的异步阶段并把嵌套结构展平成一层。判断规则可以简化为Lambda 返回值应使用的方法普通结果RthenApply()CompletableFutureRthenCompose()如果订单查询是普通同步方法ListOrderqueryOrders(longuserId)那就不需要thenCompose()直接用thenApply()即可CompletableFutureListOrderordersFuturequeryUserAsync().thenApply(user-queryOrders(user.getId()));所以thenCompose()不是所有前后依赖任务都必须使用。它专门解决“下一步本身已经返回CompletableFuture”时的嵌套问题。五、两个独立任务如何并行汇合thenCompose()表示前后依赖任务 B 必须使用任务 A 的结果才能启动。有些任务彼此独立可以同时启动最后再汇合。例如查询用户资料和查询优惠券互不依赖CompletableFutureUseruserFutureCompletableFuture.supplyAsync(()-queryUser());CompletableFutureCouponcouponFutureCompletableFuture.supplyAsync(()-queryCoupon());如果两个任务各耗时一秒依次执行大约需要两秒工作线程充足时同时执行可以接近一秒完成。最终计算价格时需要同时使用两个结果可以使用thenCombine()CompletableFuturePricepriceFutureuserFuture.thenCombine(couponFuture,(user,coupon)-calculatePrice(user,coupon));执行关系可以表示为合并函数只有在两个任务都正常完成后才会执行。它不要求哪个任务先完成只要求两个结果最终都可用。thenCompose()和thenCombine()的区别可以概括为方法任务关系含义thenCompose()前后依赖A 的结果决定如何启动 BthenCombine()并行汇合A 和 B 可以同时执行最后合并结果真正能缩短总耗时的是互不依赖的任务并行执行。前后依赖的任务即使用了异步写法也仍然要按依赖顺序完成。六、如何等待一批异步任务当任务数量较多时如果只是想知道一批任务什么时候全部完成没有必要反复嵌套thenCombine()可以使用allOf()CompletableFutureUseruserFuture...;CompletableFutureCouponcouponFuture...;CompletableFutureOrderorderFuture...;CompletableFutureVoidallFutureCompletableFuture.allOf(userFuture,couponFuture,orderFuture);allOf()返回的是CompletableFutureVoid。它只表示所有任务都已经完成不会自动把各个业务结果收集到一个列表中。各个结果仍然保存在原来的CompletableFuture中。allFuture.join();UseruseruserFuture.join();CouponcouponcouponFuture.join();OrderorderorderFuture.join();后面的几个join()通常不会继续阻塞因为allFuture.join()已经保证这些任务都完成了。如果只关心任意一个任务率先完成可以使用anyOf()CompletableFutureObjectanyFutureCompletableFuture.anyOf(futureA,futureB,futureC);anyOf()返回CompletableFutureObject保存的是最先完成任务的结果。之所以是Object是因为传入任务的结果类型可能不同编译器无法提前确定最先完成的是哪一种结果。需要注意anyOf()完成后其他任务不会自动取消此外最先完成不一定是最先成功。如果某个任务最先异常完成anyOf()也可能直接异常完成。异常处理会放到下一章单独讨论。多任务关系可以这样选择任务关系应使用的方法前一步完成后启动另一个异步任务thenCompose()两个独立任务完成后合并结果thenCombine()等待多个任务全部完成allOf()等待任意任务率先完成anyOf()七、普通方法和 Async 方法有什么区别CompletableFuture中很多方法都有普通版本和Async版本例如thenApply()thenApplyAsync()它们都要等前一阶段完成也都能使用前一阶段结果并产生新结果。区别主要在于后续操作如何被调度。假设主线程创建任务链CompletableFutureIntegerfutureCompletableFuture.supplyAsync(()-{sleep(1000);return10;});CompletableFutureIntegernextfuture.thenApply(value-value*2);如果主线程调用thenApply()时前一阶段还没有完成那么thenApply()只是登记后续操作。之后完成前一阶段的工作线程通常会继续执行这个后续操作。但如果调用thenApply()时前一阶段已经完成CompletableFutureIntegerfutureCompletableFuture.completedFuture(10);CompletableFutureIntegernextfuture.thenApply(value-value*2);那么当前调用thenApply()的线程可能直接执行这个转换。因此普通thenApply()不保证一定在哪个固定线程中执行。thenApplyAsync()会在前一阶段完成后把后续操作重新提交到线程池future.thenApplyAsync(value-slowProcess(value));这会形成一个新的调度边界但不要把Async误解成以下含义Async不代表什么不代表后续任务立即执行不代表可以跳过前置依赖不代表一定新建线程不代表一定更换线程不代表一定提高单条任务链速度如果任务 B 依赖任务 A 的结果那么无论使用thenApply()还是thenApplyAsync()B 都不能在 A 完成前执行。Async改变的是 A 完成以后B 如何被调度。八、为什么要显式指定线程池不显式指定线程池时runAsync()、supplyAsync()以及不带Executor参数的Async方法通常会使用默认公共线程池。CompletableFutureUserfutureCompletableFuture.supplyAsync(()-queryUser());实际项目中更推荐为重要业务显式指定线程池ExecutorServiceexecutorExecutors.newFixedThreadPool(8);CompletableFutureUserfutureCompletableFuture.supplyAsync(()-queryUser(),executor);thenApplyAsync()等方法也可以指定线程池CompletableFutureStringnameFutureuserFuture.thenApplyAsync(user-user.getName(),executor);显式指定线程池的价值在于控制线程数量、任务队列、拒绝策略、线程名称以及不同业务之间的隔离。尤其是可能长时间阻塞的任务不应该不加区分地全部提交到公共线程池。不同任务也可以交给不同线程池。例如数据库查询和 CPU 密集计算可以分开CompletableFuture.supplyAsync(()-queryDatabase(),ioExecutor).thenApplyAsync(data-heavyCalculate(data),cpuExecutor);这表示数据库查询由ioExecutor执行查询完成后的计算由cpuExecutor执行。不同线程池的主要价值是任务隔离、资源匹配和并发数量控制。它不一定缩短一条前后依赖任务链的总耗时如果 A 耗时一秒B 依赖 A 且耗时两秒总耗时仍然接近三秒。九、常见错误用法第一不要在任务链中频繁调用join()或get()来手动推进下一步UseruserqueryUserAsync().join();ListOrderordersqueryOrdersAsync(user.getId()).join();这种写法又回到了普通Future的阻塞编排。更合适的是使用thenCompose()描述依赖关系CompletableFutureListOrderfuturequeryUserAsync().thenCompose(user-queryOrdersAsync(user.getId()));第二当 Lambda 返回CompletableFuture时不要用thenApply()造成嵌套CompletableFutureCompletableFutureListOrderfuturequeryUserAsync().thenApply(user-queryOrdersAsync(user.getId()));这种情况应使用thenCompose()。第三不要把所有后续操作都写成Async。如果只是一次很快的内存计算.thenApply(value-value*2)通常没有必要改成.thenApplyAsync(value-value*2)Async会增加一次任务提交和调度不一定带来收益。第四不要依赖普通回调一定运行在某个固定线程中。前一阶段尚未完成时后续操作可能由完成前一阶段的线程执行前一阶段已经完成时当前调用回调方法的线程也可能直接执行。第五不要把anyOf()理解成“最快成功”。它等待的是最快完成最先异常完成也可能使结果异常完成。它也不会自动取消其他还在执行的任务。十、如何根据任务关系选择方法学习CompletableFuture时不应该先背方法名而应该先判断任务之间是什么关系。任务需求应使用的方法创建无返回值异步任务runAsync()创建有返回值异步任务supplyAsync()使用上一步结果并产生新结果thenApply()使用上一步结果但不产生新结果thenAccept()不使用上一步结果只等待它完成thenRun()下一步本身返回CompletableFuturethenCompose()两个独立任务完成后合并结果thenCombine()等待多个任务全部完成allOf()等待任意任务率先完成anyOf()将后续操作重新提交线程池对应的Async方法整章的主线可以概括为先创建异步任务再描述单个后续阶段如果后续阶段依赖前一个异步结果用thenCompose()连接如果两个任务互不依赖用thenCombine()、allOf()或anyOf()汇合如果需要隔离执行资源再用Async版本和自定义线程池建立新的调度边界。本章总结CompletableFuture的因果链条可以从普通Future的局限开始理解Future能保存一个异步任务的结果但调用线程必须主动调用get()等待结果并在结果到手后亲自推动下一步。任务之间的依赖越多调用线程越容易退化成异步流程的调度者。CompletableFuture把这种手动调度改成了阶段编排。runAsync()和supplyAsync()创建起点thenApply()、thenAccept()、thenRun()描述单个后续阶段thenCompose()连接前后依赖的异步任务thenCombine()汇合两个独立任务allOf()和anyOf()处理一批任务的完成关系Async方法和自定义线程池则决定后续阶段如何被调度。它的核心价值不是让依赖关系消失而是让依赖关系显式化。A 完成后执行 BA 的结果交给 BA 与 C 并行执行B 和 C 都完成后执行 D任意一个任务完成后继续——这些关系都可以提前写进任务链中由CompletableFuture在阶段完成时自动推动。至于任务执行失败后异常如何传播exceptionally()、handle()、whenComplete()有什么区别以及get()和join()应该如何选择将放到下一章单独讨论。