全部标签
写点什么
OpenHarmony算法元宇宙MySQL移动开发学习方法学习方法Web3.0高效工作数据库Python音视频前端AI大数据团队管理程序员运维深度思考低代码redisgolang微服务架构flutter 查看更多

异步编程利器：CompletableFuture 深度解析

作者：码农BookSea
  • 2024-01-11
    浙江

  • 本文字数：9874 字

    阅读完需：约 32 分钟

异步编程利器：CompletableFuture深度解析

本文已收录至 Github，推荐阅读 👉 Java随想录

微信公众号：Java随想录


摘要

在异步编程中，我们经常需要处理各种异步任务和操作。Java 8 引入的 CompletableFuture 类为我们提供了一种强大而灵活的方式来处理异步编程需求。CompletableFuture 类提供了丰富的方法和功能，能够简化异步任务的处理和组合。


本文将深入解析 CompletableFuture，希望对各位读者能有所帮助。


CompletableFuture 适用于以下场景


  • 并发执行多个异步任务，等待它们全部完成或获取其中任意一个的结果。

  • 对已有的异步任务进行进一步的转换、组合和操作。

  • 异步任务之间存在依赖关系，需要按照一定的顺序进行串行执行。

  • 需要对异步任务的结果进行异常处理、超时控制或取消操作。

如何使用

下面是一个演示 CompletableFuture 如何使用的代码示例：


