CompletableFuture与Phaser实战：异步编排与分阶段任务协调

写了多年 Java 服务端最近两年我在项目里越来越频繁地用上 CompletableFuture 和 Phaser 这两个 JUC 工具类。尤其是 CompletableFuture几乎成了异步编程的标准答案但真正会用的人不多。大多数人把它当 Future 的增强版调个 get() 拿结果就完事Phaser 就更冷门了很多两年经验的 Java 工程师压根没在业务代码里见过它。这次我用一个 Spring Boot 实战项目把它们完整串联起来讲清楚两个工具的核心机制、适用场景和实战编码方式Java 面试里碰到高并发工具类相关的问题也能有更多谈资。这套内容适合正在准备 JUC 面试题的人也适合后端开发想优化接口耗时、想搞懂异步编排和分阶段任务协调的人。即使你只是想把代码从能跑提升到扛得住这几个工具都值得花一个下午认真过一遍。1. 为什么把 CompletableFuture 和 Phaser 放在一起讲1.1 两者的核心定位差异很多人对 JUC 工具类的理解停留在容器 锁 线程池这老三样其实 JUC 里还有一批专门处理多线程协作的高级工具。CompletableFuture 和 Phaser 虽然都属于并发协调类但解决的问题完全不同。CompletableFuture 解决的是异步结果编排问题。比如一个接口需要同时调用用户服务、订单服务、商品服务三个结果都拿到之后再组装返回。传统写法是串行调三次或者用线程池加 CountDownLatch 手动等而 CompletableFuture 可以天然地表达并行执行、全部完成后再处理这类语义。Phaser 解决的是阶段化任务同步问题。它的核心能力是让一组线程在多个阶段上保持步调一致比如一个数据处理流水线有读取 - 转换 - 写入三个阶段每个阶段所有线程都完成之后才统一进入下一阶段。Phaser 相比老牌的 CyclicBarrier 和 CountDownLatch最大的优势是参与者数量可以动态变化任务进行中可以随时注册新的线程或者注销已有线程。我把这两个放在一起讲是因为真实的高并发系统里它们经常搭配出现CompletableFuture 负责把一组异步任务编排起来Phaser 负责协调多个参与者在阶段边界上对齐。一个管怎么组合一个管怎么同步。1.2 我对这套组合的选型思路在做这个 Spring Boot 实战项目时我先给自己定了几条选型原则。第一能用 JDK 原生工具就不用第三方框架。数据聚合、并行调用这类需求CompletableFuture 足够搞定没必要引入 RxJava 或者 Project Reactor 增加团队学习成本。Spring 自带的 Async 注解配合自定义线程池虽然也能做异步但在多个异步任务之间编排关系这种场景下CompletableFuture 的 thenCombine、allOf 这些 API 明显更灵活。第二阶段协调场景优先考虑 Phaser 而非 CountDownLatch 或 CyclicBarrier。CountDownLatch 的计数器不能重置适合一次性的闸门CyclicBarrier 虽然可循环使用但参与者数量是构造时固定的不适合动态增减。Phaser 把这两者的优点都收进来了而且 API 设计更统一onAdvance 回调还可以在阶段切换时执行额外逻辑。第三所有异步任务必须走自定义线程池。CompletableFuture 默认使用 ForkJoinPool.commonPool()这是个全局共享线程池线程数等于 CPU 核数减一。在 Spring Boot 应用里如果所有异步任务都挤在这个公共池里很容易出现互相阻塞尤其是遇到 IO 密集型的远程调用时线程不够用直接导致任务排队。2. 环境准备与工程搭建2.1 依赖引入与 Java 版本选择这个实战项目是基于 Spring Boot 3.2 Java 21 搭建的。如果你还在用 Java 8CompletableFuture 的基本能力也能用但 Java 9 以后新增的 orTimeout、completeOnTimeout 这些带超时语义的方法就享受不到了。这两个方法在真实业务里特别有用所以我强烈建议至少用 Java 11 以上的版本。工程本身只需要一个基础依赖JUC 工具类是 JDK 自带的不需要额外引入包。我的 pom.xml 里只有常规的 spring-boot-starter-web用于模拟真实的 HTTP 接口场景。dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency如果你的项目要从 Spring Boot 2.x 升级到 3.x注意 Jakarta EE 的包名变化javax.* 要改成 jakarta.*其他倒不影响 JUC 工具类的使用。2.2 自定义线程池CompletableFuture 的第一道坎在 Spring Boot 里用 CompletableFuture第一步就是配一个合适的线程池。我之前见过太多生产事故都是因为没配线程池、任务全堆在 ForkJoinPool.commonPool() 里导致接口集体超时。我这边用的是 ThreadPoolTaskExecutor这是 Spring 封装的线程池相比原生 ThreadPoolExecutor 多了一些便利性比如线程名前缀、优雅关闭等特性。核心配置如下Configuration public class AsyncConfig { Bean(businessExecutor) public ThreadPoolTaskExecutor businessExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(8); executor.setMaxPoolSize(16); executor.setQueueCapacity(100); executor.setThreadNamePrefix(biz-exec-); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setWaitForTasksToCompleteOnShutdown(true); executor.setAwaitTerminationSeconds(30); executor.initialize(); return executor; } }几个参数我解释一下为什么这么定。核心线程数先设为 8因为这台开发机的 CPU 是 8 核对于 CPU 密集和 IO 密集混合的业务场景8 个核心线程是一个比较稳妥的起点。最大线程数 16 是为了应对短暂的流量尖峰。队列容量 100 是缓冲余量。拒绝策略我选了 CallerRunsPolicy这意味着线程池满了之后新任务不会被丢弃而是由提交任务的线程自己执行代价是接口响应变慢但至少不会丢数据。这个策略在大多数业务场景里比 AbortPolicy 可靠得多。还有一个细节很多人容易忽略线程池参数不能照搬模板要根据任务的类型来定。如果你的任务以 IO 操作为主调用远程接口、读写数据库核心线程数可以适当调大因为线程大部分时间在等待 IOCPU 并没有被占满。如果是纯 CPU 计算任务线程数超过 CPU 核数反而会导致上下文切换开销增大。3. CompletableFuture 实战异步编排的正确姿势3.1 核心 API 一句话记忆法CompletableFuture 的 API 数量很多新手容易看花眼。我总结了一套记忆方法看方法名里的两个部分前半部分决定触发时机后半部分决定处理方式。触发时机有以下几类runAsync / supplyAsync开始异步执行一个任务前者没有返回值后者有返回值thenApply / thenAccept / thenRun上一个任务完成后执行下一个分别对应有入参有返回值、有入参无返回值、无入参无返回值thenCombine / thenAcceptBoth两个任务并行完成后合并结果thenCompose上一个任务的结果作为下一个任务的入参用于扁平化嵌套的 CompletableFuture避免 CompletableFutureCompletableFuture 这种地狱结构allOf / anyOf等待多个任务全部完成 / 任一完成处理方式包含两类一类是同步处理比如 thenApply执行回调的线程和前面的任务线程是同一个另一类是异步处理比如 thenApplyAsync回调任务会被提交到线程池重新调度。默认不指定线程池的情况下thenApply 和 thenApplyAsync 的行为差异很大前者在结果产生的线程上继续执行后者走 ForkJoinPool。在实际项目中我主要用的是 thenApply、thenCombine、allOf 这三个。exceptionally 和 handle 用于异常处理orTimeout 用于超时兜底。3.2 业务实战并行查询商品信息并聚合我把这个场景模拟成一个典型的商详页接口前端需要展示商品基本信息、实时库存、近期销量、用户评价摘要四个模块的数据。这四个数据来自不同的服务完全互不依赖串行调用的话耗时是四者之和并行调用耗时只取决于最慢的那个服务。核心代码这样写Service public class ProductAggregationService { private final ProductClient productClient; private final StockClient stockClient; private final SalesClient salesClient; private final ReviewClient reviewClient; private final ThreadPoolTaskExecutor businessExecutor; public ProductAggregateVO getProductDetail(Long productId) { CompletableFutureProductInfo productFuture CompletableFuture.supplyAsync(() - productClient.getInfo(productId), businessExecutor); CompletableFutureStockInfo stockFuture CompletableFuture.supplyAsync(() - stockClient.getStock(productId), businessExecutor); CompletableFutureSalesInfo salesFuture CompletableFuture.supplyAsync(() - salesClient.getSales(productId), businessExecutor); CompletableFutureReviewSummary reviewFuture CompletableFuture.supplyAsync(() - reviewClient.getReviewSummary(productId), businessExecutor); CompletableFutureVoid allDone CompletableFuture.allOf( productFuture, stockFuture, salesFuture, reviewFuture ); allDone.join(); return ProductAggregateVO.builder() .product(productFuture.join()) .stock(stockFuture.join()) .sales(salesFuture.join()) .review(reviewFuture.join()) .build(); } }这里有几个值得注意的点。第一个是每个 supplyAsync 都显式传入了 businessExecutor。如果你不传默认走 ForkJoinPool.commonPool()。在生产环境里这是一个非常隐蔽的坑。第二个是先 allOf().join()再逐个 join()。allOf 表达的是等所有任务都完成这个语义join 才是真正阻塞拿结果。如果你不等 allOf 直接调 join()其实也能工作因为 join 本身就会阻塞等待对应 future 完成但多个 join 串行等待的语义不够清晰也不利于后续扩展。第三个是join() 和 get() 的取舍。get() 抛出受检异常调用方必须 try-catchjoin() 不抛受检异常而是包装成 CompletionException 运行时异常抛出。在业务代码里我更喜欢 join()因为异常会往上层抛由统一的全局异常处理器兜底代码更干净。但要小心join() 在任务异常时抛出的异常会被包装一层排查问题时需要通过 getCause() 才能看到真实异常。3.3 依赖关系的表达thenApply 与 thenCompose并行聚合只是 CompletableFuture 的基础用法真正体现编排能力的是处理有依赖关系的异步任务。举个例子下单场景中需要先创建订单号再用订单号调用支付服务获取支付参数。这两个步骤有先后依赖但我不想在阻塞等待创建订单号之后再做下一步而是把整个链路串成一个完整的异步管道。public CompletableFuturePayParamVO createOrderAndGetPayParam(OrderCreateDTO dto) { return CompletableFuture .supplyAsync(() - orderService.createOrder(dto), businessExecutor) .thenApplyAsync(order - payService.buildPayParam(order.getOrderNo()), businessExecutor); }thenApply 的作用是上游 CompletableFuture 的结果作为入参传给下一个函数返回一个新的 CompletableFuture。如果我在 thenApply 里返回的是一个 CompletableFuture就会出现 CompletableFutureCompletableFuture 的嵌套结构这时候就要用 thenCompose 把它扁平化。public CompletableFuturePayResultVO payWithCompose(OrderCreateDTO dto) { return CompletableFuture .supplyAsync(() - orderService.createOrder(dto), businessExecutor) .thenCompose(order - CompletableFuture.supplyAsync(() - payService.pay(order.getOrderNo()), businessExecutor)); }还有一个高频场景是 thenCombine用于同时等待两个独立任务并合并结果。比如计算订单实付金额需要同时拿到商品价格和优惠信息两个结果都到了才能计算。public CompletableFutureBigDecimal computeFinalPrice(Long productId, String couponCode) { CompletableFutureBigDecimal priceFuture CompletableFuture.supplyAsync(() - productClient.getPrice(productId), businessExecutor); CompletableFutureBigDecimal discountFuture CompletableFuture.supplyAsync(() - couponClient.getDiscount(couponCode), businessExecutor); return priceFuture.thenCombine(discountFuture, (price, discount) - price.subtract(discount).max(BigDecimal.ZERO)); }3.4 超时与异常处理不要裸奔生产环境里异步任务最容易出的问题就是远端服务慢导致线程池被占满。如果 CompletableFuture 没有设置超时调用方会一直阻塞线程池资源被一个个慢任务耗尽最终全站崩溃。Java 9 以后提供了 orTimeout 和 completeOnTimeout 两个方法非常好用。CompletableFutureStockInfo stockFuture CompletableFuture .supplyAsync(() - stockClient.getStock(productId), businessExecutor) .orTimeout(2, TimeUnit.SECONDS) .exceptionally(ex - StockInfo.defaultStock());orTimeout 的含义是如果 2 秒内任务没有完成就抛出一个 TimeoutException 让这个 future 异常结束。配合 exceptionally可以在超时或出现异常时返回一个默认值保证主流程不中断。这里我要强调一个理解误区orTimeout 并不会取消正在执行的线程。底层线程该跑还是继续跑只是调用方不会再无限等下去。所以在设计降级策略时要考虑一个事实——超时的任务可能还在消耗线程池资源。解决思路是给慢任务本身也设置超时时间比如 OkHttp 或 OpenFeign 都支持连接超时和读取超时配置双管齐下才能真正保护线程池。异常处理方面我的建议是只在调用链的边界处理异常不要在每一步都用 exceptionally。每一步都包装异常会导致异常被提前吞掉真正的错误原因被层层掩盖排查问题时非常痛苦。正确姿势是业务步骤里直接抛出异常在最外层用 handle 或 exceptionally 统一兜底返回降级结果或者记录失败日志。4. Phaser 实战分阶段任务的动态调度4.1 Phaser 核心机制与和 CyclicBarrier 的对比先直接回答一个面试高频问题Phaser 和 CyclicBarrier 有什么区别CyclicBarrier 是让一组线程互相等待到达屏障点后同时放行屏障可以循环使用。但它的参与线程数量在构造时固定无法在运行中增减。CountDownLatch 是倒数门闩计数归零后放行且不能重置。Phaser 像一个加强版的 CyclicBarrier允许在任务执行过程中动态注册和注销参与方。它的核心概念有三个phase阶段号Phaser 的每轮同步都是一个阶段从 0 开始递增parties参与方数量当前注册的调用者数量可以动态变化arrive到达一个调用者到达同步点但不等待其他调用者awaitAdvance等待前进等待指定阶段完成Arrive 和 await 是分离的这就让 Phaser 的用法比 CyclicBarrier 灵活很多。比如一个线程可以先 arrive 再去做其他事情回头再 awaitAdvance 等待整组线程前进到下一阶段这种先签到后忙活的模式在真实业务里很实用。Phaser 还有一个非常有用的回调机制每次阶段的最后一个参与者 arrive 后会自动触发 onAdvance 方法可以在这个方法里做阶段切换的准备工作。onAdvance 返回 true 时Phaser 进入终止状态。4.2 实战多阶段数据处理流水线我设计了一个实证场景来展示 Phaser 的完整用法模拟一批数据需要经过校验 - 清洗 - 转换 - 落库四个阶段每个阶段有多个 worker 并行处理数据切片所有 worker 都完成当前阶段后一起进入下一阶段。public class DataPipelineSimulator { private static final int WORKER_COUNT 4; private static final int PHASE_COUNT 4; public void run() { Phaser phaser new Phaser(WORKER_COUNT 1); for (int i 0; i WORKER_COUNT; i) { final int workerId i; Thread thread new Thread(() - workerLoop(workerId, phaser), worker- i); thread.start(); } while (phaser.getPhase() PHASE_COUNT) { phaser.arriveAndAwaitAdvance(); System.out.println(All workers finished phase phaser.getPhase()); } System.out.println(Pipeline complete, phaser terminated: phaser.isTerminated()); } private void workerLoop(int workerId, Phaser phaser) { int phase 0; while (phase PHASE_COUNT) { processPhase(workerId, phase); phaser.arriveAndAwaitAdvance(); phase phaser.getPhase(); } } private void processPhase(int workerId, int phase) { System.out.println(Worker workerId processing phase phase); try { Thread.sleep((long) (Math.random() * 500)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }在这个例子中主线程也是参与者之一所以构造 Phaser 时传入了 WORKER_COUNT 1。主线程和 worker 线程每完成一个阶段就 arrive 一次最后一个到达的线程会让所有线程一起进入下一阶段。这段代码展示了 Phaser 最典型的使用模式但还不足以体现它相对 CyclicBarrier 的核心优势。Phaser 真正的杀手锏是动态增减参与者。设想这样的场景数据处理任务开始时有 4 个 worker 参与处理到第二阶段时发现数据量变大需要再加 2 个 worker 进来。CyclicBarrier 根本做不到而 Phaser 可以用 register() 或 bulkRegister() 动态注册新参与者。public void addWorkerDynamic(Phaser phaser, String name) { phaser.register(); Thread thread new Thread(() - { while (!phaser.isTerminated()) { doWork(name); phaser.arriveAndAwaitAdvance(); } }, name); thread.start(); }新注册的线程会被纳入当前阶段的参与方计数中这是在运行期实时生效的。任务中途有 worker 挂了也可以调用 arriveAndDeregister() 主动退出表示我不再参与后续阶段了其他线程不会被卡死。这个能力在分布式任务调度、数据处理流水线等场景里非常实用。比如实现一个简单的多阶段任务调度器第一阶段做参数校验第二阶段做数据预加载第三阶段做实际计算每个阶段的参与方可能都不一样用 Phaser 天然适配。面试时能讲清楚这个动态注册机制和实际场景是很加分的。4.3 一次面试题的思路延展用 Phaser 模拟旅游团行程讲一个面试官很爱问的题目变种多个游客组成旅游团每个景点参观结束后需要等所有人到齐才能去下一个景点而游客们可以随时加入或离团。这个需求用 CyclicBarrier 写会非常别扭因为 CyclicBarrier 的参与方数量是固定的。而 Phaser 的思路非常自然每次到达一个景点所有游客调用 arriveAndAwaitAdvance()新游客加入时调用 register()有游客离团时调用 arriveAndDeregister()Phaser phaser new Phaser(1); for (int i 0; i 3; i) { new Thread(() - { phaser.register(); visit(Scenic A); phaser.arriveAndAwaitAdvance(); visit(Scenic B); phaser.arriveAndAwaitAdvance(); phaser.arriveAndDeregister(); }).start(); } Thread.sleep(1000); new Thread(() - { phaser.register(); visit(Scenic B); phaser.arriveAndAwaitAdvance(); phaser.arriveAndDeregister(); }).start(); phaser.arriveAndDeregister();最后这行主线程的 arriveAndDeregister 很关键它让主线程退出参与方列表表示主线程不再等待后续阶段否则 Phaser 会一直认为还有参与者没到齐导致流程卡住。5. Spring Boot 集成要点从能跑到跑得稳5.1 优雅停机与任务关闭Spring Boot 应用在重启时如果线程池里还有正在执行的异步任务直接关闭进程会导致任务中断。所以线程池必须配置优雅关闭。ThreadPoolTaskExecutor 提供了两个关键配置setWaitForTasksToCompleteOnShutdown(true) 和 setAwaitTerminationSeconds(30)。前者表示关闭线程池时要等待已提交的任务执行完毕后者是最长等待时间避免任务长时间卡住拖慢应用停机。CompletableFuture 唯一的问题是无法强制中断正在运行的任务所以优雅停机只能覆盖已提交未执行的任务以及等待中的线程。如果某个任务本身设计有缺陷比如永远不返回那再有优雅停机配置也没用。因此所有异步任务的内部逻辑最好都加上可中断的设计至少要在等待和阻塞调用处响应中断信号。5.2 线程池吞吐量的观测与调参思路上线之后不能只看接口通不通还要观测线程池的运行指标。ThreadPoolTaskExecutor 的 getThreadPoolExecutor() 可以拿到原生 ThreadPoolExecutor从而读取活跃线程数、队列大小、完成任务数这些指标。我习惯写一个简单的监控接口来输出这些信息RestController public class ExecutorMetricsController { private final ThreadPoolTaskExecutor businessExecutor; GetMapping(/metrics/executor) public MapString, Object executorMetrics() { ThreadPoolExecutor pool businessExecutor.getThreadPoolExecutor(); MapString, Object metrics new HashMap(); metrics.put(corePoolSize, pool.getCorePoolSize()); metrics.put(maxPoolSize, pool.getMaximumPoolSize()); metrics.put(activeCount, pool.getActiveCount()); metrics.put(poolSize, pool.getPoolSize()); metrics.put(queueSize, pool.getQueue().size()); metrics.put(completedTaskCount, pool.getCompletedTaskCount()); return metrics; } }排查问题的时候这个接口特别有用。如果 poolSize 长期等于 maxPoolSize说明线程不够用如果 queueSize 长期堆积说明消费速度跟不上生产速度如果 activeCount 很高但 completedTaskCount 增长缓慢优先怀疑任务中有大量的阻塞操作比如同步调用外部服务超时。调参方面我的经验是先从核心线程数等于 CPU 核数开始跑观察一段时间如果队列经常积压就逐步增加核心线程数。IO 密集型任务可以把核心线程数调到 CPU 核数的 2 到 3 倍但不能盲目上调过高的线程数在业务高峰期可能引发 GC 压力和大量的上下文切换。6. 常见问题与排查技巧实录6.1 高频事故清单我在实际项目里踩过不少坑整理成一张速查表供参考。现象根因解决方法接口偶尔超时重启后恢复线程池被慢任务占满给异步任务设置超时合理配置线程池拒绝策略使用 CompletableFuture 后接口响应反而变慢任务体量太小线程切换开销大于并行收益只在任务耗时明显的场景使用异步短任务直接串行异常日志里看不到真实错误异常被 exceptionally 或 handle 提前吞掉在 task 内部记录错误日志或者在边界统一处理allOf().join() 抛 CompletionException某个子任务抛异常使用 getCause() 查看真实异常结合 exceptionally 做降级Spring 的 Async 注解不生效同类内部调用代理失效把方法拆分到不同 Bean或者直接使用 CompletableFuturePhaser 所有线程卡在 awaitAdvance参与者数量与实际不符检查 register 和 deregister 的调用次数是否匹配这里的 Spring Async 自调用失效问题值得多说一句。Spring 的异步代理基于 AOP只有外部调用才会走代理同类内部方法之间的调用不会经过代理Async 自然就失效了。所以在同一个类里调用加了 Async 的方法方法会同步执行接口耗时毫无变化排查起来非常隐蔽。如果是 Spring Boot 3.x 还可以用 Async 加 Lazy 的方式做一点变通但最推荐的做法还是直接把异步方法放到独立的 Service 里。6.2 排查工具与最终心得排查并发问题我常用的思路是先看线程栈再看线程池指标。生产环境可以通过 jstack 抓线程快照看看业务线程处于什么状态。如果大量线程卡在 WAITING 状态说明它们在等待某个条件如果大量线程处于 RUNNABLE 状态但 CPU 占用不高大概率是忙等或者频繁的锁竞争。线程池指标可以从上面的监控接口拿到。两个数据一结合基本能定位到大部分问题。最后说说我对 CompletableFuture 和 Phaser 的定位。CompletableFuture 不是银弹它解决的是任务编排问题前提是你对线程池、超时、降级策略有清晰的规划。如果项目里根本没有并发场景强行引入异步只会增加复杂度。Phaser 更是如此普通业务系统里可能一年都用不上一次但它解决的那类阶段化动态协作问题一旦遇到你很难找到更优雅的替代方案。我个人的建议是平时写业务代码时多用 CompletableFuture 的 thenCombine、allOf 和 exceptionally 这几个 API把它们练到条件反射的程度Phaser 则重点理解它的阶段模型和动态参与者机制面试和架构设计时都是不错的亮点。多线程编程的核心从来不是把 API 背得多熟而是真正理解任务的依赖关系和资源边界。