Java CompletableFuture组合异步操作的超时处理策略

在异步编程中，单个任务的超时控制相对简单，但多个 CompletableFuture 组合操作（如并行执行、链式调用）的超时处理往往容易失控。若未正确设置超时，系统线程可能被长时间阻塞，导致资源耗尽。

本文将针对三种常见场景，提供可直接落地的超时处理策略。

场景一：单任务超时降级（Java 9+ 方案）

如果你使用的是 Java 9 或更高版本，最直接的方法是使用 orTimeout 或 completeOnTimeout 方法。前者用于抛出异常，后者用于返回默认值。

1. 引入 必要的工具类或模拟一个耗时任务。

   public CompletableFuture<String> fetchData() {
       return CompletableFuture.supplyAsync(() -> {
           try { Thread.sleep(5000); } catch (InterruptedException e) {}
           return "原始数据";
       });
   }

2. 设置 超时异常抛出。如果任务在 2 秒内未完成，抛出 TimeoutException。

   CompletableFuture<String> future = fetchData()
           .orTimeout(2, TimeUnit.SECONDS);

3. 设置 超时默认返回值。如果任务超时，返回 指定的默认值 "默认数据"，而不抛出异常。

   CompletableFuture<String> future = fetchData()
           .completeOnTimeout("默认数据", 2, TimeUnit.SECONDS);

场景二：多任务并行执行的全局超时

当使用 CompletableFuture.allOf 或 anyOf 组合多个任务时，仅对单个任务设置超时是不够的。你需要控制整个批次的总超时时间。

1. 构建并行任务

假设你需要同时请求三个接口，只要有一个超时即视为失败。

1. 创建 一组异步任务。

   CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> "Task1");
   CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> "Task2");
   CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> "Task3");

2. 组合 所有任务。使用 allOf 等待所有任务完成。

   CompletableFuture<Void> allTasks = CompletableFuture.allOf(task1, task2, task3);

2. 实现全局超时控制

直接在 allTasks 上调用 orTimeout 是最简便的方法。

1. 添加 全局超时限制。设定整个组合操作最多等待 3 秒。

   CompletableFuture<Void> timeoutFuture = allTasks
           .orTimeout(3, TimeUnit.SECONDS);

2. 处理 异常情况。当超时发生时，捕获 TimeoutException 并进行日志记录或降级处理。

   timeoutFuture.exceptionally(ex -> {
       System.err.println("全局任务超时: " + ex.getMessage());
       return null;
   });

场景三：兼容 Java 8 的手动调度方案

如果你的项目运行在 Java 8 环境下，没有 orTimeout 方法，需要通过 ScheduledExecutorService 手动实现超时机制。

1. 定义通用超时方法

为了避免重复代码，封装 一个通用的超时处理工具方法。

1. 创建 一个静态方法 within，接收 Future、超时时间和 Scheduler。

   public static <T> CompletableFuture<T> within(CompletableFuture<T> future, long timeout, TimeUnit unit, ScheduledExecutorService scheduler) {
       final CompletableFuture<T> result = new CompletableFuture<>();
   
       // 1. 当原 Future 完成时，将结果传递给 result
       future.whenComplete((data, err) -> {
           if (err != null) {
               result.completeExceptionally(err);
           } else {
               result.complete(data);
           }
       });
   
       // 2. 启动一个定时任务，超时后尝试打断 result
       scheduler.schedule(() -> {
           if (!result.isDone()) {
               result.completeExceptionally(new TimeoutException("操作超时"));
           }
       }, timeout, unit);
   
       return result;
   }

2. 应用到组合操作

1. 初始化 一个延迟线程池。

   ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

2. 应用 上述工具方法。将 allTasks 包装在 within 方法中。

   CompletableFuture<Void> safeFuture = within(allTasks, 3, TimeUnit.SECONDS, scheduler);

3. 验证 结果。如果超时，safeFuture 会抛出异常，从而中断等待链条。

异常处理与资源清理策略

超时发生后，仅仅抛出异常是不够的，底层的线程任务可能仍在运行，浪费资源。

1. 确认超时触发流程

以下是超时处理的逻辑流向：

2. 实现自动取消底层任务

标准的 CompletableFuture 超时并不会自动取消正在运行的线程。你需要在异常处理中显式调用 cancel。

1. 修改 异常处理逻辑。在捕获超时异常后，调用 cancel 方法中断线程。

   CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
       // 模拟长时间运行
       try { Thread.sleep(10000); } catch (InterruptedException e) {}
       return "Done";
   });
   
   // 使用 Java 9+ orTimeout 或 Java 8 工具方法
   future.orTimeout(1, TimeUnit.SECONDS)
         .exceptionally(ex -> {
             if (ex instanceof TimeoutException) {
                 // 关键步骤：取消底层任务
                 future.cancel(true); 
                 return "请求超时，已降级";
             }
             return "其他错误";
         });

2. 检查 任务内的中断标志。在耗时的异步任务代码中，添加 对 Thread.currentThread().isInterrupted() 的检查，以便快速响应取消操作。

   if (Thread.currentThread().isInterrupted()) {
       return "Cancelled"; // 快速退出
   }

策略对比与选择

根据项目实际情况，选择 合适的方案。

执行 以下操作完成最终落地：

1. 若项目升级至 Java 9+，优先使用 orTimeout。
2. 若必须兼容 Java 8，使用文中提供的 within 工具方法。
3. 无论使用哪种方案，务必在 exceptionally 中调用 future.cancel(true) 释放资源。