如何在Java中使用Completable Future实现异步操作-java教程-PHP中文网

CompletableFuture是Java中实现异步操作的核心工具，通过supplyAsync和runAsync创建有无返回值的异步任务，支持自定义线程池；其链式编程如thenApply、thenAccept、thenCombine等方法可构建清晰的异步流程；相比传统Future的阻塞等待，CompletableFuture提供非阻塞回调、丰富组合能力及exceptionally、handle、whenComplete等灵活异常处理机制；典型应用场景包括微服务编排聚合、并行数据处理、异步事件响应、超时回退等，显著提升系统响应性和资源利用率。

如何在java中使用completable future实现异步操作

在Java中，要实现异步操作，

CompletableFuture

无疑是当前最优雅、功能最强大的工具之一。它彻底改变了我们处理非阻塞任务的方式，让原本复杂的回调地狱变得清晰可控，大大提升了应用的响应性和资源利用率。简单来说，它提供了一种声明式的方式来构建异步任务链，让你能像写同步代码一样思考异步流程。

解决方案

在我看来，理解

CompletableFuture

的核心在于认识到它不仅仅是一个表示异步计算结果的

Future

，更是一个可以被“完成”的

Future

，并且支持丰富的链式操作。这就像是你点了一份外卖，传统

Future

只是给你一个订单号，你得不断打电话问外卖到哪了（阻塞），或者设个闹钟去查（轮询）。而

CompletableFuture

呢，就像外卖小哥在送达前会通知你，或者你可以在下单时就告诉平台，等外卖到了，自动帮你把餐具摆好，甚至把电视打开。

最基本的异步任务创建，我们可以用

supplyAsync

和

runAsync

。

runAsync

适用于不需要返回值的异步任务：

CompletableFuture future = CompletableFuture.runAsync(() -> {
    // 模拟一个耗时操作
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    System.out.println("这是一个没有返回值的异步任务，执行线程：" + Thread.currentThread().getName());
});
// 可以在这里做其他事情，不用等待
System.out.println("主线程继续执行...");
future.join(); // 阻塞等待任务完成，但不推荐在主线程频繁使用
System.out.println("异步任务完成！");

而

supplyAsync

则用于需要返回结果的异步任务：

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return "异步操作的结果，执行线程：" + Thread.currentThread().getName();
});

// 在这里可以继续处理其他业务逻辑
System.out.println("主线程在等待结果的同时，可以做点别的...");

future.thenAccept(result -> {
    // 当异步任务完成时，处理结果
    System.out.println("收到异步结果：" + result + "，处理结果的线程：" + Thread.currentThread().getName());
});
// 为了演示，这里阻塞一下，实际应用中通常不会这样
// future.join();

默认情况下，

supplyAsync

和

runAsync

会使用

ForkJoinPool.commonPool()

来执行任务。如果需要自定义线程池，可以传入

Executor

参数：

立即学习“Java免费学习笔记（深入）”；

ExecutorService customExecutor = Executors.newFixedThreadPool(5);
CompletableFuture customFuture = CompletableFuture.supplyAsync(() -> {
    // ... 耗时操作 ...
    return "使用自定义线程池的结果";
}, customExecutor);
// ...
customExecutor.shutdown();

CompletableFuture

真正的魅力在于它的链式编程。你可以将多个异步操作串联起来，形成一个处理流程：

```
thenApply(Function)
```
：当上一个
```
CompletableFuture
```
完成时，将其结果作为输入，执行一个函数并返回一个新的
```
CompletableFuture
```
。
```
thenAccept(Consumer)
```
：当上一个
```
CompletableFuture
```
完成时，将其结果作为输入，执行一个消费操作，没有返回值。
```
thenRun(Runnable)
```
：当上一个
```
CompletableFuture
```
完成时，执行一个不带输入也不带返回值的操作。
```
thenCompose(Function)
```
：当上一个
```
CompletableFuture
```
完成时，将其结果作为输入，执行一个函数，这个函数返回一个新的
```
CompletableFuture
```
。这对于扁平化嵌套的
```
CompletableFuture
```
非常有用，类似于
```
flatMap
```
。

