深入理解 Java 8 中的 CompletableFuture

时间：2023-02-16 11:03:09

CompletableFuture 是 Java 8 中引入的一个新的并发编程工具，它为开发者提供了一种简单、高效的方式来处理异步操作和并发任务。CompletableFuture 可以看作是 Future 的增强版，它提供了更丰富的功能和更方便的使用方式。

在本篇教程中，我们将学习 CompletableFuture 的原理、应用和示例代码，以便让你更好地了解和掌握 CompletableFuture 的使用。

一、CompletableFuture 原理

1.1 CompletableFuture 的基本概念

在介绍 CompletableFuture 的原理之前，我们先来了解一些 CompletableFuture 的基本概念。

CompletableFuture 是一个实现了 Future 接口的类，它可以表示一个异步计算任务的结果。CompletableFuture 可以以同步或异步的方式完成计算，并且支持链式调用、组合和合并多个 CompletableFuture。

在 CompletableFuture 中，我们可以通过 thenApply、thenAccept、thenRun、thenCompose 等方法来串联多个 CompletableFuture，以实现多个异步任务之间的依赖关系。而且，CompletableFuture 还提供了 thenCombine、thenAcceptBoth、runAfterBoth 等方法，以方便实现两个异步任务之间的协作。

CompletableFuture 可以通过 complete、completeExceptionally、cancel 等方法来手动完成计算任务，也可以通过 get、join、getNow、getTimeout 等方法来获取计算结果。

1.2 CompletableFuture 的执行模型

CompletableFuture 的执行模型主要分为两种：同步执行和异步执行。

  • 同步执行：同步执行指的是当前线程等待 CompletableFuture 的计算结果。如果 CompletableFuture 的计算已经完成，那么当前线程将立即继续执行下去。如果 CompletableFuture 的计算还没有完成，那么当前线程将被阻塞，直到计算完成。
  • 异步执行：异步执行指的是当前线程不需要等待 CompletableFuture 的计算结果，而是继续执行下去。当 CompletableFuture 的计算完成后，会通知调用者线程，以便让它处理计算结果。

CompletableFuture 的异步执行主要是通过 ForkJoinPool 实现的。ForkJoinPool 是 Java 7 中新增的一个线程池，它是一种工作窃取模型的线程池，可以将任务分割成多个子任务并行执行。在 CompletableFuture 中，如果没有显式指定 Executor，那么 CompletableFuture 将会默认使用 ForkJoinPool 实现异步执行。

1.3 CompletableFuture 的状态转换

CompletableFuture 有四种状态：未完成、已完成、异常完成和已取消。当 CompletableFuture 的状态发生变化时，会触发 CompletionStage 的相关回调方法，以便让调用者线程处理计算结果。

下面是 CompletableFuture 的四种状态：

  • 未完成：CompletableFuture 的计算还没有完成，此时 CompletableFuture 处于未完成状态。
  • 已完成：CompletableFuture 的计算已经完成，且成功返回了结果。此时 CompletableFuture 处于已完成状态。
  • 异常完成：CompletableFuture 的计算已经完成，但是出现了异常。此时 CompletableFuture 处于异常完成状态。
  • 已取消：CompletableFuture 的计算被取消了。此时 CompletableFuture 处于已取消状态。

CompletableFuture 的状态转换可以通过 complete、completeExceptionally、cancel 等方法来触发。下面是 CompletableFuture 的状态转换图：

         +-------------------------+
|          NEW            |
+-------------------------+
|
                    v
+-------------------------+
|        COMPLETING       |
+-------------------------+
/                   \
/                     \
/                       \
/                         \
/                           \
      v                             v
+-------------------------+  +-------------------------+
|       NORMAL VALUE      |  |       EXCEPTIONAL        |
+-------------------------+  +-------------------------+

在状态转换图中，我们可以看到 CompletableFuture 有两种中间状态：NEW 和 COMPLETING。当我们创建一个 CompletableFuture 对象时，它的状态为 NEW。当我们调用 CompletableFuture 的计算方法，且计算还没有完成时，CompletableFuture 的状态会从 NEW 转换为 COMPLETING。在 COMPLETING 状态下，CompletableFuture 的结果还没有确定，可能是正常的计算结果，也可能是异常结果。

当 CompletableFuture 的计算完成后，它的状态会从 COMPLETING 转换为 NORMAL VALUE 或 EXCEPTIONAL。如果计算结果是正常的，那么 CompletableFuture 的状态会变为 NORMAL VALUE。如果计算结果是异常的，那么 CompletableFuture 的状态会变为 EXCEPTIONAL。

1.4 CompletableFuture 的异常处理

