我经常在别人的代码里看到这种写法一个接口方法把所有耗时操作扔进 CompletableFuture然后直接.get()等结果最后调用超时设置时才发现get(5, TimeUnit.SECONDS)这种同步超时倒是管用可一旦换成thenApply、thenAccept这类异步回调链超时参数的“手感”就完全不一样了。这个“CompletableFutureTimeout”的痛点说白了就是异步编程里超时不是加一个参数就完事而是一整套关于任务取消、线程池隔离、回调链传播的设计问题。这篇博文就围绕我实际踩过的坑聊聊 CompletableFuture 里解决超时的一套可落地方案。包括最容易被忽略的orTimeout和completeOnTimeout的底层逻辑、线程池为什么必须独立配置、回调链上的超时为什么会“失效”以及最后我自己沉淀下来的一套超时骨架代码。适合正在用 CompletableFuture 做服务端开发尤其是接口聚合、并行调用外部服务这类场景的同学参考。1. 整体思路为什么 CompletableFuture 的超时比其他并发工具更麻烦1.1 核心问题FutureTask 时代留下的思维惯性很多同学是从ExecutorService.submit()返回的Future开始接触异步编程的对超时的第一反应就是future.get(timeout, TimeUnit)。这套思维迁移到 CompletableFuture 上第一个坑就出现了CompletableFuture不是FutureTask它内部没有“等待任务执行”的阻塞语义而是一张状态流转的有向图。我用一个例子说明这个区别。ExecutorService pool Executors.newFixedThreadPool(4); FutureString future pool.submit(() - { Thread.sleep(5000); return done; }); try { String result future.get(2, TimeUnit.SECONDS); } catch (TimeoutException e) { // 这个超时能让主线程不再等待 // 但线程池里的任务其实还在跑 }future.get(2, SECONDS)做的是“主线程只等 2 秒”它并没有“让任务 2 秒后停止”的能力任务还在线程池里继续消耗资源。到了 CompletableFuture 这里问题更复杂因为你可以把多个任务串成链thenApply、thenAccept、thenCompose这些方法会生成新的 CompletableFuture如果你对链的中间节点做超时下游节点感知不到状态已经流转到 new CompletableFuture 上了。所以处理 CompletableFuture 超时的第一步是转变思路不是给“等待”加时间上限而是给“任务流转图”加一个兜底状态。1.2 两条路线JDK 原生 API 与手动安排超时线程JDK 9 开始CompletableFuture 增加了两个非常有用的方法orTimeout(long timeout, TimeUnit unit)如果任务在指定时间内没完成强制让这个 CompletableFuture 以TimeoutException异常结束。completeOnTimeout(T value, long timeout, TimeUnit unit)如果任务超时用给定的默认值让 CompletableFuture 正常完成。但使用这两个方法有一个前置条件执行任务的线程池里的线程必须对中断有响应。否则会出现一种“表面上 CompletableFuture 已经超时完成了但底层线程还在拼命算”的尴尬局面。很多人不理解为什么这两个方法在 JDK 9 才出现其实是因为它们底层依赖TimeoutController这样一个内部机制。简单来说orTimeout会在内部启动一个Delayer一个基于ScheduledThreadPoolExecutor的单线程延迟调度器到了时间直接调用completeExceptionally或者complete把 CompletableFuture 的状态强行变成终态。所以我的整体设计思路分成两条线使用orTimeout/completeOnTimeout作为“状态兜底”手动确保线程池里正在执行的耗时任务能被取消或及时感知超时。后面所有方案都是围绕这两条线展开的。2. orTimeout 与 completeOnTimeout 的底层行为和正确用法2.1 orTimeout异常结束是一种信号orTimeout设计的核心语义是“告诉调用方这个任务没戏了”它会将 CompletableFuture 的结果替换为CompletionException具体抛出的异常是TimeoutException包裹在CompletionException里。实操中我见过一种错误用法在链式调用的中间环节加orTimeout以为能阻断整条链。CompletableFutureString future CompletableFuture.supplyAsync(() - { sleep(5); return A; }, pool).thenApply(a - { sleep(5); return a B; }).orTimeout(2, TimeUnit.SECONDS); // 你以为上面这行 orTimeout 会让整个链路 2 秒后全部结束 // 实际上它只是让最后一个 CompletableFuture 超时中间的任务 A 和任务 B 还在跑这涉及 CompletableFuture 的状态传递机制thenApply返回的是一个新 CompletableFutureorTimeout又是基于这个新对象生成另一个 CompletableFuture。每一个操作符都是一次包装orTimeout只对调用它的那个 CompletableFuture 生效。正确姿势是把orTimeout放在链路的末端而且最好配合whenComplete或handle来统一处理超时结果CompletableFutureString future CompletableFuture .supplyAsync(() - queryRemote(), pool) .thenApply(r - parse(r)) .orTimeout(3, TimeUnit.SECONDS) .handle((result, ex) - { if (ex ! null) { if (ex instanceof CompletionException ex.getCause() instanceof TimeoutException) { return timeout_default; } return error_default; } return result; });注意handle里拿到的异常类型orTimeout触发的异常会被包装成CompletionException其getCause()是TimeoutException。而且当你直接在外面 catch 的时候捕获到的也是ExecutionException这又是一层包装。我建议所有超时判断都使用getCause()链判断不要直接用instanceof TimeoutException判断最外层。2.2 completeOnTimeout默认值不是免费的completeOnTimeout很适合降级场景比如“查询用户详情失败就给默认 VIP 等级”。但有一个细节很多人没注意completeOnTimeout的默认值参数是在调用那一刻就计算好的。CompletableFutureString future CompletableFuture .supplyAsync(() - query(), pool) .completeOnTimeout(loadDefaultConfig(), 2, TimeUnit.SECONDS);你以为loadDefaultConfig()只在超时后执行实际上它在你调用completeOnTimeout的瞬间就执行了。如果你的默认值是从数据库或者远程配置中心加载的这就把一个异步任务又变成了同步阻塞。正确写法是把默认值也异步化或者至少延迟计算CompletableFutureString defaultFuture CompletableFuture .supplyAsync(() - loadDefaultConfig(), pool); CompletableFutureString future CompletableFuture .supplyAsync(() - query(), pool) .completeOnTimeout(null, 2, TimeUnit.SECONDS) .thenCombine(defaultFuture, (r, d) - r ! null ? r : d);或者更简单一些用orTimeout加handle的方式这样默认值只在真正超时时才去算CompletableFutureString future CompletableFuture .supplyAsync(() - query(), pool) .orTimeout(2, TimeUnit.SECONDS) .handle((r, ex) - ex ! null ? loadDefaultConfig() : r);注意这里handle里如果loadDefaultConfig()又是一个耗时操作它还是在 CompletableFuture 的线程里执行会占用池子里的线程。更稳妥的做法是handle里返回CompletableFuture再用thenCompose展平。2.3 时间基准千万别忽略调度延迟orTimeout和completeOnTimeout的内部实现依赖一个公共的ScheduledThreadPoolExecutor它是一个守护线程在类加载时被初始化。所有 CompletableFuture 实例共享这一个延迟调度器。这就带来几个实际问题如果系统里有成千上万个 CompletableFuture 同时设置超时它们的调度任务会排队进入同一个延迟队列时间精度会受负载影响。orTimeout是从“调用这个方法的那一刻”开始计时而不是从任务开始执行时计时。如果你的任务在队列里排了很久等它真正开始跑时超时时间已经消耗了一部分。时间精度是毫秒级的要求做到微秒级精准超时的场景不适合用这套机制。我曾经遇到过一个问题一个核心接口的 P99 耗时从一开始的 80ms 增长到 1.5s排查下来发现是全局的 CompletableFuture 超时任务太多公共调度器成了瓶颈。后来我改用每个业务独立的ScheduledExecutorService来设置超时或者直接在线程池包装层做超时控制问题才缓解。3. 线程池隔离与超时的关系3.1 为什么说线程池配置决定超时是否有效orTimeout能改变 CompletableFuture 的状态但不一定能“停下来”正在执行的任务。看下面这个例子ExecutorService pool Executors.newFixedThreadPool(2); CompletableFuture.supplyAsync(() - { while (true) { // 模拟死循环式的高CPU任务 } }, pool).orTimeout(1, TimeUnit.SECONDS);运行这段代码 10 秒后线程池的两个工作线程仍然处于 RUNNABLE 状态CPU 占用率拉满。orTimeout只是把 CompletableFuture 标记成异常完成了但线程池里的任务完全不受影响。这就是为什么我需要强调“线程池与超时是两件事”。CompletableFuture做事后的结果通知ThreadPoolExecutor才做事中的资源管控。两者必须配合才算是完整的超时方案。具体配合方式使用支持中断的任务或者通过协作式取消标记来让任务感知到超时信号。3.2 线程池选择的标准我一般建议 CompletableFuture 的线程池遵循几个原则永远不要直接使用ForkJoinPool.commonPool()跑业务任务。每个业务模块或者接口组维护一个独立的 ThreadPoolExecutor线程数和队列长度有明确上限。线程池拒绝策略不能是AbortPolicy因为 CompletableFuture 的supplyAsync遇到拒绝时会把RejectedExecutionException包进结果里但这个异常有时候不会立刻触发导致调用方感知到的延迟远比预期大。线程池中的线程要允许设置自定义的UncaughtExceptionHandler否则业务异常可能在某个角落被吞掉。我常用的一个配置模板ThreadPoolExecutor ioPool new ThreadPoolExecutor( 8, // 核心线程 16, // 最大线程 30, TimeUnit.SECONDS, new LinkedBlockingQueue(1000), new ThreadFactory() { private final AtomicInteger idx new AtomicInteger(); Override public Thread newThread(Runnable r) { Thread t new Thread(r, biz-io- idx.getAndIncrement()); t.setDaemon(false); t.setUncaughtExceptionHandler((thread, throwable) - log.error(task error, throwable)); return t; } }, new CallerRunsPolicy() );选CallerRunsPolicy而不是AbortPolicy是因为在超时场景下任务被拒绝往往意味着系统已经过载此时让提交任务的线程自己去执行至少能保证任务不丢。但要注意CallerRunsPolicy下如果提交线程也是某个线程池的工作线程可能会出现层层回推所以实际上我经常用自定义拒绝策略丢到另一个降级线程池里处理。3.3 带超时的线程池包装如果不想依赖orTimeout这套状态机制也可以在线程池层做超时。做法是提交一个 Callable然后用Future.get(timeout)控制等待时间。这和 CompletableFuture 的优点互补Future.get的超时能保证主线程不卡死但无法中断任务除非任务本身响应中断。因此我在实际项目中通常做两层设计外层CompletableFuture 的orTimeout保证调用方能快速拿到结果内层任务内部通过循环检查Thread.currentThread().isInterrupted()或一个共享的AtomicBoolean取消标记保证线程能被释放。下面是一个比较完整的配合写法AtomicBoolean cancelled new AtomicBoolean(false); CompletableFutureString future CompletableFuture.supplyAsync(() - { try { while (!cancelled.get()) { String partial doChunkWork(); if (partial ! null) { return partial; } } throw new CancellationException(task cancelled); } finally { // 清理资源 } }, ioPool).orTimeout(2, TimeUnit.SECONDS) .whenComplete((r, ex) - cancelled.set(true));这里的思路是whenComplete在超时触发时也会执行把cancelled置为 true任务内部在下一个循环里感知到并退出。虽然不是强杀线程但比完全放任不管好得多。4. 完整方案可落地的 CompletableFuture 超时工具类4.1 设计目标与 API我在项目里沉淀了一个FutureTimeout工具类核心目标是给任意 CompletableFuture 链加上超时兜底同时尽量不污染业务代码。API 只有三个静态方法public static T CompletableFutureT within( CompletableFutureT future, Duration timeout) public static T CompletableFutureT withinOrDefault( CompletableFutureT future, Duration timeout, T defaultValue) public static T CompletableFutureT withinOrThrow( CompletableFutureT future, Duration timeout, Supplier? extends Throwable exSupplier)设计上强调一点输入参数是“已经构建好的 CompletableFuture”因为业务代码里经常出现异步链最后一步可能是一个thenCompose返回的 CompletableFuture。这个方法要能把它包起来。4.2 实现细节统一使用orTimeout实现因为它的语义最干净。public static T CompletableFutureT within( CompletableFutureT future, Duration timeout) { return future.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS); }withinOrDefault增加默认值处理public static T CompletableFutureT withinOrDefault( CompletableFutureT future, Duration timeout, T defaultValue) { return future .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS) .handle((r, ex) - ex ! null ? defaultValue : r); }withinOrThrow稍微复杂一些因为它要把超时异常转换为业务自定义异常public static T CompletableFutureT withinOrThrow( CompletableFutureT future, Duration timeout, Supplier? extends Throwable exSupplier) { return future .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS) .handle((r, ex) - { if (ex null) { return CompletableFuture.completedFuture(r); } Throwable cause (ex instanceof CompletionException) ? ex.getCause() : ex; if (cause instanceof TimeoutException) { CompletableFutureT failed new CompletableFuture(); failed.completeExceptionally(exSupplier.get()); return failed; } CompletableFutureT failed new CompletableFuture(); failed.completeExceptionally(cause); return failed; }) .thenCompose(f - f); }有人会问为什么不用completeOnTimeout来实现withinOrDefault因为completeOnTimeout的默认值参数是立即求值的对于复杂对象或需要延迟计算的默认值不友好而handle里使用默认值是惰性的只有超时时才真正用到。4.3 配合线程池的完整调用示例下面是一个典型的多服务并行查询场景。public CompletableFutureUserProfile queryUserProfile(String uid) { CompletableFutureBasicInfo basicFuture FutureTimeout.within( CompletableFuture.supplyAsync(() - rpc.queryBasic(uid), ioPool), Duration.ofMillis(500)); CompletableFutureOrderSummary orderFuture FutureTimeout.withinOrDefault( CompletableFuture.supplyAsync(() - rpc.queryOrders(uid), ioPool), Duration.ofMillis(300), OrderSummary.empty()); CompletableFutureRiskInfo riskFuture FutureTimeout.within( CompletableFuture.supplyAsync(() - rpc.queryRisk(uid), ioPool), Duration.ofMillis(800)); return basicFuture .thenCombine(orderFuture, UserProfile::withOrder) .thenCombine(riskFuture, UserProfile::withRisk) .exceptionally(ex - UserProfile.degraded(uid)); }这里的核心点每个 RPC 调用都有独立的超时时间互不影响。某个服务超时了只会影响最终组合的结果不会拖垮整个接口的响应时间。我还专门做了一个小测试验证within的生效时机long start System.currentTimeMillis(); CompletableFutureString result FutureTimeout.within( CompletableFuture.supplyAsync(() - { LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(3)); return ok; }, ioPool), Duration.ofMillis(800)); result.whenComplete((r, ex) - { System.out.println(cost (System.currentTimeMillis() - start) ex ex); }); // 输出cost约800ms exjava.util.concurrent.CompletionException: java.util.concurrent.TimeoutException实测下来orTimeout的触发时间非常接近设定值800ms 的误差在 5ms 以内。但这个是依赖公共调度器的结果负载高时另说。5. 常见问题与排查技巧实录5.1 超时没生效接口还是等满超时时间这通常不是orTimeout失效而是整个 CompletableFuture 链里有一段没有包上超时。举个例子CompletableFuture.supplyAsync(() - rpc.callA(), pool) .thenCompose(a - rpc.callBAsync(a)) // callBAsync 返回 CompletableFuture .orTimeout(2, TimeUnit.SECONDS);这个写法是对的orTimeout会覆盖到thenCompose返回的新 CompletableFuture。但如果你写成CompletableFuture.supplyAsync(() - rpc.callA(), pool) .orTimeout(2, TimeUnit.SECONDS) .thenCompose(a - rpc.callBAsync(a));那orTimeout只作用于 callA 这一个阶段callBAsync 这个阶段的耗时完全不受限制。排查方法很简单在.orTimeout()前打日志或加peek看阶段名确认超时点是包在整个链外面还是只包了某一段。5.2 超时后线程池线程数还在增长这是最经典的“假超时”。orTimeout标记了 CompletableFuture 为异常完成但底层任务没有停止。尤其当一个任务内部是RestTemplate或者HttpClient的同步调用时它可能卡在 socket 读超时上完全不理会线程中断。这种情况我有两个处理经验对 HTTP 调用设置 client 级别的 connect/read timeout其值要小于 CompletableFuture 的 orTimeout 值。对数据库调用或 RPC 调用设置底层框架的 timeout 参数并且开启“线程中断传播”选项。不能把orTimeout当成底层框架超时的替代品。5.3 异常类型混乱TimeoutException 还是 ExecutionException这是新手最容易懵的地方。直接对 CompletableFuture 调用.get()捕获超时异常时抛的是TimeoutException当orTimeout触发并且你通过handle/whenComplete观察时拿到的是CompletionException且getCause()是TimeoutException如果你在thenApply里继续操作这个已超时的 CompletableFuture它拿到的是CompletionException但 source 是TimeoutException。我建议统一用工具方法做异常解码public static Throwable unwrap(Throwable ex) { Throwable cur ex; while (cur instanceof CompletionException || cur instanceof ExecutionException) { Throwable cause cur.getCause(); if (cause null) { break; } cur cause; } return cur; }然后在任何超时判断处使用Throwable root unwrap(ex); if (root instanceof TimeoutException) { // 超时处理 }这样无论异常被包了几层都能准确判断。5.4 公共调度器过载导致的超时不准前面提到所有orTimeout共享一个内部调度线程虽然它只会执行简单的completeExceptionally操作但海量任务涌入时延迟队列的排序和唤醒依然会产生压力。如果遇到超时不准的现象我建议做一次压测观察orTimeout实际触发时间与设定时间之间的偏差。如果偏差超过 10%就不要再依赖全局调度器了可以自己在业务线程池外层包一个ScheduledExecutorServiceScheduledExecutorService scheduler Executors.newScheduledThreadPool(2); public static T CompletableFutureT withinWithScheduler( CompletableFutureT future, Duration timeout, ScheduledExecutorService scheduler) { CompletableFutureT guard new CompletableFuture(); scheduler.schedule(() - { future.completeExceptionally(new TimeoutException(timeout)); }, timeout.toMillis(), TimeUnit.MILLISECONDS); future.whenComplete((r, ex) - guard.complete(r)); return guard; }注意这个实现里有个小细节future.completeExceptionally只能把还没完成的状态置为异常如果future已经正常完成了这个调用就什么都不做。所以不会误伤已经成功返回的任务。5.5 测试环境一切正常线上却频繁超时这是线程池大小和 IO 等待比例的问题。测试环境里线程多、任务少每个 CompletableFuture 都能及时被调度。线上环境里一个 Tomcat 线程同时发起 10 个异步任务如果线程池核心技术线程只有 8 个而每个任务都要等下游 200ms那 10 个任务排队后最后几个的等待时间会超过设定的超时值。这类问题看线程池监控就能定位活跃线程数长期接近 maxPoolSize队列堆积明显。我的调整思路不是无限调大线程池而是设置合理的超时策略和降级路径。比如把不重要的旁路数据用户画像、推荐位设为withinOrDefault把核心数据设为withinOrThrow配合CallerRunsPolicy或自定义降级池来保证主路径不被拖垮。我在实际项目中做过一次这样的改造把某个详情页的 6 个并行查询改为 3 个withinOrThrow 3 个withinOrDefaultP99 耗时从 1.2s 降到了 350ms线程池拒绝次数也降为 0。6. 更进一步超时与背压的结合有时候不只希望超时后返回降级结果还想控制并发上限防止下游被瞬时流量压垮。CompletableFuture 不适合做复杂的背压控制但可以结合信号量做一个轻量限流。一个简单的做法是给每个异步任务包一层Semaphore.tryAcquireSemaphore permits new Semaphore(4); public CompletableFutureString limitedTask() { if (!permits.tryAcquire()) { return CompletableFuture.completedFuture(limited); } return CompletableFuture.supplyAsync(() - { try { return doRpc(); } finally { permits.release(); } }, ioPool).orTimeout(2, TimeUnit.SECONDS); }为什么需要这个因为 CompletableFuture 的任务一旦提交给线程池线程池只能控制线程数无法控制某个下游服务的并发请求数。如果线程池有 16 个线程下游服务只能承受 4 个并发那 16 个请求同时打过去超时和拒绝会成片出现。这个信号量方案我用在过一个批量对账场景里效果很直接。Batch 任务一次提交 100 个 CompletableFuture信号量限制到 8下游数据库的抖动概率大幅下降。需要注意一点tryAcquire失败时不要抛异常否则调用方要写一堆 catch。直接返回一个默认值 CompletableFuture 更优雅。7. 关于日志与监控超时问题不能靠猜最后分享一个我在实践中反复验证的结论超时问题排查日志里必须有“环节耗时”。我在within工具类里加了可选的耗时埋点返回的 CompletableFuture 会带上一个elapsed字段。这样每次超时日志能直接看到是哪个 stage 超了而不是只看到TimeoutException。public static T CompletableFutureT withinWithLog( CompletableFutureT future, Duration timeout, String stageName, Logger logger) { long start System.nanoTime(); return future.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS) .whenComplete((r, ex) - { long costMs TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); if (ex ! null) { logger.warn(stage{} timeout{}ms cost{}ms error{}, stageName, timeout.toMillis(), costMs, ex.getMessage()); } }); }这个埋点不会增加太多开销但排查问题时能省下大量时间。我遇到过的很多“幽灵超时”最终都是靠这个日志定位到具体某个下游依赖的。监控指标上我一般关注三个future_timeout_total每个 stage 超时次数累计。future_timeout_cost_histogram超时时实际耗时分布。pool_queue_depth线程池队列深度。这三个指标共同配合能很快区分是“下游慢”还是“线程池排队慢”。最后再分享一个小经验我刚开始用 CompletableFuture 时特别喜欢把超时时间设得很短以为这样能保证接口响应快。结果上线后一堆降级数据用户反馈“页面好多信息都看不到”。后来才意识到超时时间要和业务价值挂钩而不是和技术指标挂钩。我的建议是核心主链路的超时时间可以严格一些但旁路数据的超时时间要放宽一点同时用默认值方案做兜底。与其让 500ms 超时丢掉一个本可以 800ms 拿到的推荐结果不如让用户看到这个数据但晚一点或者展示一个默认状态。CompletableFutureTimeout 这个问题说到底不是“设置一个时间参数”这么简单它涉及任务调度、线程池隔离、异常传播、业务降级、监控告警一整条链路。把一个工具类写出来容易把它放到真实环境里经得住流量考验才是另一回事。我上面的这些方案可能不是最优解但都是实测可以用、能扛线上压力的一套组合。你们如果有更好的实践也欢迎在评论区交流尤其是关于中断机制和超时传播这两块的细节我很想看看大家是怎么处理的。
