在现代 Java 开发中，异步编程已成为提升应用性能和响应性的关键技术之一。Java 8 引入的 CompletableFuture 为我们提供了一个强大的工具，用于构建可组合、非阻塞的异步流程。本文将从基础概念开始，逐步演示如何使用 CompletableFuture 搭建一个异步流水线，并分享几个常见的实战技巧。

1. CompletableFuture 基础

CompletableFuture 继承自 Future 接口，并且实现了 CompletionStage 接口，意味着它不仅可以像 Future 一样等待结果，还能在结果完成后继续执行后续操作。其核心优势在于：

• 异步执行：不阻塞调用线程，结果通过回调或继续链式处理；
• 组合能力：可以并行执行多个任务，并在全部完成后合并结果；
• 异常处理：支持统一的异常捕获和恢复机制。

2. 构建一个简单的异步流水线

假设我们需要完成以下业务流程：

1. 从数据库读取用户信息（耗时操作）；
2. 根据用户信息调用外部支付接口进行支付（耗时操作）；
3. 将支付结果写回数据库并发送通知。

我们可以将每一步封装为 CompletableFuture，并使用 thenCompose 将它们串联起来。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class PaymentPipeline {

    private final ExecutorService dbExecutor = Executors.newFixedThreadPool(5);
    private final ExecutorService paymentExecutor = Executors.newFixedThreadPool(10);

    public CompletableFuture <Void> processPayment(String userId, double amount) {
        // 第一步：读取用户信息
        CompletableFuture <User> userFuture = CompletableFuture.supplyAsync(
                () -> dbQueryUser(userId), dbExecutor);

        // 第二步：执行支付
        CompletableFuture <PaymentResult> paymentFuture = userFuture.thenCompose(
                user -> CompletableFuture.supplyAsync(
                        () -> callPaymentApi(user, amount), paymentExecutor));

        // 第三步：写入结果并发送通知
        return paymentFuture.thenAcceptAsync(
                result -> {
                    dbUpdatePaymentResult(result);
                    sendNotification(result);
                }, dbExecutor);
    }

    // 以下为占位方法，实际业务中会有完整实现
    private User dbQueryUser(String userId) { /* ... */ }
    private PaymentResult callPaymentApi(User user, double amount) { /* ... */ }
    private void dbUpdatePaymentResult(PaymentResult result) { /* ... */ }
    private void sendNotification(PaymentResult result) { /* ... */ }
}

关键点说明

• supplyAsync：在指定线程池中执行耗时任务，返回 `CompletableFuture `；
• thenCompose：将前一步的结果作为参数传递给后续异步任务，保持链式调用；
• thenAcceptAsync：在完成后执行副作用操作（如写入 DB、发送通知），同样指定线程池。

3. 并行执行与组合

在某些场景下，我们需要并行执行多个子任务，然后合并结果。例如，批量查询订单详情：

List<CompletableFuture<Order>> futures = orderIds.stream()
        .map(id -> CompletableFuture.supplyAsync(() -> fetchOrder(id), dbExecutor))
        .collect(Collectors.toList());

CompletableFuture <Void> allDone = CompletableFuture.allOf(
        futures.toArray(new CompletableFuture[0]));

CompletableFuture<List<Order>> allOrders = allDone.thenApply(v ->
        futures.stream()
               .map(CompletableFuture::join)
               .collect(Collectors.toList()));

• CompletableFuture.allOf：等待所有任务完成；
• join：获取已完成任务的结果，若有异常会抛出 CompletionException。

4. 异常处理

CompletableFuture 的异常处理方式与传统 Future 不同。我们可以在链中使用 exceptionally、handle 或 whenComplete：

paymentFuture
    .exceptionally(ex -> {
        log.error("Payment failed", ex);
        return defaultPaymentResult(); // 兜底结果
    })
    .thenAcceptAsync(...);

• exceptionally：仅处理异常，返回替代值；
• handle：无论成功或失败都执行，能同时访问结果和异常；
• whenComplete：不影响链中后续步骤，常用于日志记录。

5. 性能调优建议

1. 合理配置线程池：CPU 密集型任务使用 Executors.newWorkStealingPool()，I/O 密集型使用固定大小池；
2. 避免线程切换：尽量在同一个线程池内完成相关操作，减少上下文切换；
3. 使用 ForkJoinPool.commonPool()：对于不需要高并发控制的轻量任务，可以直接使用全局共享池；
4. 监控任务延迟：通过自定义 CompletableFuture 或使用 CompletableFuture 的 thenApplyAsync 包装监控逻辑。

6. 总结

• CompletableFuture 让 Java 的异步编程变得更直观、可组合；
• 通过 thenCompose、thenAcceptAsync 等方法可以轻松构建流水线式流程；
• 并行组合、异常处理与性能调优是使用时需要关注的关键点。

掌握这些技巧后，你就可以在自己的项目中用 CompletableFuture 替代繁琐的回调、线程池管理，让代码更简洁、易维护，同时充分利用多核 CPU 提升吞吐量。祝你编码愉快！