在 CompletableFuture 中，我们可以通过 handle、exceptionally、whenComplete、whenCompleteAsync 等方法来处理异常。

  • handle：可以在计算完成时处理计算结果或异常情况。
  • exceptionally：可以在计算出现异常时处理异常情况。
  • whenComplete 和 whenCompleteAsync：可以在计算完成时处理计算结果或异常情况，且不会阻塞当前线程。

下面是一个使用 handle 方法处理异常的示例：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
// 模拟一个会抛出异常的计算过程
throw new RuntimeException("计算过程中出现异常");
});

CompletableFuture<String> handleFuture = future.handle((result, exception) -> {
if (exception != null) {
return "计算出现异常：" + exception.getMessage();
    } else {
return "计算结果：" + result;
    }
});

System.out.println(handleFuture.get()); // 输出："计算出现异常：计算过程中出现异常"

在这个示例中，我们创建了一个 CompletableFuture 对象 future，它模拟了一个会抛出异常的计算过程。然后，我们通过 handle 方法对 future 的计算结果或异常情况进行处理，以便生成一个新的 CompletableFuture 对象 handleFuture。

如果 future 的计算过程中出现了异常，那么 handleFuture 的计算结果将会是一个包含异常信息的字符串。否则，handleFuture 的计算结果将会是一个包含计算结果的字符串。

二、CompletableFuture 的应用

2.1 异步调用

CompletableFuture 最常见的用途之一是进行异步调用。我们可以使用 CompletableFuture.supplyAsync、CompletableFuture.runAsync 等方法来异步执行一个计算过程，并在计算完成后获取计算结果或进行一些后续处理。

下面是一个使用 CompletableFuture.supplyAsync 方法异步调用一个计算过程的示例：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
// 模拟一个耗时的计算过程
Thread.sleep(1000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
return "Hello, CompletableFuture!";
});

// 等待计算完成，并获取计算结果
String result = future.get();

System.out.println(result); // 输出："Hello, CompletableFuture!"

在这个示例中，我们使用 CompletableFuture.supplyAsync 方法异步执行了一个计算过程，该计算过程模拟了一个耗时的操作，然后返回了一个字符串。我们使用 future.get() 方法来等待计算完成，并获取计算结果。

2.2 组合计算

CompletableFuture 另一个非常强大的用途是组合计算。我们可以使用 thenCompose、thenApply、thenAccept、thenRun、thenCombine、thenAcceptBoth 等方法将多个 CompletableFuture 对象组合在一起进行计算，并在计算完成后获取计算结果或进行一些后续处理。

  • thenCompose 和 thenApply：可以将两个 CompletableFuture 对象组合在一起进行计算，然后返回一个新的 CompletableFuture 对象。
  • thenAccept 和 thenRun：可以在 CompletableFuture 对象计算完成后执行一个任务，且不会返回任何结果。
  • thenCombine 和 thenAcceptBoth：可以将两个 CompletableFuture 对象的计算结果组合在一起进行计算，然后返回一个新的 CompletableFuture 对象或执行一个任务。

下面是一个使用 thenCompose 和 thenApply 方法组合计算的示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
try {
// 模拟一个耗时的计算过程
Thread.sleep(1000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
return "Hello,";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
try {
// 模拟一个耗时的计算过程
Thread.sleep(1000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
return "CompletableFuture!";
});

CompletableFuture<String> future3 = future1.thenCompose(result1 -> {
return future2.thenApply(result2 -> {
return result1 + " " + result2;
    });
});

// 等待计算完成，并获取计算结果
String result = future3.get();

System.out.println(result); // 输出："Hello, CompletableFuture!"

在这个示例中，我们使用 CompletableFuture.supplyAsync 方法异步执行了两个计算过程，分别返回了字符串 "Hello," 和 "CompletableFuture!"。然后，我们使用 thenCompose 和 thenApply 方法将这两个 CompletableFuture 对象组合在一起进行计算，并生成一个新的 CompletableFuture 对象 future3。在这个计算过程中，我们将两个字符串拼接在一起，并返回一个新的字符串。最后，我们使用 future3.get() 方法等待计算完成，并获取计算结果。

2.3 异常处理

CompletableFuture 也支持对计算过程中出现的异常进行处理。我们可以使用 exceptionally、handle、whenComplete、whenCompleteAsync 等方法来处理异常，并进行相应的后续处理。

下面是一个使用 exceptionally 方法处理异常的示例：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
// 抛出一个异常
throw new RuntimeException("计算出现异常！");
});

CompletableFuture<String> exceptionFuture = future.exceptionally(ex -> {
// 处理异常并返回一个默认值
return "计算出现异常，返回默认值。";
});

// 等待计算完成，并获取计算结果
String result = exceptionFuture.get();

System.out.println(result); // 输出："计算出现异常，返回默认值。"

