CompletableFuture 异步编程

前言

在 Java 中，异步执行任务通常通过线程池 Executor 来实现：

• 不需要返回值：任务实现 Runnable 接口；
• 需要返回值：任务实现 Callable 接口，调用 Executor.submit()，再通过 Future 获取结果。

然而，当多个异步任务之间存在依赖关系（如 A 的结果传给 B，或 A、B 都完成后再执行 C）时，使用 CountDownLatch、CyclicBarrier 等同步组件会比较繁琐。

CompletableFuture 正是为解决这类场景而生。它不仅扩展了 Future 的能力，还提供了函数式编程风格的链式调用，让异步任务的组合变得简洁优雅。

---

什么是 CompletableFuture

CompletableFuture 是 Java 8 引入的异步编程类，它同时实现了 Future 和 CompletionStage 两个接口：

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
}

• Future 接口：提供查询任务状态、获取结果、取消任务等基础能力，但功能有限（无法链式回调、无法组合任务）。
• CompletionStage 接口：描述异步计算的一个阶段，支持将多个阶段串联组合，形成异步计算的流水线。

CompletableFuture 在 Future 的基础上，增加了函数式回调、任务组合、异常处理等能力，是 Java 异步编程的核心工具。

Future 接口的方法

方法	说明
cancel(boolean mayInterruptIfRunning)	尝试取消任务
isCancelled()	判断任务是否被取消
isDone()	判断任务是否已完成
get()	阻塞等待并获取结果
get(long timeout, TimeUnit unit)	带超时地等待结果

Future 的局限性：get() 方法会阻塞线程，且无法在任务完成后自动执行回调，也无法将多个 Future 组合起来。

---

使用步骤

一、创建 CompletableFuture

创建 CompletableFuture 有两种常见方式：

1. new 关键字：手动创建并手动完成（适合需要外部控制完成时机的场景）
2. 静态工厂方法：runAsync() / supplyAsync()（适合自动执行异步任务的场景）

1. new 关键字

通过 new 创建的 CompletableFuture 不会自动执行任何任务，需要手动调用 complete() 传入结果：

// 创建异步运算的载体
CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();

// 在未来的某个时刻，手动传入结果
resultFuture.complete(rpcResponse);

// 阻塞等待并获取结果
rpcResponse = resultFuture.get();

complete() 只能调用一次，后续调用会被忽略。

如果已知结果，可以使用 completedFuture() 创建一个已完成的 CompletableFuture：

CompletableFuture<String> future = CompletableFuture.completedFuture("hello!");
assertEquals("hello!", future.get());

代码示例：

@Test
void testNew() throws ExecutionException, InterruptedException {
    // 1. 正常完成
    CompletableFuture<String> future = new CompletableFuture<>();
    future.complete("Hello, World!");
    assertEquals("Hello, World!", future.get());

    // 2. 异常完成
    CompletableFuture<Void> failingFuture = new CompletableFuture<>();
    failingFuture.completeExceptionally(new RuntimeException("Oops!"));

    try {
        failingFuture.get();
        fail("Expected exception not thrown");
    } catch (ExecutionException e) {
        assertInstanceOf(RuntimeException.class, e.getCause());
        assertEquals("Oops!", e.getCause().getMessage());
    }
}

2. 静态工厂方法

方法	参数	返回值	适用场景
supplyAsync(Supplier)	Supplier<U>	CompletableFuture<U>	需要返回值
runAsync(Runnable)	Runnable	CompletableFuture<Void>	不需要返回值

两者都支持传入自定义线程池（推荐），否则默认使用 ForkJoinPool.commonPool()。

// supplyAsync：有返回值
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);

// runAsync：无返回值
public static CompletableFuture<Void> runAsync(Runnable runnable);
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);

代码示例：

@Test
void testSupplyAsyncAndRunAsync() throws ExecutionException, InterruptedException {
    // runAsync：无返回值
    CompletableFuture<Void> runFuture = CompletableFuture.runAsync(
            () -> System.out.println("hello runAsync!"));
    runFuture.get();

    // supplyAsync：有返回值
    CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(
            () -> "hello supplyAsync!");
    assertEquals("hello supplyAsync!", supplyFuture.get());
}

@Test
void testWithCustomExecutor() {
    // 自定义线程池（推荐）
    ExecutorService executor = Executors.newCachedThreadPool();

    CompletableFuture<Void> runFuture = CompletableFuture.runAsync(
            () -> System.out.println("run, cmty256"), executor);

    CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(
            () -> "supply, cmty256", executor);

    System.out.println(runFuture.join());   // null
    System.out.println(supplyFuture.join()); // supply, cmty256

    executor.shutdown();
}

