Java 并发编程核心：CompletableFuture 实战与原理解析

这些接口直接决定了 CompletableFuture 方法签名的设计逻辑。

CompletableFuture 概览

1. 类结构

CompletableFuture 实现了两个核心接口：Future 和 CompletionStage。

2. 实现 Future 接口

具备 Future 的基本特性，如获取结果、判断是否完成等。需注意 get() 方法在任务结束前会阻塞当前线程。

3. 实现 CompletionStage 接口

这是 CompletableFuture 的核心价值所在。它描述了任务的时序关系：串行、并行或组合。

• 串行：线程 1 完成后执行线程 2。
• 并行：线程 1 和线程 2 独立运行，互不干扰。
• 组合：等待所有任务完成后再进入下一阶段，或者任意一个完成即可触发。

CompletableFuture 提供了约 50 种方法处理各种场景，我们按功能分类来看。

实战案例演示

1. 创建对象

最简单的实例化方式是通过构造函数，配合 get() 获取结果。

CompletableFuture<String> completableFuture = new CompletableFuture<>();
// 注意：此时调用 get() 会永久阻塞，因为任务未执行
String result = completableFuture.get(); 

需手动调用 complete() 结束任务：

completableFuture.complete("Manual Result Here");

2. 异步执行 (runAsync / supplyAsync)

使用 runAsync 进行无返回值的异步计算：

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    System.out.println("运行在单独线程中");
});
future.get();

若需要返回值，使用 supplyAsync：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    return "我有返回值";
});
System.out.println(future.get());

3. 链式调用 (thenApply / thenAccept)

为了避免阻塞，推荐使用回调机制。thenApply 用于转换结果：

CompletableFuture<String> comboText = CompletableFuture.supplyAsync(() -> {
    try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    return "赞";
})
.thenApply(first -> first + ", 在看")
.thenApply(second -> second + ", 转发");

System.out.println(comboText.get());

注意：后续操作不一定与前序操作在同一线程执行，这取决于线程池配置。

如果不需要返回值，仅做副作用操作，可使用 thenAccept：

CompletableFuture<Void> voidFuture = CompletableFuture.supplyAsync(() -> {
    // 模拟 API 调用
    return "Product Object";
}).thenAccept(product -> {
    System.out.println("获取到产品：" + product);
});
voidFuture.get();

4. 无参串行 (thenRun)

thenRun 类似于 thenAccept，但回调函数不接受前序结果：

CompletableFuture.supplyAsync(() -> "Result")
.thenRun(() -> {
    System.out.println("前序操作完成");
});

5. 嵌套 Future 的拍平 (thenCompose)

当业务方法本身返回 CompletableFuture 时，使用 thenApply 会导致嵌套（CompletableFuture<CompletableFuture<T>>）。此时应使用 thenCompose 进行'拍平'：

// 假设方法返回 CompletableFuture<User>
CompletableFuture<User> getUsersDetail(String userId) {
    return CompletableFuture.supplyAsync(() -> new User(123L, "Name"));
}

// 假设方法返回 CompletableFuture<Double>
CompletableFuture<Double> getCreditRating(User user) {
    return CompletableFuture.supplyAsync(() -> 7.5);
}

// 错误写法：产生嵌套
CompletableFuture<CompletableFuture<Double>> badResult = 
    getUsersDetail("123").thenApply(user -> getCreditRating(user));

// 正确写法：使用 thenCompose
CompletableFuture<Double> goodResult = 
    getUsersDetail("123").thenCompose(user -> getCreditRating(user));

这与 Stream 中的 flatMap 逻辑一致。

6. 聚合结果 (thenCombine / allOf / anyOf)

聚合两个独立 Future 的结果：

CompletableFuture<Double> weightFuture = CompletableFuture.supplyAsync(() -> 65.0);
CompletableFuture<Double> heightFuture = CompletableFuture.supplyAsync(() -> 183.8);

CompletableFuture<Double> bmiFuture = weightFuture
    .thenCombine(heightFuture, (weight, height) -> {
        double hInMeter = height / 100;
        return weight / (hInMeter * hInMeter);
    });

System.out.println("BMI: " + bmiFuture.get());

对于多个任务，可使用 allOf（全部完成）或 anyOf（任意一个完成）。

7. 异常处理 (exceptionally / handle)

exceptionally 相当于 catch，捕获异常并返回默认值：

CompletableFuture<String> maturityFuture = CompletableFuture.supplyAsync(() -> {
    if (true) throw new IllegalArgumentException("非法年龄");
    return "成年人";
}).exceptionally(ex -> {
    System.out.println("发生异常：" + ex.getMessage());
    return "Unknown";
});

handle 则更强大，同时处理正常结果和异常（类似 finally）：

CompletableFuture<String> resultFuture = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Error");
}).handle((res, ex) -> {
    if (ex != null) {
        return "Handled Exception";
    }
    return res;
});

线程池机制

带有 Async 后缀的方法默认会使用线程池执行。如果不指定，默认使用 ForkJoinPool.commonPool()，其线程数通常为 CPU 核心数。

private static final Executor ASYNC_POOL = ForkJoinPool.commonPool();

最佳实践：不要所有业务共用一个线程池。若某任务涉及慢速 I/O，可能导致整个线程池饥饿，影响系统性能。建议根据业务类型隔离线程池。

结语

本文涵盖了 CompletableFuture 的核心用法。JDK 9 对其进行了部分升级，实际开发中可根据需求选择合适的方法。掌握这些工具，能让你的并发代码更加健壮且易读。