在这个示例中，我们使用 CompletableFuture.supplyAsync 方法异步执行了一个计算过程，并抛出了一个 RuntimeException 异常。然后，我们使用 exceptionally 方法来处理这个异常，并返回一个默认值。最后，我们使用 exceptionFuture.get() 方法等待计算完成，并获取计算结果。

2.4 组合多个异步操作

CompletableFuture 也支持将多个异步操作组合在一起进行计算。我们可以使用 thenCombine、thenAcceptBoth、runAfterBoth 等方法将多个 CompletableFuture 对象组合在一起进行计算，然后返回一个新的 CompletableFuture 对象或执行一个任务。

下面是一个使用 thenCombine 方法组合多个异步操作的示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
try {
// 模拟一个耗时的计算过程
Thread.sleep(1000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
return 100;
});

CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
try {
// 模拟一个耗时的计算过程
Thread.sleep(1000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
return 200;
});

CompletableFuture<Integer> future3 = future1.thenCombine(future2, (result1, result2) -> {
return result1 + result2;
});

// 等待计算完成，并获取计算结果
Integer result = future3.get();

System.out.println(result); // 输出：300

在这个示例中，我们使用 CompletableFuture.supplyAsync 方法异步执行了两个计算过程，分别返回了整数 100 和 200。然后，我们使用 thenCombine 方法将这两个 CompletableFuture 对象组合在一起进行计算，并生成一个新的 CompletableFuture 对象 future3。在这个计算过程中，我们将两个整数相加，并返回一个新的整数。最后，我们使用 future3.get() 方法等待计算完成，并获取计算结果。

2.5 其他方法

除了上述方法外，CompletableFuture 还提供了很多其他有用的方法，例如：allOf、anyOf、delayedExecutor、completedFuture 等。

allOf：当所有的 CompletableFuture 对象都计算完成后，返回一个新的CompletableFuture 对象，这个对象的计算结果是一个 Void 类型的值。

下面是一个使用 allOf 方法组合多个异步操作的示例：

CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
System.out.println("任务 1 开始执行...");
try {
// 模拟一个耗时的计算过程
Thread.sleep(1000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
System.out.println("任务 1 执行完成。");
});

CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
System.out.println("任务 2 开始执行...");
try {
// 模拟一个耗时的计算过程
Thread.sleep(2000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
System.out.println("任务 2 执行完成。");
});

CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2);

// 等待所有的计算完成
allFutures.get();

System.out.println("所有的任务都已经执行完成。");

在这个示例中，我们使用 CompletableFuture.runAsync 方法异步执行了两个任务。然后，我们使用 allOf 方法将这两个 CompletableFuture 对象组合在一起，并生成一个新的 CompletableFuture 对象 allFutures。在这个计算过程中，我们没有进行任何计算操作，只是等待所有的任务执行完成。最后，我们使用 allFutures.get() 方法等待所有的计算完成。

anyOf：当任意一个 CompletableFuture 对象计算完成后，返回一个新的 CompletableFuture 对象，这个对象的计算结果是一个 Object 类型的值。

下面是一个使用 anyOf 方法组合多个异步操作的示例：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
try {
// 模拟一个耗时的计算过程
Thread.sleep(1000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
return "Hello";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
try {
// 模拟一个耗时的计算过程
Thread.sleep(2000);
    } catch (InterruptedException e) {
e.printStackTrace();
    }
return "World";
});

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2);

// 等待计算完成，并获取计算结果
Object result = anyFuture.get();

System.out.println(result); // 输出："Hello" 或 "World"

在这个示例中，我们使用 CompletableFuture.supplyAsync 方法异步执行了两个计算过程，分别返回了字符串 "Hello" 和 "World"。然后，我们使用 anyOf 方法将这两个 CompletableFuture 对象组合在一起，并生成一个新的 CompletableFuture 对象 anyFuture。在这个计算过程中，我们只需要等待任意一个计算完成即可。最后，我们使用 anyFuture.get() 方法等待计算完成，并获取计算结果。

delayedExecutor：返回一个 Executor 对象，这个对象会延迟指定的时间再执行任务。

下面是一个使用 delayedExecutor 方法延迟执行任务的示例：

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
System.out.println("任务开始执行...");
}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));

// 等待任务执行完成
future.get();

在这个示例中，我们使用 CompletableFuture.delayedExecutor 方法创建一个 Executor 对象，这个对象会延迟 1 秒再执行任务。然后，我们使用 CompletableFuture.runAsync 方法异步执行了一个任务，并指定了这个 Executor 对象作为参数。最后，我们使用 future.get() 方法等待任务执行完成。