---

二、简单任务异步回调

获取异步结果后，可以使用回调方法对结果进行进一步处理。常用方法如下：

方法	接收上一步结果	有返回值	说明
thenRun / thenRunAsync	✗	✗	执行下一个任务，无入参无返回值
thenAccept / thenAcceptAsync	✓	✗	消费上一步结果，无返回值
thenApply / thenApplyAsync	✓	✓	转换上一步结果，有返回值
whenComplete	✓	✗	任务完成后回调（正常或异常都触发），返回值沿用上一步

同步版与异步版的区别

• thenXxx：沿用上一步任务使用的线程池
• thenXxxAsync：使用默认的 ForkJoinPool（或传入的自定义线程池）

private static final Executor asyncPool = useCommonPool
        ? ForkJoinPool.commonPool()
        : new ThreadPerTaskExecutor();

public CompletableFuture<Void> thenRun(Runnable action) {
    return uniRunStage(null, action);       // 沿用上一步的线程池
}

public CompletableFuture<Void> thenRunAsync(Runnable action) {
    return uniRunStage(asyncPool, action);  // 使用 asyncPool
}

Tips：thenAccept / thenAcceptAsync、thenApply / thenApplyAsync 等的区别同理。

thenRun

执行完第一个任务后，再执行第二个任务。无入参、无返回值。

@Test
void testThenRun() throws ExecutionException, InterruptedException {
    CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("第一个任务执行");
        return "沉梦听雨";
    });

    CompletableFuture<Void> thenRunFuture = firstFuture.thenRun(() ->
            System.out.println("第二个任务执行"));

    System.out.println("返回值：" + thenRunFuture.get());
    // 输出：
    // 第一个任务执行
    // 第二个任务执行
    // 返回值：null
}

thenAccept

接收上一步的结果作为入参，但无返回值。

@Test
void testThenAccept() throws ExecutionException, InterruptedException {
    CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("第一个任务执行");
        return "沉梦听雨";
    });

    CompletableFuture<Void> thenAcceptFuture = firstFuture.thenAccept(result -> {
        if ("沉梦听雨".equals(result)) {
            System.out.println("入参校验成功");
        }
        System.out.println("第二个任务执行，收到：" + result);
    });

    System.out.println("返回值：" + thenAcceptFuture.get());
    // 输出：
    // 第一个任务执行
    // 入参校验成功
    // 第二个任务执行，收到：沉梦听雨
    // 返回值：null
}

thenApply

接收上一步的结果作为入参，且有返回值（结果转换）。

@Test
void testThenApply() throws ExecutionException, InterruptedException {
    CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("第一个任务执行");
        return "cmty256";
    });

    CompletableFuture<String> thenApplyFuture = firstFuture.thenApply(result -> {
        if ("沉梦听雨".equals(result)) {
            return "匹配成功";
        }
        return "thenApply-转换后的返回值";
    });

    System.out.println("返回值：" + thenApplyFuture.get());
    // 输出：
    // 第一个任务执行
    // 返回值：thenApply-转换后的返回值
}

whenComplete

任务完成后触发回调（无论成功还是异常都会执行），回调无返回值，返回的 CompletableFuture 的结果沿用上一步。

@Test
void testWhenComplete() throws ExecutionException, InterruptedException {
    CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("线程：" + Thread.currentThread().getName());
        try {
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "沉梦听雨";
    });

    CompletableFuture<String> whenCompleteFuture = firstFuture.whenComplete((result, ex) -> {
        System.out.println("线程：" + Thread.currentThread().getName());
        System.out.println("收到结果：" + result);
        if (ex != null) {
            System.out.println("发生异常：" + ex.getMessage());
        }
    });

    System.out.println("返回值：" + whenCompleteFuture.get());
    // 输出：
    // 线程：ForkJoinPool.commonPool-worker-19
    // 线程：ForkJoinPool.commonPool-worker-19
    // 收到结果：沉梦听雨
    // 返回值：沉梦听雨
}

---

异常处理

异步操作可能失败，CompletableFuture 提供了多种异常处理方式：

方法	说明
handle	同时处理结果和异常，有返回值
exceptionally	仅处理异常，有返回值（提供兜底默认值）
completeExceptionally	手动将 CompletableFuture 标记为异常完成

handle

同时接收结果和异常，可以根据情况返回不同的值：