public class CompletableFutureExample {
    public static void main(String[] args) {        // 创建CompletableFuture对象，并定义异步任务        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {            // 异步任务的逻辑代码            // 在这里执行耗时操作或其他需要异步执行的任务            try {                TimeUnit.SECONDS.sleep(2); // 模拟耗时操作            } catch (InterruptedException e) {                e.printStackTrace();            }            return "Hello, ";        });
        // 添加任务完成后的回调方法        CompletableFuture<String> resultFuture = future.thenApplyAsync(result -> {            // 任务完成后的处理逻辑            // result为上一步任务的结果            return result + "World!";        });
        // 组合多个CompletableFuture对象        CompletableFuture<String> combinedFuture = future.thenCombine(resultFuture, (result1, result2) -> {            // 对多个CompletableFuture的结果进行组合处理            return result1 + result2 + " Welcome to the CompletableFuture world!";        });
        // 异常处理        CompletableFuture<String> exceptionHandledFuture = combinedFuture.exceptionally(ex -> {            // 异常处理逻辑            System.out.println("任务执行出现异常：" + ex.getMessage());            return "Fallback Result";        });
        // 等待并获取任务的结果        try {            String result = exceptionHandledFuture.get();            System.out.println("任务的最终结果为：" + result);        } catch (InterruptedException | ExecutionException e) {            // 处理异常情况            e.printStackTrace();        }    }}
复制代码


结果输出：


任务的最终结果为：Hello, Hello, World! Welcome to the CompletableFuture world!
复制代码


首先，我们创建了一个CompletableFuture对象future。在future中，我们使用supplyAsync方法定义了一个异步任务，其中 lambda 表达式 中的代码会在另一个线程中执行。在这个例子中，我们模拟了一个耗时操作，通过TimeUnit.SECONDS.sleep(2)暂停了 2 秒钟。


然后，我们添加了一个回调方法resultFuture。在这个回调方法中，将前一个异步任务的结果作为参数进行处理，并返回处理后的新结果。在这个例子中，我们将前一个任务的结果与字符串 "World!" 连接起来，形成新的结果。


接下来，我们使用thenCombine方法组合了两个CompletableFuture对象：future和resultFuture。在这个组合任务中，我们将两个任务的结果进行组合处理，返回最终的结果。在这个例子中，我们将前两个任务的结果与字符串 " Welcome to the CompletableFuture world!" 连接起来。


此外，我们还处理了异常情况。通过exceptionally方法，我们定义了一个异常处理回调方法。如果在任务执行过程中发生了异常，我们可以在这里对异常进行处理，并返回一个默认值作为结果。


最后，我们使用get方法等待并获取最终的任务结果。需要注意的是，get方法可能会阻塞当前线程，直到任务完成并返回结果。在这个例子中，我们使用try-catch块捕获可能的异常情况，并打印出最终的任务结果。


这个例子只是部分展示了CompletableFuture的功能，实际上它比你想象的还要强大！

源码解析

CompletableFuture 的源码非常庞大和复杂，涉及到并发、线程池、同步机制等多方面的知识。在这里，我们只重点介绍 CompletableFuture 的核心实现原理。

基本结构


CompletableFuture 的作者是大名鼎鼎的 Doug Lea。CompletableFuture 类是实现了 Future 和 CompletionStage 接口的一个关键类。它可以表示异步计算的结果，并提供了一系列方法来操作和处理这些结果。


CompletableFuture 内部使用了一个属性result来保存计算结果，以及若干个属性waiters来保存等待结果的任务。当计算完成后，CompletableFuture 将会通知所有等待结果的任务，并将结果传递给它们。


为了实现链式操作，CompletableFuture 还定义了内部类：Completion, UniCompletion, 和 BiCompletion。


Completion, UniCompletion, 和 BiCompletion 是 CompletableFuture 内部用于处理异步任务完成的辅助类。


  • Completion 是一个通用的辅助类，它包含了任务完成后的回调方法，以及处理异常的方法。

  • UniCompletion 是 Completion 的子类，是一元依赖的基类，用于处理单个任务的完成情况，并提供了更多的方法来处理结果和异常。

  • BiCompletion 是 UniCompletion 的子类，是二元依赖的基类，同时也是多元依赖的基类，用于处理两个任务的完成情况，并提供了更多的方法来组合和处理这两个任务的结果和异常。


这些辅助类在 CompletableFuture 的内部被使用，以实现异步任务的执行、结果的处理和组合等操作。它们提供了一种灵活的方式来处理异步任务的完成情况，并通过回调方法或其他一些方法来处理任务的结果和异常。

内部原理


CompletableFuture 中包含两个字段：result 和 stack。result 用于存储当前 CF 的结果，stack (Completion）表示当前 CF 完成后需要触发的依赖动作(Dependency Actions)，去触发依赖它的 CF 的计算，依赖动作可以有多个(表示有多个依赖它的 CF)，以栈（Treiber stack）的形式存储，stack 表示栈顶元素。


CompletableFuture 在设计思想上类似 “观察者模式，每个 CompletableFuture 都可以被看作一个被观察者，其内部有一个 Completion 类型的链表成员变量 stack，用来存储注册到其中的所有观察者。当被观察者执行完成后会弹栈 stack 属性，依次通知注册到其中的观察者。

执行流程

CompletableFuture 的执行流程如下：


  1. 创建 CompletableFuture 对象：通过调用CompletableFuture类的构造方法或静态工厂方法创建一个新的 CompletableFuture 对象。

  2. 定义异步任务：使用supplyAsync()、runAsync()等方法定义需要在后台线程中执行的异步任务，这些方法接受一个 lambda 表达式 或 Supplier/Runnable 接口作为参数。

  3. 启动异步任务：一旦 CompletableFuture 对象创建并定义了异步任务，任务会立即在后台线程中开始执行，并返回一个代表异步计算结果的 CompletableFuture 对象。

  4. 异步任务执行过程：

  5. 当异步任务完成时，它会设置自己的结果值，将状态标记为已完成。

  6. 如果有其他线程在此之前调用了complete()、completeExceptionally()、cancel()等方法，可能会影响任务的最终状态。

  7. 注册回调方法：

  8. 使用thenApply(), thenAccept(), thenRun()等方法来注册回调函数，当异步任务完成或异常时，这些回调函数会被触发。

  9. 回调函数也可以是异步的，通过thenApplyAsync(), thenAcceptAsync(), thenRunAsync()等方法注册。

  10. 组合多个 CompletableFuture：

  11. 使用thenCompose(), thenCombine(), allOf(), anyOf()等方法，可以将多个 CompletableFuture 对象进行组合，形成更复杂的异步任务处理流程。

  12. 处理异常：

  13. 通过使用exceptionally(), handle(), whenComplete()等方法，可以注册异常处理函数，当异步任务出现异常时，这些处理函数会被触发。

  14. 等待结果：

  15. 使用get()或join()方法来阻塞当前线程，并等待 CompletableFuture 对象的完成并获取最终的结果。

  16. get()方法会抛出可能的异常（InterruptedException, ExecutionException）。

  17. join()方法与get()类似，但不会抛出 checked 异常。

  18. 取消任务：通过调用 CompletableFuture 对象的cancel()方法取消异步任务的执行。


请注意，以上步骤的顺序和具体实现可能略有不同，但大致上反映了 CompletableFuture 的执行流程。在实际应用中，我们可以根据需求选择适合的方法来处理异步任务的完成情况、结果、异常以及任务之间的关系。

方法介绍

CompletableFuture 类提供了一系列用于处理和组合异步任务的方法。以下是这些方法的介绍：

创建对象

创建一个 CompletableFuture 对象有以下几种方法：


  • 使用 CompletableFuture 的构造方法


CompletableFuture<String> future = new CompletableFuture<>();
复制代码


  • 使用 CompletableFuture 的静态工厂方法


CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {    // 异步任务逻辑    return "Result";});
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {    // 异步任务逻辑});
复制代码


  • 使用转换方法


CompletableFuture<Integer> transformedFuture = originalFuture.thenApply(result -> {    // 转换逻辑    return result.length();});
originalFuture.thenAccept(result -> {    // 处理结果逻辑    System.out.println("Result: " + result);});
CompletableFuture<Void> runnableFuture = originalFuture.thenRun(() -> {    // 在结果完成后执行的操作});
复制代码


  • 直接创建一个已完成状态的 CompletableFuture


//CompletableFuture.completedFuture()直接创建一个已完成状态的CompletableFutureCompletableFuture<String> cf2 = CompletableFuture.completedFuture("result");
//先初始化一个未完成的CompletableFuture，然后通过complete()、completeExceptionally()，也完成该CompletableFutureCompletableFuture<String> cf = new CompletableFuture<>();cf.complete("success");
复制代码


  • toCompletableFuture


CompletionStage<Integer> stage = CompletableFuture.supplyAsync(() -> 42);
CompletableFuture<Integer> future = stage.toCompletableFuture();
复制代码


用于将当前的 CompletionStage 对象转换为一个 CompletableFuture 对象。

异步执行任务

以下是在 CompletableFuture 对象上异步执行任务的一些方法示例：


  • supplyAsync(Supplier<U> supplier)：异步执行一个有返回值的供应商（Supplier）任务。


CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {    // 异步任务逻辑    return "Result";});
复制代码


  • runAsync(Runnable runnable)：异步执行一个没有返回值的任务。


CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {    // 异步任务逻辑});
复制代码

链式操作

CompletableFuture 提供了不同的方式来对异步任务进行链式操作。


  • thenRun


CompletableFuture<Void> executedFuture = future.thenRun(() -> executeTask());
复制代码


thenRun方法用于在 CompletableFuture 完成后执行一个 Runnable 任务。它返回一个新的 CompletableFuture 对象，该对象没有返回值。


  • thenAccept


CompletableFuture<Void> acceptedFuture = future.thenAccept(result -> processResult(result));
复制代码


thenAccept方法用于在 CompletableFuture 完成后对结果进行处理。它接收一个 Consumer 函数作为参数，并返回一个新的 CompletableFuture 对象。


  • thenApply


CompletableFuture<U> appliedFuture = future.thenApply(result -> transformResult(result));
复制代码


thenApply方法用于在 CompletableFuture 完成后对结果进行转换。它接收一个 Function 函数作为参数，并返回一个新的 CompletableFuture 对象。


  • thenCompose


CompletableFuture<U> composedFuture = future.thenCompose(result -> executeAnotherTask(result));
复制代码


用于对异步任务的结果进行处理，并返回一个新的异步任务。


  • whenComplete


CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 42);
CompletableFuture<Void> whenCompleteFuture = future.whenComplete((result, exception) -> {    if (exception != null) {        System.out.println("Exception occurred: " + exception.getMessage());    } else {        System.out.println("Result: " + result);    }});
whenCompleteFuture.join();
复制代码


用于在异步任务完成后执行指定的动作。它允许你在任务完成时处理结果或处理异常。


  • thenCompose() 用于对异步任务的结果进行处理，并返回一个新的异步任务。它接受一个函数式接口参数，根据原始任务的结果创建并返回一个新的 CompletionStage 对象。

  • whenComplete() 用于在异步任务完成后执行指定的动作。它接受一个消费者函数式接口参数，用于处理任务的结果或异常，但没有返回值。

异步任务组合

CompletableFuture 还提供了一系列方法来组合和处理多个异步任务的结果。


  • allOf


CompletableFuture<Void> allFuture = CompletableFuture.allOf(future1, future2, future3);
复制代码


allOf方法接收一组 CompletableFuture 对象作为参数，并返回一个新的 CompletableFuture 对象，该对象在所有给定的 CompletableFuture 都完成时完成。这样我们可以等待所有任务都完成后再进行下一步操作。


  • anyOf


CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2, future3);
复制代码


anyOf方法与allOf类似，不同之处在于它返回的 CompletableFuture 对象在任何一个给定的 CompletableFuture 完成时就完成。这样我们可以获取最先完成的任务的结果。


  • thenCombine


CompletableFuture<U> combinedFuture = future1.thenCombine(future2, (result1, result2) -> combineResults(result1, result2));
复制代码


thenCombine方法接收两个 CompletableFuture 对象和一个函数作为参数，用于指定当这两个 CompletableFuture 都完成时如何处理它们的结果。返回的新的 CompletableFuture 对象将接收到计算后的结果。


  • applyToEither


CompletableFuture<U> resultFuture = future1.applyToEither(future2, result -> processResult(result));
复制代码


applyToEither方法用于获取两个 CompletableFuture 中任意一个完成的结果，并对该结果进行处理。它接收一个 Function 函数作为参数，并返回一个新的 CompletableFuture 对象。


  • acceptEither


CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 20);
future1.acceptEither(future2, result -> {    System.out.println("Result: " + result);});
复制代码


用于在两个 CompletableFuture 对象中任意一个完成时执行指定的操作。该方法接收两个参数：另一个 CompletableFuture 对象和一个消费者函数（Consumer）。当其中任何一个 CompletableFuture 完成时，将其结果作为参数传递给消费者函数进行处理。


  • runAfterBoth


CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 42);CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<Void> combinedFuture = future1.runAfterBoth(future2, () -> {    System.out.println("Both futures completed");});
combinedFuture.join();
复制代码


用于在两个异步任务都完成后执行指定的动作，需要注意的是，runAfterBoth() 方法是一个非阻塞方法，动作将在两个异步任务都完成后立即执行。


  • runAfterEither


CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {    try {        Thread.sleep(500);    } catch (InterruptedException e) {        e.printStackTrace();    }    return 42;});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {    try {        Thread.sleep(1000);    } catch (InterruptedException e) {        e.printStackTrace();    }    return "Hello";});
CompletableFuture<Void> eitherFuture = future1.runAfterEither(future2, () -> {    System.out.println("One of the futures completed");});
eitherFuture.join();
复制代码


用于在两个异步任务中任意一个完成后执行指定的动作。


  • thenAcceptBoth


CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 42);CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<Void> thenAcceptBothFuture = future1.thenAcceptBoth(future2, (result1, result2) -> {    System.out.println("Action executed with thenAcceptBoth(): " + result1 + ", " + result2);});
thenAcceptBothFuture.join();
复制代码


用于在两个异步任务都完成后执行指定的动作。它的作用是接收两个异步任务的结果，并将结果作为参数传递给指定的消费者函数。

异常处理

CompletableFuture 提供了多种方式来处理异步任务的异常情况。


  • exceptionally


CompletableFuture<U> exceptionHandledFuture = future.exceptionally(ex -> handleException(ex));
复制代码


通过exceptionally方法，我们可以对 CompletableFuture 的异常情况进行处理。它接收一个 Function 函数作为参数，用于处理异常并返回一个新的 CompletableFuture 对象。


  • handle


CompletableFuture<U> handledFuture = future.handle((result, ex) -> handleResult(result, ex));
复制代码


handle方法可以同时处理正常结果和异常情况。它接收一个 BiFunction 函数作为参数，用于处理结果和异常，并返回一个新的 CompletableFuture 对象。


  • completeExceptionally


future.completeExceptionally();
复制代码


异常地完成 CompletableFuture，将结果设置为一个异常。


  • isCompletedExceptionally


CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {    throw new RuntimeException("Something went wrong");});
boolean completedExceptionally = future.isCompletedExceptionally();System.out.println("Is completed exceptionally: " + completedExceptionally);
复制代码


该方法返回一个布尔值，表示当前异步任务是否已经异常完成。


  • obtrudeException


CompletableFuture<Integer> future = new CompletableFuture<>();
future.obtrudeException(new RuntimeException("Something went wrong"));
boolean completedExceptionally = future.isCompletedExceptionally();System.out.println("Is completed exceptionally: " + completedExceptionally);
复制代码


用于强制将指定的异常作为异步任务的结果，调用 obtrudeException(Throwable ex) 方法后，异步任务将立即完成，并将指定的异常作为结果返回。

取值与状态

  • join


future.join()
复制代码


join() 方法不会抛出已检查异常，因为它是基于 CompletableFuture 类设计的，如果异步任务抛出异常，join() 方法会将该异常包装在 CompletionException 中并抛出。


  • get


future.get()
复制代码


get() 方法会抛出一个 InterruptedException 异常和一个 ExecutionException 异常，前者表示获取结果时被中断，后者表示获取结果时任务本身抛出了异常。


future.get(1,TimeUnit.Hours)
复制代码


有异常则抛出异常，最长等待一个小时，一个小时之后，如果还没有数据，则异常。


  • getNow


CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {    // 异步任务逻辑    return 42;});
int result = future.getNow(0); // 获取异步操作的结果，如果尚未完成，则返回默认值0
System.out.println("Result: " + result);
复制代码


getNow(T value) 是 CompletableFuture 类的一个方法，用于获取异步操作的结果，如果异步操作尚未完成，则返回给定的默认值，该方法会立即返回结果，不会阻塞当前线程。

超时控制与取消操作

CompletableFuture 也支持超时控制和取消操作，以便更好地管理异步任务的执行。


  • completeOnTimeout


CompletableFuture<U> timeoutFuture = future.completeOnTimeout(defaultResult, timeout, timeUnit);
复制代码


completeOnTimeout方法在指定的超时时间内等待 CompletableFuture 的完成，如果超时则将其设置为默认结果。它返回一个新的 CompletableFuture 对象。


  • cancel


boolean isCancelled = future.cancel(true);
复制代码


cancel方法可用于取消 CompletableFuture 的执行。它接收一个 boolean 参数，指示是否中断正在执行的任务。返回值表示是否成功取消了任务。


  • isCancelled


CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {    // 异步任务逻辑    return 42;});
future.cancel(true); // 取消异步任务
boolean isCancelled = future.isCancelled();System.out.println("Is cancelled: " + isCancelled);
复制代码


isCancelled() 是 CompletableFuture 类的一个方法，用于判断当前异步任务是否已被取消。如果异步任务已被取消，则返回 true；否则返回 false。

依赖

  • getNumberOfDependents


CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 20);
CompletableFuture<Integer> combinedFuture = future1.thenCombine(future2, (result1, result2) -> result1 + result2);
int numberOfDependents = combinedFuture.getNumberOfDependents();System.out.println("Number of dependents: " + numberOfDependents);
复制代码


getNumberOfDependents() 用于获取当前 CompletableFuture 对象所依赖的其他异步任务的数量。如果没有任何依赖任务，或者所有依赖任务已经完成，则返回的数量为 0。

完成

  • complete


future.complete("米饭");
复制代码


complete(T value)：该方法返回布尔值，表示是否成功地将结果设置到 CompletableFuture 中。如果 CompletableFuture 未完成，则将结果设置，并返回 true；如果 CompletableFuture 已经完成，则不进行任何操作并返回 false。


  • obtrudeValue


CompletableFuture<Integer> future = new CompletableFuture<>();
future.obtrudeValue(42);
boolean completedNormally = future.isDone() && !future.isCompletedExceptionally();System.out.println("Is completed normally: " + completedNormally);
复制代码


用于强制将指定的值作为异步任务的结果，调用 obtrudeValue(T value) 方法后，异步任务将立即完成，并将指定的值作为结果返回。


与 complete() 不同，obtrudeValue() 必须在任务已经完成的情况下调用，否则会引发 IllegalStateException 异常。并且complete() 方法对于已经完成的任务会忽略额外的完成操作，并返回 false。而obtrudeValue() 方法即使任务已经完成，仍然会强制使用新的结果值，并返回 true。


  • isDone


CompletableFuture<Integer> future = CompletableFuture.completedFuture(42);
boolean done = future.isDone();System.out.println("Is done: " + done);
复制代码


用于判断当前异步任务是否已经完成（无论是正常完成还是异常完成）。

并发限制

CompletableFuture 也支持并发限制，以控制同时执行的异步任务数量。


Executor executor = Executors.newFixedThreadPool(10);CompletableFuture<U> future = CompletableFuture.supplyAsync(() -> doSomething(), executor);
复制代码


我们可以通过使用线程池来限制 CompletableFuture 的并发执行数量。通过创建一个固定大小的线程池，并将其作为参数传递给 CompletableFuture，就可以控制并发执行任务的数量。

记忆窍门

CompletableFuture 类提供了许多方法，但实际上常用的方法只有几个。为了方便记忆，以下是一些总结的规律：


  • 方法名带 Async 的都是异步方法，对应的没有 Async 则是同步方法，比如 thenAccept 与 thenAcceptAsync 。

  • 方法名带 run 的入参为 Runnable，且无返回值。

  • 方法名带 supply 的入参为 Supplier，且有返回值。

  • 方法名带 Accept 的入参为 Consumer，且无返回值。

  • 方法名带 Apply 的入参为 Function，且有返回值。

  • 方法名带 Either 的方法表示谁先完成就消费谁。

  • 方法名带 Both 的方法表示两个任务都完成才消费。


掌握以上规律后，就可以基本记住大部分方法，剩下的其他方法可以单独记忆。

总结

本文详细探讨了 CompletableFuture 的原理和方法，学习了如何在任务完成后执行操作、处理结果和转换结果。


CompletableFuture 是 Java 中强大的异步编程工具之一，合理利用它的方法和策略可以更好地处理异步任务和操作。


希望本文对读者有所启发和帮助。

划线
评论
复制
发布于: 刚刚阅读数: 8

版权声明: 本文为 InfoQ 作者【码农BookSea】的原创文章。

原文链接:【http://xie.infoq.cn/article/ed4def6d12da202fe0eb0a762】。文章转载请联系作者。

Java多线程并发
用户头像

码农BookSea

关注

Java开发工程师 2021-12-26 加入

Java开发菜鸟工程师，写博客的初衷是为了沉淀我所学习，累积我所见闻，分享我所体验。希望和更多的人交流学习。

评论

发布
暂无评论
Copyright © 2024, Geekbang Technology Ltd. All rights reserved. 极客邦控股（北京）有限公司 | 京 ICP 备 16027448 号 - 5京公网安备京公网安备 11010502039052号 | 产品资质
异步编程利器：CompletableFuture深度解析_Java_码农BookSea_InfoQ写作社区