thenCombine(otherFuture, BiFunction)

：将两个独立的

CompletableFuture

的结果合并处理。

一个简单的链式调用例子：

CompletableFuture fetchUserData = CompletableFuture.supplyAsync(() -> {
    System.out.println("1. 异步获取用户数据...");
    try { Thread.sleep(500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    return "用户ID: 123";
});

CompletableFuture processData = fetchUserData.thenApplyAsync(userId -> {
    System.out.println("2. 处理用户数据：" + userId);
    try { Thread.sleep(700); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    return userId + ", 状态: 活跃";
});

CompletableFuture sendNotification = processData.thenAcceptAsync(processedInfo -> {
    System.out.println("3. 发送通知：" + processedInfo);
    // 模拟发送通知
});

sendNotification.join(); // 等待所有链式操作完成
System.out.println("所有异步操作链完成。");

注意

thenApplyAsync

等带

Async

后缀的方法，它们会在另一个线程中执行回调，而没有

Async

后缀的，可能会在完成上一个任务的同一个线程中执行，也可能在

ForkJoinPool.commonPool()

中执行，这取决于具体情况。我个人倾向于在回调逻辑比较复杂或耗时时使用

Async

版本，以确保非阻塞性。

CompletableFuture 与传统 Future 有何不同？

这真是一个经典的问题，也是我刚接触

CompletableFuture

时最大的困惑。在我看来，传统的

Future

就像一个“只读”的承诺，你拿到了它，就只能等待它完成，然后调用

get()

方法来获取结果。问题是，

get()

方法是阻塞的！这意味着如果你在主线程调用

future.get()

，那么主线程就得傻傻地等着，直到异步任务完成。这不就又变回了同步吗？

更糟糕的是，

Future

缺乏组合能力。如果你有多个

Future

，你想等它们都完成再做点什么，或者等其中一个完成就继续，

Future

本身并没有提供优雅的API。你可能需要手动管理线程池，或者使用

CountDownLatch

、

ExecutorCompletionService

等工具，代码会变得相当复杂和冗长。

CompletableFuture

则完全不同。它是一个“可写”的

Future

，你可以手动完成它（

complete()

或

completeExceptionally()

），也可以通过它提供的丰富的链式API来组合、转换和处理异步操作的结果。它最大的亮点就是非阻塞和声明式。

非阻塞性与回调：
```
CompletableFuture
```
通过回调机制避免了阻塞。当你调用
```
thenApply
```
、
```
thenAccept
```
等方法时，你不是在等待结果，而是在注册一个当结果可用时会被执行的回调函数。这使得你的主线程或调用线程可以继续执行其他任务，大大提高了程序的响应性。
丰富的组合能力：
```
CompletableFuture
```
提供了
```
thenCompose
```
（扁平化
```
Future
```
）、
```
thenCombine
```
（合并两个
```
Future
```
的结果）、
```
allOf
```
（等待所有
```
Future
```
完成）、
```
anyOf
```
（等待任意一个
```
Future
```
完成）等方法，让你可以像搭建乐高积木一样构建复杂的异步流程。这在处理微服务调用、并行数据处理等场景下简直是神器。
异常处理：
```
Future
```
的异常处理非常原始，
```
get()
```
方法会抛出
```
ExecutionException
```
。而
```
CompletableFuture
```
提供了
```
exceptionally
```
、
```
handle
```
、
```
whenComplete
```
等更细粒度的异常处理机制，可以优雅地捕获和恢复异常，甚至在异常发生时提供备用值。
可完成性：
```
CompletableFuture
```
可以在任务未完成时，通过
```
complete()
```
或
```
completeExceptionally()
```
方法手动设置其结果或异常。这在某些特定场景下，比如超时处理或外部事件触发时，非常有用。

在我看来，

CompletableFuture

真正代表了现代Java并发编程的范式转变，从命令式的“等待-获取”转向了声明式的“注册-响应”。

如何处理 CompletableFuture 中的异常？

处理异步操作中的异常，在我看来，是衡量一个并发工具是否成熟的关键标准之一。

CompletableFuture

在这方面做得相当出色，它提供了多种机制来捕获和处理异步任务中可能出现的错误，远比传统

Future

简单粗暴的

ExecutionException

要优雅得多。

主要的异常处理方法有：

造好物

一站式AI造物设计平台

下载

exceptionally(Function)

：这个方法就像一个“备胎”机制。如果之前的

CompletableFuture

链中任何一个环节抛出了异常，

exceptionally

就会被触发，你可以提供一个函数来处理这个异常，并返回一个备用值。这个备用值会作为当前

CompletableFuture

的结果，后续的正常回调（如

thenApply

）会继续执行。

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    System.out.println("开始执行任务...");
    if (Math.random() < 0.5) {
        throw new RuntimeException("随机错误发生！");
    }
    return "任务成功完成";
}).exceptionally(ex -> {
    System.err.println("捕获到异常：" + ex.getMessage());
    return "任务失败，返回默认值"; // 返回备用值
});

System.out.println("最终结果：" + future.join());

这里要注意的是，

exceptionally

接收的是

Throwable

，但通常我们处理的是

Exception

或其子类。它允许你从异常中恢复，让整个异步链继续“正常”地走下去。

handle(BiFunction)

：

handle

方法更加通用，它无论任务是正常完成还是异常完成，都会被调用。它的回调函数接收两个参数：任务的结果和可能发生的异常。如果任务正常完成，

Throwable

参数为

null

；如果任务异常完成，结果参数为

null

。你可以在

handle

中决定是返回正常结果、备用结果，还是继续抛出异常。

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    System.out.println("执行任务...");
    if (Math.random() < 0.5) {
        throw new IllegalStateException("状态异常！");
    }
    return "成功数据";
}).handle((result, ex) -> {
    if (ex != null) {
        System.err.println("Handle中捕获异常：" + ex.getMessage());
        return "从异常中恢复的数据"; // 恢复
    } else {
        System.out.println("Handle中处理正常结果：" + result);
        return result + " (已处理)"; // 处理正常结果
    }
});
System.out.println("最终结果：" + future.join());