@Test
void testHandle() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        throw new RuntimeException("Computation error!");
    }).handle((result, ex) -> {
        if (ex != null) {
            System.out.println("异常被捕获：" + ex.getMessage());
            return "world!";  // 异常时返回默认值
        }
        return result;         // 正常时返回原结果
    });

    assertEquals("world!", future.get());
}

exceptionally

仅在发生异常时触发，提供兜底返回值：

@Test
void testExceptionally() throws ExecutionException, InterruptedException {
    CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
        throw new RuntimeException("Computation error!");
    }).exceptionally(ex -> {
        System.out.println(ex.toString());
        return "world!";  // 兜底默认值
    });

    assertEquals("world!", future.get());
}

completeExceptionally

手动将 CompletableFuture 标记为异常完成状态，后续 get() 会抛出异常：

@Test
void testCompleteExceptionally() throws InterruptedException {
    CompletableFuture<String> future = new CompletableFuture<>();
    future.completeExceptionally(new RuntimeException("Calculation failed!"));

    try {
        future.get();
    } catch (ExecutionException e) {
        System.out.println("捕获到异常：" + e.getCause().getMessage());
    }
}

---

三、多个任务组合处理

AND 组合（两个任务都完成）

thenCombine / thenAcceptBoth / runAfterBoth 都表示：两个任务都完成后，才执行第三个任务。

方法	接收两个任务的结果	有返回值
thenCombine	✓	✓
thenAcceptBoth	✓	✗
runAfterBoth	✗	✗

thenCombine 示例：

@Test
void testThenCombineAsync() {
    CompletableFuture<String> firstFuture = CompletableFuture.completedFuture("第一个异步任务");

    ExecutorService executor = Executors.newFixedThreadPool(10);

    CompletableFuture<String> future = CompletableFuture
            .supplyAsync(() -> "第二个异步任务", executor)
            .thenCombineAsync(firstFuture, (s, other) -> {
                System.out.println(s);      // 第二个异步任务
                System.out.println(other);  // 第一个异步任务
                return "两个任务的组合结果";
            }, executor);

    System.out.println(future.join());
    executor.shutdown();
    // 输出：
    // 第二个异步任务
    // 第一个异步任务
    // 两个任务的组合结果
}

allOf（全部完成）

等待所有任务都完成：

@Test
void testAllOf() throws ExecutionException, InterruptedException {
    CompletableFuture<Void> a = CompletableFuture.runAsync(() -> System.out.println("任务A完成"));
    CompletableFuture<Void> b = CompletableFuture.runAsync(() -> System.out.println("任务B完成"));

    CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b)
            .whenComplete((res, ex) -> System.out.println("finish"));

    allOfFuture.get();  // 返回 null
    // 输出（A、B 顺序不定）：
    // 任务A完成
    // 任务B完成
    // finish
}

如果任意一个任务异常，allOf 返回的 CompletableFuture 调用 get() 会抛出异常。

anyOf（任一完成）

任意一个任务完成即继续：

@Test
void testAnyOf() {
    CompletableFuture<Void> a = CompletableFuture.runAsync(() -> {
        try {
            Thread.sleep(3000L);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("任务A完成");
    });

    CompletableFuture<Void> b = CompletableFuture.runAsync(() -> System.out.println("任务B完成"));

    CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b)
            .whenComplete((res, ex) -> System.out.println("finish"));

    anyOfFuture.join();
    // 输出（B 先完成）：
    // 任务B完成
    // finish
}

如果先完成的任务异常，anyOf 返回的 CompletableFuture 调用 get() 会抛出异常。

---

get() 和 join() 的区别

两者都用于阻塞等待并获取结果，区别在于异常处理：

对比项	get()	join()
受检异常	抛出 InterruptedException、ExecutionException	不抛受检异常
异常类型	ExecutionException	CompletionException
适用场景	需要区分中断异常和执行异常	简化代码，链式调用中更方便

// get()：需要 try-catch
try {
    String result = future.get();
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
} catch (ExecutionException e) {
    // 处理执行异常
}

// join()：无需 try-catch（异常封装为 CompletionException）
String result = future.join();

使用建议：

• 需要精细处理中断异常时，用 get()
• 在 Stream、Lambda 等链式调用中，用 join() 更简洁

---

学习参考

• 异步编程利器：CompletableFuture详解 ｜Java 开发实战 - 掘金 (juejin.cn) (opens new window)
• CompletableFuture 详解 | JavaGuide(Java面试 + 学习指南) (opens new window)