第 7 集：用了 CompletableFuture，两个查询为什么还是串行？

发布时间：2026/11/2 23:41:17
第 7 集：用了 CompletableFuture，两个查询为什么还是串行？

资深 Java 面试到 Agent 开发工程师 · 第 7 集

假设要生成一份报表，需要分别查询订单金额和退款金额，两个查询互不依赖。很容易写出这样的代码：

CompletableFuture<Integer>orders=CompletableFuture.supplyAsync(()->queryOrders(),ioPool);intorderAmount=orders.join();CompletableFuture<Integer>refunds=CompletableFuture.supplyAsync(()->queryRefunds(),ioPool);intrefundAmount=refunds.join();

supplyAsync把一个能返回结果的任务交给线程池，返回的CompletableFuture<Integer>用来取得这次任务的整数结果。join()会等待结果；如果任务已经完成，就直接取值。

这段代码刚提交订单查询，就停在join()等它结束。退款查询还没提交，当然无法与订单查询同时执行。换了异步 API，两个查询的先后关系却没变。

1. 两个查询一起提交，结果怎么合并？

把两个任务都提交出去，再写它们完成后要做的事：

importjava.util.concurrent.*;publicclassReportDemo{publicstaticvoidmain(String[]args){ExecutorServiceioPool=newThreadPoolExecutor(2,2,0,TimeUnit.SECONDS,newArrayBlockingQueue<>(10),newThreadPoolExecutor.AbortPolicy());try{CompletableFuture<Integer>orders=CompletableFuture.supplyAsync(()->100,ioPool);CompletableFuture<Integer>refunds=CompletableFuture.supplyAsync(()->20,ioPool);CompletableFuture<String>report=orders.thenCombine(refunds,(amount,returned)->"报表净额："+(amount-returned));System.out.println(report.join());}finally{ioPool.shutdown();}}}

这段完整程序可以在 Java 8 及以上运行，输出报表净额：80。两个数字代替查询返回值，不用它们衡量并行性能；真实调用时，再把两个 lambda 换成查询方法。线程数和队列容量只是演示配置。

thenCombine表示：订单和退款都正常完成后，把两个结果交给合并函数。amount是订单金额，returned是退款金额。查询还没完成时，就登记这个动作并返回，不用占着当前线程等查询。如果两个结果都已经就绪，合并函数也可能当场执行。

现在两个查询已经分别提交，线程池有可用线程时就能并行执行。最后的report.join()仍然会让主线程等待整份报表。并行查询缩短的是两个独立步骤原本串行的等待，不能让调用它的线程自动变成非阻塞。

如果 Web 框架支持直接返回异步结果，可以按框架约定返回这个 future；如果接口里仍调用join()，处理请求的线程就会等着。数据库查询本身若是阻塞调用，执行它的工作线程也一直被占用。

2. 下一步依赖上一步，为什么用 thenCompose？

报表净额算出来后，假设还要上传文件。用supplyAsync把上传交给线程池，完成后才能拿到下载地址：

CompletableFuture<String>url=report.thenCompose(text->CompletableFuture.supplyAsync(()->upload(text),ioPool));

这里的report沿用上一节，upload(text)假设返回上传后的地址。把这段接在report创建之后，最后等待url，再关闭线程池。上传需要报表内容，所以必须在报表完成后才能开始。

如果换成thenApply，返回类型会多套一层：

CompletableFuture<CompletableFuture<String>>nested=report.thenApply(text->CompletableFuture.supplyAsync(()->upload(text),ioPool));

thenApply把函数返回的东西，当成下一步的结果。函数返回的是 future，它就把这个 future 当作结果收下。因此，外层完成，只能说明已经取得了“上传任务的 future”，不能说明上传已经成功。

thenCompose会继续跟随上传任务的完成状态。上传成功，url才拿到地址；上传失败，url也会失败。调用方只需要处理一个 future。

选方法时可以直接看业务动作：把金额转成文字，用thenApply；拿到报表后启动另一个异步任务，并等它给出地址，用thenCompose；合并订单、退款这两个独立任务，用thenCombine。

连续写两次join()也能拿到嵌套结果，但等待逻辑又回到了调用方。已经知道任务之间的依赖，就把依赖写进链里。

3. thenApply 后面的代码，到底是谁执行？

方法名里没有Async，就不能指望它另找一个线程。下面这段可以直接运行：

CompletableFuture.completedFuture(100).thenApply(amount->{System.out.println(Thread.currentThread().getName());returnamount-20;}).join();

放在main方法里，打印的就是main。completedFuture(100)创建时已经有结果，调用thenApply的线程当场就能执行这个函数。

如果登记动作时，上游还没完成，函数则可能由完成上游的线程执行。例如查询线程拿到订单金额后，顺手完成下一步计算。非Async方法没有承诺固定在哪个线程执行，业务代码不能依赖某个线程名称。

这也是“异步查询已经结束，线程为什么还不空闲”的一个原因：后面接了耗时的文件生成，查询线程又接着生成文件。

