调用CompletableFuture.allOf()超时后如何安全取消所有待处理Future

核心逻辑 #

Java 原生 CompletableFuture.allOf() 的超时逻辑仅会使 get()/join() 调用抛出异常，不会主动修改内部关联的子Future状态，因此需要手动补充超时后的子Future处理逻辑：

1. 提前将所有待等待的子Future存入可遍历的集合
2. 捕获到 TimeoutException 时，遍历所有子Future，主动调用 completeExceptionally(new TimeoutException()) 完成未完成的子Future
3. 若使用同步回调（非*Async后缀的回调方法），completeExceptionally 会同步执行完该Future关联的所有回调后再返回，因此遍历完成后即可保证所有异常处理器都执行完毕

改造后的示例代码 #

// Example outcomes
final CompletableFuture<String> completedFuture
    = CompletableFuture.completedFuture("hello");
final CompletableFuture<String> failedFuture
    = new CompletableFuture<>();
failedFuture.completeExceptionally(new RuntimeException("Test Stub Exception"));
final CompletableFuture<String> incompleteFuture
    = new CompletableFuture<>();

final AtomicBoolean timeoutHandled = new AtomicBoolean(false);
final CompletableFuture<String> checkedFuture
    = incompleteFuture.whenComplete(
         (x, e) -> timeoutHandled.set(e instanceof TimeoutException));

// 统一收集所有待处理的Future
List<CompletableFuture<?>> allFutures = List.of(completedFuture, checkedFuture, failedFuture);

// this example timeouts after 1ms
try {
    CompletableFuture
        .allOf(allFutures.toArray(new CompletableFuture[0]))
        .get(1, TimeUnit.MILLISECONDS);
} catch (final InterruptedException e) {
    Thread.currentThread().interrupt();
} catch (final TimeoutException e) {
    // 遍历所有Future，主动注入超时异常
    for (CompletableFuture<?> future : allFutures) {
        // 已经完成的Future调用该方法会直接返回false，无副作用
        future.completeExceptionally(e);
    }
}

// 断言现在可以通过
assertTrue(checkedFuture.isCompletedExceptionally());
assertTrue(timeoutHandled.get());

异步回调场景的额外处理 #

如果你的回调使用了 whenCompleteAsync/exceptionallyAsync 等异步回调方法指定了独立线程池执行，那么 completeExceptionally 不会等待异步回调执行完成。如需保证回调全部执行完毕再继续后续逻辑，可以额外收集所有回调返回的Future，最后统一等待：

// 收集所有回调的Future
List<CompletableFuture<?>> callbackFutures = new ArrayList<>();
CompletableFuture<String> checkedFuture = incompleteFuture.whenCompleteAsync(
    (x, e) -> timeoutHandled.set(e instanceof TimeoutException),
    executor
);
// 把回调返回的Future加入集合
callbackFutures.add(checkedFuture);

// 超时注入异常后，等待所有回调执行完成
CompletableFuture.allOf(callbackFutures.toArray(new CompletableFuture[0])).join();

内容的提问来源于stack exchange，提问作者tkruse