handle

的灵活性在于它能统一处理两种情况，我个人在需要根据结果或异常来决定下一步操作时更倾向于使用它。

whenComplete(BiConsumer)

：

whenComplete

方法也无论任务成功或失败都会执行，但它不修改

CompletableFuture

的结果。它主要用于执行一些副作用操作，比如日志记录、资源清理等，而不会影响后续链的执行。如果

whenComplete

内部抛出异常，这个异常会被传递下去，而不是被它自身捕获。

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    System.out.println("任务开始...");
    if (Math.random() < 0.5) {
        throw new ArithmeticException("数学错误！");
    }
    return "计算结果";
}).whenComplete((result, ex) -> {
    if (ex != null) {
        System.err.println("任务失败，记录日志：" + ex.getMessage());
    } else {
        System.out.println("任务成功，记录日志：" + result);
    }
    // 注意：这里不能修改结果或恢复异常
}).exceptionally(ex -> { // 异常仍然会传递到这里
    System.err.println("最终异常恢复：" + ex.getMessage());
    return "备用计算结果";
});
System.out.println("最终状态：" + future.join());

whenComplete

更像是一个观察者，它能让你知道任务的最终状态，但不会干预结果。

最后，值得一提的是

join()

和

get()

方法。

get()

会抛出

ExecutionException

（包装了实际的异常）和

InterruptedException

，而

join()

会直接抛出非受检异常（

CompletionException

，同样包装了实际异常）。在处理异常时，我个人更喜欢使用链式方法，因为它们提供了更细粒度的控制，避免了在

join()

或

get()

处突然中断整个流程。

CompletableFuture 在实际项目中有哪些典型应用场景？

在我的开发经验中，

CompletableFuture

几乎是现代Java应用，尤其是微服务架构中不可或缺的工具。它让原本复杂的多服务协调、并行数据处理变得清晰高效。

微服务编排与聚合： 这是最常见的场景之一。假设你的前端页面需要展示一个用户的完整信息，而这些信息分散在不同的微服务中：用户基本信息服务、订单服务、评论服务、积分服务。如果同步调用这些服务，会非常慢。使用