如果文件生成主要消耗 CPU，可以明确交给计算线程池：

CompletableFuture<byte[]>file=report.thenApplyAsync(text->renderFile(text),cpuPool);

renderFile假设把报表文字生成文件字节。这里把计算任务交给cpuPool，便于分别限制计算和查询的并发量。具体需要几个线程，要结合上一集讲的资源与积压情况判断。Async表示按异步方式安排执行，并不保证新建线程；如果所选执行器直接调用任务，或线程池触发CallerRunsPolicy，任务仍可能在提交者线程里执行。

不传线程池的supplyAsync、thenApplyAsync等方法，普通CompletableFuture默认使用共享的ForkJoinPool.commonPool()；公共池的目标并行度不足 2 时，会改为每个任务创建线程。目标并行度是线程池期望同时执行任务的数量，不等于当前空闲线程数。

前一步传了ioPool，后面的thenApplyAsync(fn)也不会自动沿用它；要指定线程池，仍需写thenApplyAsync(fn, ioPool)。多个业务共用公共池时，阻塞查询可能占住其他任务也需要的执行资源。把慢查询放进受控的业务线程池，至少能分别配置容量、拒绝处理和监控；下游数据库的处理能力仍需要单独评估。

还有一种更难发现的等待。假设single是只有一个工作线程的线程池：

CompletableFuture<Integer>stuck=CompletableFuture.supplyAsync(()->{returnCompletableFuture.supplyAsync(()->80,single).join();},single);

外面的任务占着唯一的线程，在join()处等里面的任务。里面的任务却在同一个池里排队，等不到可用线程。如果没有超时或外部干预，两边就一直等下去。

改成表达依赖，让外面的函数结束，把线程还给池：

CompletableFuture<Integer>result=CompletableFuture.supplyAsync(()->100,single).thenCompose(amount->CompletableFuture.supplyAsync(()->amount-20,single));

thenCompose返回的 future 跟随内层任务完成，登记依赖时不占着工作线程等待，内层任务才能获得执行机会。

4. 链写出来以后，JDK 怎样接着往下执行？

CompletableFuture里保存任务结果，也保存等待这个结果的后续动作。上游没有完成时，后续动作先登记下来；上游完成后，JDK 再尝试执行它们。这里登记的是依赖动作，不是让每个动作各占一个线程等结果。

OpenJDK 21 中，thenApply把函数交给uniApplyStage，并传入空的执行器；显式指定执行器的thenApplyAsync则把执行器一并传下去。uniApplyStage的实现如下：

private<V>CompletableFuture<V>uniApplyStage(Executore,Function<?superT,?extendsV>f){if(f==null)thrownewNullPointerException();Objectr;if((r=result)!=null)returnuniApplyNow(r,e,f);CompletableFuture<V>d=newIncompleteFuture();unipush(newUniApply<T,V>(e,d,this,f));returnd;}

e是执行器，f是我们传入的函数，result保存上游结果。结果已经有了，就走uniApplyNow。上游正常完成时，有执行器就交给执行器，没有就直接调用函数；上游已经失败，则把失败传给下一步，不执行函数。已完成的 future 能在登记动作时直接执行函数，走的就是前一种情况。

上游没完成时，代码创建一个新的 futured，用来保存下一步的结果。UniApply把执行器、下游d、上游this和函数f关联起来；unipush把这个动作登记到上游保存后续动作的链表中，并处理登记期间上游刚好完成的情况。

上游完成后，postComplete()取出这些动作，调用动作的tryFire()，尝试按执行器规则执行函数。函数返回的值写入d，d后面的动作才有条件继续。假如函数抛异常，就把失败写入d。链里每一步都有自己的结果，后面的步骤跟着前面的完成状态推进。
任务结果的写入还用了前几集讲过的 CAS。源码里的关键语句是：

RESULT.compareAndSet(this,null,r)

RESULT用于原子访问保存结果的字段，this是当前 future，r是这次准备写入的结果。字段为null表示尚未完成；任务正常返回null时，JDK 会保存一个专门的标记，避免两种情况混淆。任务完成、超时和取消都可能尝试写入最终状态，先成功的那次决定结果。

这样，任务执行完和超时触发即使撞在一起，也只有一方能先把结果写进去。写入完成状态，不会自动停止正在执行的代码。

5. 查询失败了，为什么接口反而返回了正常报表？

假设退款查询失败，有人为了让链继续，给它加了默认值：

CompletableFuture<Integer>refunds=CompletableFuture.supplyAsync(()->queryRefunds(),ioPool).exceptionally(error->0);

exceptionally只在上游失败时执行。它返回0后，新产生的 future 就正常完成了，thenCombine会把退款当作零，继续生成报表。数据库故障因此变成了一个看起来正常的业务数字。

退款金额缺失时能不能按零计算，要由报表规则决定。如果必须准确，就让失败继续传给调用方，明确告诉用户报表生成失败。

记录错误可以用whenComplete：

