使用CompletableFuture实现带超时和异常恢复的异步任务编排 -

题目

使用CompletableFuture实现带超时和异常恢复的异步任务编排

信息

类型：问答
难度：⭐⭐⭐

考点

CompletableFuture高级用法,异步编程,异常处理,超时控制,函数式编程

快速回答

实现要点：

使用supplyAsync启动异步任务，指定自定义线程池
通过orTimeout()设置超时控制（Java 9+）
使用exceptionally()处理业务异常和超时异常
通过thenCompose()实现任务链式编排
使用completeOnTimeout()提供超时默认值
注意线程池资源管理和异常传播机制

## 解析

场景需求

在分布式系统中，需要编排多个异步服务调用：

调用用户服务获取用户信息
基于用户信息调用订单服务
两个服务都可能超时或抛出异常
要求：
- 每个调用超时时间为800ms
- 任一服务失败时提供降级结果
- 订单服务依赖用户服务结果

完整解决方案

import java.util.concurrent.*;
import java.util.function.Function;

public class AsyncServiceOrchestrator {

    // 自定义线程池避免使用公共ForkJoinPool
    private static final ExecutorService executor = 
        Executors.newFixedThreadPool(4, r -> {
            Thread t = new Thread(r);
            t.setDaemon(true); // 避免阻止JVM关闭
            return t;
        });

    // 模拟远程服务调用
    private String callUserService(String userId) throws InterruptedException {
        Thread.sleep(ThreadLocalRandom.current().nextInt(1000)); // 随机延迟
        if ("error".equals(userId)) throw new RuntimeException("User service error");
        return "UserInfo:" + userId;
    }

    private String callOrderService(String userInfo) throws InterruptedException {
        Thread.sleep(ThreadLocalRandom.current().nextInt(1000));
        if (userInfo.contains("invalid")) throw new RuntimeException("Order service error");
        return "Orders for " + userInfo;
    }

    public CompletableFuture<String> getUserOrders(String userId) {
        // 第一阶段：调用用户服务（带超时和异常恢复）
        CompletableFuture<String> userFuture = CompletableFuture
            .supplyAsync(() -> {
                try {
                    return callUserService(userId);
                } catch (InterruptedException e) {
                    throw new CompletionException(e);
                }
            }, executor)
            .orTimeout(800, TimeUnit.MILLISECONDS) // Java 9+ 超时控制
            .exceptionally(ex -> {
                System.err.println("用户服务降级: " + ex.getCause().getMessage());
                return "fallback_user"; // 提供降级值
            });

        // 第二阶段：链式调用订单服务
        return userFuture.thenCompose(userInfo -> 
            CompletableFuture
                .supplyAsync(() -> {
                    try {
                        return callOrderService(userInfo);
                    } catch (InterruptedException e) {
                        throw new CompletionException(e);
                    }
                }, executor)
                .completeOnTimeout("订单服务超时默认值", 800, TimeUnit.MILLISECONDS)
                .exceptionally(ex -> {
                    System.err.println("订单服务降级: " + ex.getCause().getMessage());
                    return "fallback_orders";
                })
        );
    }

    public static void main(String[] args) throws Exception {
        AsyncServiceOrchestrator orchestrator = new AsyncServiceOrchestrator();

        // 正常情况
        orchestrator.getUserOrders("123")
            .thenAccept(System.out::println)
            .join();

        // 用户服务超时
        orchestrator.getUserOrders("slow")
            .thenAccept(System.out::println)
            .join();

        // 订单服务异常
        orchestrator.getUserOrders("invalid")
            .thenAccept(System.out::println)
            .join();

        executor.shutdown();
    }
}

核心原理说明

异步编排：thenCompose()用于连接有依赖关系的异步任务，避免回调地狱
超时控制：
- orTimeout()：超时抛出CompletionException（需Java 9+）
- completeOnTimeout()：超时返回默认值不抛异常
异常处理：
- exceptionally()捕获所有异常（包括超时）并提供恢复值
- 业务异常需包装为CompletionException
线程模型：
- 使用独立线程池避免ForkJoinPool资源竞争
- Daemon线程确保JVM正常退出

最佳实践

资源管理：始终为异步任务指定专用线程池，根据任务类型（IO/CPU密集型）配置合适大小
超时设置：分布式调用必须设置超时，时间应大于P99响应时间
异常隔离：每个异步阶段单独处理异常，避免级联失败
降级策略：提供有意义的默认值，记录原始异常
组合操作：使用allOf()/anyOf()处理并行任务

常见错误

❌ 使用默认ForkJoinPool导致资源耗尽
❌ 在exceptionally()中未处理TimeoutException
❌ 忘记调用join()/get()导致主线程提前退出
❌ 在异步任务中阻塞线程（如使用同步IO）
❌ 未处理CompletionException包装的底层异常

扩展知识

响应式编程：复杂场景考虑使用Reactor或RxJava
虚拟线程：Java 19+虚拟线程可简化异步代码编写
性能优化：
- 使用CompletableFuture的异步回调避免线程阻塞
- 监控线程池队列堆积情况

Java 8兼容方案：

// 手动实现超时
CompletableFuture.supplyAsync(() -> ...)
    .applyToEither(timeoutAfter(800, TimeUnit.MILLISECONDS), 
        Function.identity())
    .exceptionally(...);

static <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) {
    CompletableFuture<T> result = new CompletableFuture<>();
    ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    scheduler.schedule(() -> result.completeExceptionally(new TimeoutException()), timeout, unit);
    return result;
}