Java CompletableFuture 实战指南：场景、最佳实践与优化

为什么使用 thenCompose 而不是 thenApply？

• 如果验证失败，异常会被抛出，后续的 thenAccept 不会执行
• 异常会被 exceptionally 捕获并处理
• 如果使用 thenApply，这个查询会在验证任务的线程中同步执行，阻塞线程
• 使用 thenCompose，查询会在新的线程中异步执行，不阻塞

对比示例：

// ❌ 错误用法：使用 thenApply（会产生嵌套的 Future）
.thenApply(isValid -> {
    if (!isValid) {
        throw new RuntimeException("用户验证失败");
    }
    return CompletableFuture.supplyAsync(() -> dataService.getUserData(userId));
    // 返回类型：CompletableFuture<CompletableFuture<UserData>>
})

// ✅ 正确用法：使用 thenCompose（扁平化 Future）
.thenCompose(isValid -> {
    if (!isValid) {
        throw new RuntimeException("用户验证失败");
    }
    return CompletableFuture.supplyAsync(() -> dataService.getUserData(userId));
    // 返回类型：CompletableFuture<UserData>
})

适用场景：

• 多步骤的业务流程（如订单处理：验证 → 扣款 → 发货 → 通知）
• 需要依赖前一步结果的场景
• 需要条件判断决定是否继续执行的场景
• 下一步是重量级异步操作的场景

1.3 超时控制和降级处理：提高系统可用性（快速失败，避免无限等待）

设置超时时间，超时后使用降级方案，提高系统可用性。这是生产环境中必须考虑的场景。

public String getDataWithFallback(String key) {
    CompletableFuture<String> remoteFuture = CompletableFuture.supplyAsync(() -> remoteService.getData(key)); // 可能很慢或失败
    CompletableFuture<String> localFuture = CompletableFuture.supplyAsync(() -> localCache.getData(key)); // 降级方案

    // 设置超时时间
    CompletableFuture<String> timeoutFuture = remoteFuture
            .orTimeout(500, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> {
                if (ex instanceof TimeoutException) {
                    log.warn("远程服务超时，使用本地缓存");
                    return localFuture.join();
                }
                return localFuture.join();
            });
    return timeoutFuture.join();
}

适用场景：

• 调用外部服务，可能超时或失败
• 需要降级方案保证系统可用性
• 对响应时间有要求的场景

1.4 批量任务处理：批量提交，并行执行，统一收集（性能提升可达数十倍）

批量提交任务，并行执行，统一收集结果，提高处理效率。这是 CompletableFuture 在批量处理场景中的典型应用。

public List<String> processBatchTasks(List<String> tasks) {
    // 批量提交任务
    List<CompletableFuture<String>> futures = tasks.stream()
            .map(task -> CompletableFuture.supplyAsync(() -> processTask(task)))
            .collect(Collectors.toList());

    // 统一等待和收集
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    return futures.stream().map(CompletableFuture::join).collect(Collectors.toList());
}

性能提升：

• 串行处理：总耗时 = 任务数 × 单个任务耗时
• 并行处理：总耗时 ≈ 单个任务耗时（如果线程池足够大）
• 性能提升：可达数倍甚至数十倍

适用场景：

• 批量数据处理
• 批量 API 调用
• 批量文件处理

二、最佳实践与注意事项（Best Practices）

2.1 线程池管理：合理配置线程池大小和类型（避免线程泄漏和资源浪费）

根据任务特点选择合适的线程池大小和类型，避免线程泄漏和资源浪费。这是 CompletableFuture 性能优化的关键。

// ✅ 正确：为 IO 密集型任务使用自定义线程池
ExecutorService ioExecutor = new ThreadPoolExecutor(
    10, 50, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000),
    new ThreadFactoryBuilder().setNameFormat("io-pool-%d").build()
);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // IO 密集型任务
    return httpClient.get(url);
}, ioExecutor);

// ✅ 正确：为 CPU 密集型任务使用 ForkJoinPool
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
    // CPU 密集型任务
    return computeHeavyTask();
});

最佳实践：

• CPU 密集型任务：强调计算：使用默认的 ForkJoinPool，线程数 = CPU 核心数
• IO 密集型任务：强调能同时拉取更多的数据，使用自定义线程池，线程数可以更大（如 50-100）
• 避免线程泄漏：长时间运行的任务建议使用自定义线程池，便于管理
• 命名线程池：使用有意义的线程名称，便于问题排查

2.2 异常处理：完善的异常处理机制（避免异常被吞掉，保证系统稳定）

务必处理异常，避免异常被吞掉，影响业务逻辑。

// ✅ 正确：使用 exceptionally 处理异常
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 可能抛出异常的操作
    return riskyOperation();
}).exceptionally(ex -> {
    // 异常处理
    log.error("操作失败", ex);
    return "默认值";
});

// ✅ 正确：使用 handle 统一处理成功和异常
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "结果")
    .handle((result, ex) -> {
        if (ex != null) {
            log.error("处理异常", ex);
            return "异常处理";
        }
        return result;
    });

最佳实践：

• 记录日志：异常发生时记录详细日志，便于排查问题
• 提供降级方案：异常时返回默认值或使用降级逻辑
• 避免吞掉异常：不要忽略异常，至少要记录日志

2.3 避免阻塞：尽量使用回调而不是阻塞等待（提高系统响应能力）

尽量使用回调而不是阻塞等待，避免阻塞主线程。这是 CompletableFuture 的核心优势之一。

// ❌ 错误：在主线程中阻塞等待
String result = future.get(); // 阻塞主线程

// ✅ 正确：使用回调处理
future.thenAccept(result -> {
    // 处理结果，不阻塞主线程
    processResult(result);
});

适用场景：

• Web 请求处理：使用回调，避免阻塞请求线程
• 事件驱动系统：使用回调，响应事件
• 批处理任务：可以使用 join() 等待，因为是后台任务

三、性能优化建议（Performance Optimization）

3.1 合理使用线程池：根据任务特点选择合适的线程池大小和类型（避免资源浪费和性能瓶颈）

根据任务特点选择合适的线程池大小和类型，避免资源浪费和性能瓶颈。这是性能优化的第一步。

• CPU 密集型：线程数 = CPU 核心数，使用 ForkJoinPool
• IO 密集型：线程数可以更大（如 50-100），使用 ThreadPoolExecutor
• 混合型：根据 IO 等待时间调整线程数

3.2 避免过度嵌套：链式调用不要过深（提高代码可读性和性能）

核心结论：链式调用不要过深，影响可读性和性能。扁平化处理可以让代码更清晰，性能更好。

// ❌ 过度嵌套
CompletableFuture.supplyAsync(() -> ...)
    .thenCompose(r1 -> CompletableFuture.supplyAsync(() -> ...)
        .thenCompose(r2 -> CompletableFuture.supplyAsync(() -> ...)
            .thenCompose(r3 -> ...)));

// ✅ 扁平化处理
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> ...);
CompletableFuture<String> f2 = f1.thenCompose(r1 -> CompletableFuture.supplyAsync(() -> ...));
CompletableFuture<String> f3 = f2.thenCompose(r2 -> CompletableFuture.supplyAsync(() -> ...));

3.3 批量处理优化：多个相似任务可以批量处理（性能提升可达数十倍）

// 批量提交任务
List<CompletableFuture<String>> futures = tasks.stream()
    .map(task -> CompletableFuture.supplyAsync(() -> processTask(task)))
    .collect(Collectors.toList());

// 统一等待和收集
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();