exceptionally：处理异步操作的异常情况，返回一个新的 CompletableFuture 对象，这个对象的计算结果是一个与原始 CompletableFuture 对象相同的值或默认值。

下面是一个使用 exceptionally 方法处理异常情况的示例：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
// 抛出一个异常
throw new RuntimeException("计算出现错误");
}).exceptionally(e -> {
// 处理异常情况
System.out.println(e.getMessage());
// 返回默认值
return 0;
});

// 等待计算完成，并获取计算结果
int result = future.get();

System.out.println("计算结果为：" + result);

在这个示例中，我们使用 CompletableFuture.supplyAsync 方法异步执行了一个计算过程，然后在计算过程中抛出了一个异常。我们使用 exceptionally 方法处理这个异常情况，并返回了默认值 0。最后，我们使用 future.get() 方法等待计算完成，并获取计算结果。

handle：处理异步操作的结果和异常情况，返回一个新的 CompletableFuture 对象，这个对象的计算结果是一个与原始 CompletableFuture 对象不同的值。

下面是一个使用 handle 方法处理结果和异常情况的示例：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
// 返回一个整数值
return 100;
}).handle((result, e) -> {
if (e != null) {
// 处理异常情况
System.out.println(e.getMessage());
// 返回默认值
return 0;
    } else {
// 处理正常情况
System.out.println("计算结果为：" + result);
// 返回计算结果的两倍
return result * 2;
    }
});

// 等待计算完成，并获取计算结果
int result = future.get();

System.out.println("计算结果为：" + result);

在这个示例中，我们使用 CompletableFuture.supplyAsync 方法异步执行了一个计算过程，返回了一个整数值 100。我们使用 handle 方法处理计算结果和异常情况，并返回计算结果的两倍。最后，我们使用 future.get() 方法等待计算完成，并获取计算结果。

到此为止，我们已经介绍了 CompletableFuture 的基本用法和几个重要的方法，包括 thenApply、thenAccept、thenRun、thenCompose、thenCombine、exceptionally、handle、allOf 和 anyOf。在实际应用中，我们可能会使用更多的 CompletableFuture 方法，这里就不一一介绍了。读者可以参考 Java 官方文档和其他相关资源，学习更多关于 CompletableFuture 的知识。


三 结论

在本文中，我们介绍了 Java 8 中的 CompletableFuture 类，这是一种基于 Future 的异步编程模型，提供了方便的 API 和强大的组合能力，帮助我们实现异步操作和并行计算。我们首先介绍了 CompletableFuture 的基本概念和用法，包括如何创建 CompletableFuture 对象、如何异步执行计算过程、如何使用回调函数处理计算结果等等。然后，我们介绍了几个常用的方法，包括 thenApply、thenAccept、thenRun、thenCompose、thenCombine、exceptionally、handle、allOf 和 anyOf，帮助我们更灵活地组合 CompletableFuture 对象。最后，我们通过几个示例代码演示了 CompletableFuture 的用法，包括异步执行单个任务、异步执行多个任务并等待它们都完成、等待任意一个任务完成等等。

CompletableFuture 是一种强大的异步编程模型，它提供了丰富的 API 和强大的组合能力，帮助我们处理异步操作和并行计算的复杂性，同时也使得我们的代码更加简洁和易于理解。在实际应用中，我们可以将 CompletableFuture 与 Java 8 中的 Stream API、Lambda 表达式和方法引用等新特性结合使用，进一步提高代码的可读性和可维护性。

需要注意的是，虽然 CompletableFuture 提供了方便的 API 和强大的组合能力，但也需要我们合理地使用它，避免滥用或误用。在实际应用中，我们应该注意一些潜在的问题，比如线程安全、异常处理、内存泄漏、并发性能等等。特别是在处理复杂的异步操作和并行计算时，我们需要仔细考虑算法、数据结构、任务划分、负载均衡、并发控制等方面的问题，以充分利用多核 CPU 和分布式计算资源，提高程序的性能和可扩展性。

本文中的示例代码仅为演示 CompletableFuture 的基本用法和常见方法，可以根据自己的需求和实际情况，修改和扩展这些代码，实现更加复杂和实用的异步操作和并行计算。如果想深入了解 CompletableFuture 的原理和实现细节，可以参考 Java 8 的源代码和其他相关资源，了解 CompletableFuture 的内部实现和优化策略，以及如何使用 CompletableFuture 更好地实现并发编程和异步操作。

总之，CompletableFuture 是一种非常有用的异步编程模型，它提供了丰富的 API 和强大的组合能力，帮助我们实现异步操作和并行计算。在实际应用中，我们可以根据自己的需求和实际情况，选择适合的方法和策略，合理使用 CompletableFuture，进一步提高代码的质量和性能，实现更加复杂和实用的应用程序。

标签：

相关文章