CompletableFuture<String>observed=report.whenComplete((value,error)->{if(error!=null){logger.error("报表生成失败",error);}});

这里假设已有日志对象logger。whenComplete可以看到结果或异常。回调正常返回时，它不会把上游失败改成成功；调用方对observed.join()取值，仍会收到异常。回调自己抛异常，也可能让返回的 future 失败，所以不能把它当作绝不会影响结果的旁路代码。

handle则无论成功还是失败都会执行，并用函数返回值决定下一步结果。需要把成功与失败都转换成统一响应对象时，可以考虑它。使用这三种方法时，先想清楚自己是在观察错误，还是决定接受一个替代结果。

join()遇到普通任务异常，通常抛出CompletionException，原始原因在getCause()里；get()则通过受检的ExecutionException报告，还要求处理等待时的中断。不要只记录包装异常的名称，把实际失败原因丢掉。

6. 已经超时了，为什么数据库还在查？

只想限制调用方等多久，可以用：

report.get(200,TimeUnit.MILLISECONDS);

到时还没拿到结果，这次等待就抛TimeoutException。它不会把report自动改成失败，也不会停止订单和退款查询。200毫秒只是演示值，不能直接照搬成业务超时。

Java 9 起还有另一种写法：

report.orTimeout(200,TimeUnit.MILLISECONDS);

orTimeout会在规定时间内仍未完成时，尝试让这个 future 以TimeoutException失败。它返回的就是原来的report，没有额外创建一份互不影响的结果。如果其他代码也持有report，它们看到的也是同一个超时状态。Java 8 没有这个方法。

但即使超时先写入了失败状态，后台查询仍可能继续。CompletableFuture.cancel(true)也不能保证中断任务；这个类的取消机制不通过参数里的true控制执行线程的中断。

于是报表接口可能已经告诉用户超时，数据库连接却还被查询占用。用户不断重试，还可能再启动一批查询。需要同时检查数据库连接获取超时、查询超时、驱动对取消的支持，以及重复请求是否又创建了同一份报表。设置 future 超时，可以让调用方及时收到失败；要让查询停止，还得依靠数据库驱动和查询本身的超时、取消机制。

如果上传步骤已经产生外部效果，future 失败也不会自动删除已上传的文件。失败后是否清理、重新执行会不会生成重复文件，都要由业务处理。

常见面试题怎么答？

thenApply、thenCompose、thenCombine 有什么区别？

thenApply把上一步正常结果转换成下一个值；函数返回 future 时，会得到嵌套的 future。thenCompose用于下一步本身也是异步任务的情况，返回的 future 跟随这个任务完成，不额外套一层。thenCombine合并两个任务的正常结果，适合查询订单与退款这类互不依赖的步骤。它不会因为两个 future 被合并，就自动启动原本没有启动的任务。

CompletableFuture 的 Async 方法一定会新开线程吗？

不会。带Async的方法把动作安排给执行器，执行器可能复用线程，也可能直接在提交线程执行。普通CompletableFuture未指定执行器时，通常使用公共的ForkJoinPool，公共池并行度不足 2 时使用逐任务建线程的方式。没有Async的动作可能由完成上游的线程执行；上游已完成时，也可能在登记动作的线程直接执行。耗时动作放在哪里，需要明确执行器和拒绝策略，不能只看方法名称。

exceptionally、handle、whenComplete 怎么选？为什么兜底后接口不报错了？

exceptionally在失败时给替代结果；handle同时处理成功、失败，用返回值生成新结果；whenComplete适合观察完成情况，回调正常返回时保留原来的成功或失败。退款查询失败后恢复成零，会让后面的报表正常生成，是否允许这样做取决于业务含义。记录异常与恢复业务不能混成一件事，回调自身也可能抛异常。

读完自测

  1. 两个独立查询都使用supplyAsync，但订单提交后立刻join()，等它返回才提交退款。线程池有十个线程，两个查询会因此并行吗？
  2. 查询与后续文件生成都放在链上，文件生成用thenApply。为什么日志显示文件生成占用了查询线程？怎样调整？
  3. 同一个report被两个调用方持有，其中一个调用orTimeout后超时了，另一个还能从这个 future 等到正常结果吗？数据库查询会自动停止吗？

第 1 题参考：不会。退款还未提交，线程再多也无法执行它。先提交两个查询，用thenCombine合并，最后在需要结果的位置等待；还要保证执行资源允许并行。

第 2 题参考：非Async动作可能由完成查询的线程接着执行。文件生成主要耗 CPU 时，可用thenApplyAsync并显式传入计算线程池，同时检查线程池容量和拒绝策略。仅增加Async而不看执行器，仍不能保证避开提交线程执行。

第 3 题参考：如果这次超时已成功让report失败，两个调用方拿到的都是这个失败结果；后台查询后来返回也不能把它恢复成正常结果。查询是否结束取决于底层调用的超时、取消与中断支持。

参考资料：

  • Java 21：CompletableFuture API
  • Java 21：CompletionStage API
  • OpenJDK 21：CompletableFuture 源码