4. 异步任务编排基础：线程池原理、Future/Callable、CompletableFuture 入门

在中间件集成场景中，数据库操作、缓存更新、远程调用往往需要并发执行或编排依赖关系。本章从最基础的线程池原理出发，逐步深入到 Future/Callable 模型，最后引入 CompletableFuture 这一现代异步编排利器，为后续的通信机制优化打下坚实基础。

4.1 线程池核心原理与参数调优

线程池是异步任务执行的底层容器。它通过复用线程、控制并发数、管理任务队列来避免频繁创建/销毁线程的开销。Java 中 ThreadPoolExecutor 是最核心的实现。

4.1.1 核心参数

4.1.2 线程池工作流程

下图展示了线程池从提交任务到执行或拒绝的完整生命周期：

4.2 Future 与 Callable：获取异步结果

Callable 与 Runnable 的区别在于：Callable 可以返回结果并抛出异常。通过 Future 接口，我们可以获取异步任务的执行状态和最终结果。

4.2.1 基本用法

// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(4);

// 提交 Callable 任务，返回 Future
Future<String> future = executor.submit(() -> {
    // 模拟耗时计算
    Thread.sleep(1000);
    return "任务结果";
});

// 主线程可以继续做其他事情
System.out.println("等待结果...");

// 获取结果（阻塞直到任务完成）
String result = future.get();  // 可指定超时：future.get(2, TimeUnit.SECONDS)
System.out.println("结果: " + result);

executor.shutdown();

4.2.2 Future 的局限性

• 阻塞获取：get() 方法会阻塞调用线程，无法实现真正的非阻塞回调。
• 无法编排：多个 Future 之间难以表达依赖关系（如 A 完成后执行 B，或 A、B 都完成后执行 C）。
• 异常处理不便：异常被封装在 ExecutionException 中，需要手动拆解。
• 无法组合：没有提供 thenApply、thenCombine 等组合方法。

4.3 CompletableFuture 入门：声明式异步编排

CompletableFuture 是 Java 8 引入的异步编程增强工具，它实现了 Future 和 CompletionStage 接口，提供了函数式编程风格的异步任务编排能力。

4.3.1 核心概念

4.3.2 典型编排模式

下图展示了 CompletableFuture 在中间件集成中的典型编排场景：

4.3.3 代码示例：缓存+数据库+远程调用编排

// 自定义线程池（避免使用 ForkJoinPool.commonPool()）
ExecutorService executor = Executors.newFixedThreadPool(8);

// 任务A：查询缓存
CompletableFuture<String> cacheFuture = CompletableFuture.supplyAsync(() -> {
    // 模拟缓存查询
    return cacheService.get(key);
}, executor);

// 任务B：查询数据库
CompletableFuture<String> dbFuture = CompletableFuture.supplyAsync(() -> {
    // 模拟数据库查询
    return databaseService.query(sql);
}, executor);

// 任务C：远程调用
CompletableFuture<String> rpcFuture = CompletableFuture.supplyAsync(() -> {
    // 模拟远程RPC
    return rpcService.call(request);
}, executor);

// 编排：A和B合并后，再与C合并，最后处理异常
CompletableFuture<String> resultFuture = cacheFuture
    .thenCombine(dbFuture, (cacheResult, dbResult) -> {
        // 合并缓存和数据库结果
        return mergeCacheAndDb(cacheResult, dbResult);
    })
    .thenCombine(rpcFuture, (mergedResult, rpcResult) -> {
        // 合并中间结果与RPC结果
        return buildFinalResponse(mergedResult, rpcResult);
    })
    .exceptionally(ex -> {
        // 统一异常处理，返回兜底数据
        log.error("异步编排失败", ex);
        return "fallback_response";
    });

// 非阻塞获取最终结果
String finalResult = resultFuture.get(5, TimeUnit.SECONDS);

4.4 线程池与 CompletableFuture 的配合

CompletableFuture 默认使用 ForkJoinPool.commonPool()，但在生产环境中，必须显式传入自定义线程池，原因如下：

• 资源隔离：不同业务模块使用独立线程池，避免相互影响。
• 避免饥饿：commonPool 的并行度默认等于 CPU 核数，IO 密集型任务会耗尽线程。
• 监控可控：自定义线程池可以设置名称、监控队列深度、拒绝策略等。

// 推荐：为 CompletableFuture 创建专用线程池
ThreadPoolExecutor asyncPool = new ThreadPoolExecutor(
    8,                    // corePoolSize
    16,                   // maximumPoolSize
    60, TimeUnit.SECONDS, // keepAliveTime
    new ArrayBlockingQueue<>(1000),  // 有界队列
    new ThreadFactoryBuilder()
        .setNameFormat("async-pool-%d")
        .build(),
    new ThreadPoolExecutor.CallerRunsPolicy()  // 调用者运行策略
);

// 所有 CompletableFuture 都使用该线程池
CompletableFuture.supplyAsync(() -> doSomething(), asyncPool);