CompletableFuture

，你可以并行地调用所有这些服务，然后用

thenCombine

或

allOf

等方法将它们的结果聚合起来。

// 假设这些是调用不同微服务的异步方法
CompletableFuture userInfoFuture = userService.getUserInfo(userId);
CompletableFuture> ordersFuture = orderService.getOrders(userId);
CompletableFuture pointsFuture = pointService.getPoints(userId);

// 等待所有服务调用完成，然后聚合结果
CompletableFuture fullProfileFuture = CompletableFuture.allOf(userInfoFuture, ordersFuture, pointsFuture)
    .thenApply(v -> { // v是Void，因为allOf返回Void
        UserInfo userInfo = userInfoFuture.join(); // join在这里是安全的，因为allOf保证它们已完成
        List orders = ordersFuture.join();
        Integer points = pointsFuture.join();
        return new FullUserProfile(userInfo, orders, points);
    })
    .exceptionally(ex -> {
        System.err.println("聚合用户档案失败：" + ex.getMessage());
        // 返回一个部分数据或默认值
        return new FullUserProfile(null, Collections.emptyList(), 0);
    });

FullUserProfile profile = fullProfileFuture.join();
// ... 处理完整的用户档案

这样一来，整个页面的加载时间就取决于最慢的那个服务，而不是所有服务时间之和，极大地提升了用户体验。

并行数据处理与计算： 当需要处理大量数据，并且每个数据块的处理是独立的，或者需要对数据进行多个独立的计算时，

CompletableFuture

可以用来并行化这些任务。例如，你有一个包含百万条记录的文件，需要对每条记录进行复杂的转换。你可以将文件分成多个块，每个块用一个

CompletableFuture

来处理，最后再将结果合并。

List largeDataSet = Arrays.asList("data1", "data2", "data3", "data4"); // 模拟大数据集
List> futures = largeDataSet.stream()
    .map(data -> CompletableFuture.supplyAsync(() -> {
        // 模拟耗时的数据处理
        try { Thread.sleep(new Random().nextInt(1000)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return data.toUpperCase(); // 转换为大写
    }))
    .collect(Collectors.toList());

// 等待所有处理完成
CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

// 获取所有结果
CompletableFuture> allResultsFuture = allOf.thenApply(v ->
    futures.stream()
        .map(CompletableFuture::join)
        .collect(Collectors.toList())
);

List processedData = allResultsFuture.join();
System.out.println("所有数据处理完成：" + processedData);

这比传统的

ExecutorService.submit()

结合

future.get()

循环要简洁和高效得多。

异步事件处理： 在事件驱动的架构中，当一个事件发生时，可能需要触发多个不相关的异步操作，比如发送邮件、更新缓存、记录日志等。

public void onUserRegistered(User user) {
    CompletableFuture.runAsync(() -> emailService.sendWelcomeEmail(user))
        .exceptionally(ex -> {
            System.err.println("发送欢迎邮件失败：" + ex.getMessage());
            return null;
        });

    CompletableFuture.runAsync(() -> cacheService.updateUserCache(user))
        .exceptionally(ex -> {
            System.err.println("更新用户缓存失败：" + ex.getMessage());
            return null;
        });

    CompletableFuture.runAsync(() -> auditLogService.logUserRegistration(user))
        .exceptionally(ex -> {
            System.err.println("记录注册日志失败：" + ex.getMessage());
            return null;
        });

    System.out.println("用户注册事件已处理，异步任务已启动。");
}

主业务流程不会因为这些副作用操作的耗时而阻塞。

超时控制与回退机制：

CompletableFuture

配合

CompletableFuture.anyOf

和

CompletableFuture.runAfter

可以实现优雅的超时和回退。

CompletableFuture primaryServiceCall = CompletableFuture.supplyAsync(() -> {
    try { Thread.sleep(3000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    return "主服务结果";
});

// 模拟一个超时Future
CompletableFuture timeout = CompletableFuture.supplyAsync(() -> {
    try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    return "超时！";
});

// 哪个先完成就用哪个结果
CompletableFuture