Java CompletableFuture allOf实现并行任务等待的策略
Java CompletableFuture allOf实现并行任务等待的策略
CompletableFuture简介
在Java的并发编程领域，CompletableFuture是JDK 8引入的一个强大工具，它极大地简化了异步编程，允许我们以一种更直观和高效的方式处理异步任务及其结果。CompletableFuture实现了Future和CompletionStage接口，既可以用于获取异步计算的结果（类似于传统的Future），又能通过CompletionStage接口定义的各种方法对异步任务进行组合、转换和链式调用。
传统的Future接口存在一些局限性，例如获取结果时如果任务未完成会导致阻塞，缺乏对异步任务组合和转换的便捷方法等。而CompletableFuture通过提供丰富的方法，如thenApply、thenAccept、thenRun等，允许我们在异步任务完成后执行一系列操作，并且这些操作可以异步执行，不会阻塞主线程。
并行任务的概念与需求
在实际的应用开发中，经常会遇到需要同时执行多个任务的场景，这些任务之间可能相互独立，没有先后顺序的严格依赖。例如，一个电商应用在展示商品详情时，可能需要同时从不同的数据源获取商品的基本信息、价格信息、用户评价等。如果依次执行这些任务，整个操作的耗时将是各个任务耗时之和，这在性能上是不可接受的。
并行任务的执行可以显著提高应用的性能和响应速度。通过将多个任务并行执行，我们可以利用多核CPU的优势，让不同的任务在不同的CPU核心上同时运行。然而，在并行任务执行完毕后，我们通常需要等待所有任务都完成，然后再进行后续的处理，比如将各个任务的结果进行汇总和整合。这就引出了对并行任务等待策略的需求。
allOf方法概述
CompletableFuture的allOf方法正是为了解决上述并行任务等待的问题而设计的。allOf方法接受一个可变参数列表，其中每个参数都是一个CompletableFuture对象。它返回一个新的CompletableFuture，这个新的CompletableFuture会在所有传入的CompletableFuture都完成（无论是正常完成还是异常完成）时完成。
allOf方法的签名如下：
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
可以看到，allOf方法返回的CompletableFuture的泛型类型是Void，这是因为它并不关心各个子任务的具体返回值，只关注所有子任务是否都完成。
allOf方法的使用示例
下面通过一个简单的代码示例来展示allOf方法的基本用法。假设我们有三个异步任务，分别模拟从不同数据源获取数据的操作：
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureAllOfExample {
public static CompletableFuture<String> fetchData1() {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000); // 模拟耗时操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return "Data from source 1";
});
}
public static CompletableFuture<String> fetchData2() {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(3000); // 模拟耗时操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return "Data from source 2";
});
}
public static CompletableFuture<String> fetchData3() {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000); // 模拟耗时操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return "Data from source 3";
});
}
public static void main(String[] args) {
CompletableFuture<String> future1 = fetchData1();
CompletableFuture<String> future2 = fetchData2();
CompletableFuture<String> future3 = fetchData3();
CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2, future3);
allFutures.join(); // 等待所有任务完成
try {
String result1 = future1.get();
String result2 = future2.get();
String result3 = future3.get();
System.out.println("Combined data: " + result1 + ", " + result2 + ", " + result3);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}
在上述代码中：
fetchData1、fetchData2和fetchData3方法分别创建了三个CompletableFuture，每个CompletableFuture通过supplyAsync方法异步执行一个模拟的耗时操作，并返回一个字符串结果。- 使用
CompletableFuture.allOf方法创建了一个新的CompletableFuture，传入前面三个CompletableFuture对象。 - 调用
allFutures.join()方法等待所有任务完成。join方法与get方法类似，但它不会抛出已检查异常（InterruptedException和ExecutionException），而是将异常包装成CompletionException。 - 最后通过
future1.get()、future2.get()和future3.get()获取各个任务的结果，并进行打印。
allOf方法的执行原理
allOf方法的实现原理涉及到CompletableFuture内部的状态管理和异步任务调度机制。当调用allOf方法时，它会为每个传入的CompletableFuture注册一个完成监听器。这些监听器会在对应的CompletableFuture完成时被触发。
在allOf返回的新CompletableFuture内部，维护了一个计数器，初始值为传入的CompletableFuture的数量。每当一个子CompletableFuture完成时，这个计数器就会减1。当计数器减为0时，意味着所有子CompletableFuture都已完成，此时allOf返回的CompletableFuture也会完成。
如果在任何一个子CompletableFuture执行过程中抛出了异常，allOf返回的CompletableFuture也会以异常的方式完成。后续调用join或get方法时，会抛出包含子任务异常信息的CompletionException或ExecutionException。
异常处理策略
在使用allOf方法时，异常处理是一个重要的方面。由于allOf返回的CompletableFuture只关注所有子任务是否完成，并不直接提供获取各个子任务异常信息的便捷方法。
一种常见的异常处理策略是在每个子CompletableFuture中单独处理异常。例如，可以使用exceptionally方法为每个子任务设置异常处理逻辑：
CompletableFuture<String> future1 = fetchData1().exceptionally(ex -> {
System.err.println("Error in fetchData1: " + ex.getMessage());
return "Default value for data1";
});
CompletableFuture<String> future2 = fetchData2().exceptionally(ex -> {
System.err.println("Error in fetchData2: " + ex.getMessage());
return "Default value for data2";
});
CompletableFuture<String> future3 = fetchData3().exceptionally(ex -> {
System.err.println("Error in fetchData3: " + ex.getMessage());
return "Default value for data3";
});
CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2, future3);
allFutures.join();
try {
String result1 = future1.get();
String result2 = future2.get();
String result3 = future3.get();
System.out.println("Combined data: " + result1 + ", " + result2 + ", " + result3);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
在上述代码中，通过exceptionally方法为每个CompletableFuture设置了异常处理逻辑。如果某个子任务抛出异常，会打印异常信息并返回一个默认值。这样即使某个子任务出现异常，整个并行任务的流程也不会被中断，并且可以继续获取其他子任务的结果并进行后续处理。
另一种策略是在allOf返回的CompletableFuture完成后，检查各个子CompletableFuture的状态。例如，可以通过isCompletedExceptionally方法判断某个CompletableFuture是否以异常方式完成：
CompletableFuture<String> future1 = fetchData1();
CompletableFuture<String> future2 = fetchData2();
CompletableFuture<String> future3 = fetchData3();
CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2, future3);
allFutures.join();
if (future1.isCompletedExceptionally()) {
try {
future1.get();
} catch (InterruptedException | ExecutionException e) {
System.err.println("Error in future1: " + e.getMessage());
}
}
if (future2.isCompletedExceptionally()) {
try {
future2.get();
} catch (InterruptedException | ExecutionException e) {
System.err.println("Error in future2: " + e.getMessage());
}
}
if (future3.isCompletedExceptionally()) {
try {
future3.get();
} catch (InterruptedException | ExecutionException e) {
System.err.println("Error in future3: " + e.getMessage());
}
}
try {
String result1 = future1.get();
String result2 = future2.get();
String result3 = future3.get();
System.out.println("Combined data: " + result1 + ", " + result2 + ", " + result3);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
这种方法可以获取每个子任务的详细异常信息，但代码相对繁琐，需要对每个CompletableFuture进行单独的状态检查。
与其他并行处理方式的比较
-
与传统线程和
Future的比较：传统的线程和Future方式在处理并行任务等待时，需要手动管理线程的创建、启动和等待，代码复杂度较高。例如，使用ExecutorService提交多个任务并获取结果时，需要遍历Future列表并调用get方法，这可能会导致阻塞。而CompletableFuture.allOf方法通过链式调用和异步回调的方式，极大地简化了并行任务的管理和等待逻辑，使代码更加简洁和可读。 -
与
CountDownLatch的比较：CountDownLatch是Java并发包中用于同步的工具类，它可以实现让一个或多个线程等待其他线程完成一组操作。在功能上，CountDownLatch可以实现类似allOf的等待所有任务完成的效果，但CountDownLatch更侧重于线程的同步，而CompletableFuture.allOf更专注于异步任务的处理和结果获取。CompletableFuture提供了丰富的方法用于异步任务的组合、转换和异常处理，这是CountDownLatch所不具备的。 -
与
Fork/Join框架的比较：Fork/Join框架主要用于解决能够被分解为多个子任务，并且子任务的结果可以合并的问题，它采用分治算法的思想。而CompletableFuture.allOf更适用于多个相互独立的异步任务并行执行并等待所有任务完成的场景。Fork/Join框架更注重任务的递归分解和结果合并，而CompletableFuture在异步任务的灵活性和易用性方面更具优势。
allOf方法在实际项目中的应用场景
-
微服务调用：在微服务架构中，一个业务操作可能需要调用多个不同的微服务接口来获取数据。例如，一个订单查询功能可能需要同时调用用户服务获取用户信息、商品服务获取商品信息和订单服务获取订单详情。通过
CompletableFuture.allOf可以并行调用这些微服务接口，并在所有调用完成后进行数据的整合和返回。 -
数据预处理：在大数据处理场景中，可能需要对数据进行多个预处理步骤，如数据清洗、格式转换、特征提取等。这些步骤之间可能相互独立，可以并行执行。使用
CompletableFuture.allOf可以在所有预处理任务完成后，将处理后的数据传递给后续的分析或存储模块。 -
并发数据加载：在应用启动时，可能需要从多个数据源加载配置信息、初始化缓存数据等。通过并行加载这些数据，并使用
CompletableFuture.allOf等待所有加载任务完成，可以加快应用的启动速度。
性能优化与注意事项
-
线程池的合理配置：
CompletableFuture的异步任务执行依赖于线程池。在使用allOf方法时，要确保线程池的大小和配置能够满足并行任务的需求。如果线程池过小，可能会导致任务排队等待，降低并行效率；如果线程池过大，可能会造成资源浪费和线程上下文切换开销。 -
避免不必要的阻塞：虽然
CompletableFuture提供了异步处理的能力，但在获取任务结果时，如果不注意，仍然可能会引入阻塞。例如，在allOf返回的CompletableFuture完成之前就调用子CompletableFuture的get方法，会导致主线程阻塞。应尽量使用异步回调的方式处理任务结果，避免不必要的阻塞。 -
异常处理的完整性：在实际项目中，要确保对并行任务中的异常进行全面处理。不仅要处理子
CompletableFuture的异常，还要考虑allOf返回的CompletableFuture在异常情况下的处理，以保证系统的稳定性和可靠性。 -
资源管理：并行任务可能会占用较多的系统资源，如内存、网络连接等。在任务完成后，要及时释放这些资源，避免资源泄漏。例如，在从数据库获取数据的异步任务完成后，要关闭数据库连接。
高级应用与拓展
- 嵌套并行任务：在实际应用中，可能会遇到嵌套并行任务的情况，即一个并行任务中的某个子任务又包含多个并行子任务。
CompletableFuture的allOf方法可以很好地应对这种情况。例如：
CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> "Task 1 result");
CompletableFuture<String> subTask1 = CompletableFuture.supplyAsync(() -> "Sub - task 1 result");
CompletableFuture<String> subTask2 = CompletableFuture.supplyAsync(() -> "Sub - task 2 result");
CompletableFuture<Void> innerAll = CompletableFuture.allOf(subTask1, subTask2);
CompletableFuture<String> task2 = innerAll.thenApply(v -> {
try {
String subResult1 = subTask1.get();
String subResult2 = subTask2.get();
return "Task 2 result with sub - results: " + subResult1 + ", " + subResult2;
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
return "Error in task 2";
}
});
CompletableFuture<Void> outerAll = CompletableFuture.allOf(task1, task2);
outerAll.join();
try {
String result1 = task1.get();
String result2 = task2.get();
System.out.println("Final results: " + result1 + ", " + result2);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
在上述代码中，task2依赖于subTask1和subTask2的结果，通过innerAll等待subTask1和subTask2完成，然后在thenApply中处理子任务的结果。最后通过outerAll等待task1和task2完成。
- 动态生成并行任务：在某些情况下，并行任务的数量和具体内容可能是动态生成的。例如，根据用户的输入或数据库查询结果决定要并行执行的任务数量和类型。可以通过动态构建
CompletableFuture数组并传递给allOf方法来实现：
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
public class DynamicParallelTasks {
public static CompletableFuture<String> createTask(int taskId) {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(taskId * 1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return "Task " + taskId + " result";
});
}
public static void main(String[] args) {
List<CompletableFuture<String>> taskList = new ArrayList<>();
for (int i = 1; i <= 3; i++) {
CompletableFuture<String> task = createTask(i);
taskList.add(task);
}
CompletableFuture<Void> allFutures = CompletableFuture.allOf(taskList.toArray(new CompletableFuture[0]));
allFutures.join();
taskList.forEach(future -> {
try {
System.out.println(future.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
});
}
}
在上述代码中，根据循环动态创建了多个CompletableFuture任务，并将它们添加到列表中。然后通过allOf方法等待所有任务完成，并获取每个任务的结果。
- 与流（Stream）的结合使用：Java 8的流（Stream）API提供了强大的集合处理能力，
CompletableFuture可以与流结合使用，进一步简化并行任务的创建和处理。例如：
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
public class CompletableFutureWithStream {
public static CompletableFuture<String> processData(String data) {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return "Processed " + data;
});
}
public static void main(String[] args) {
List<String> dataList = Arrays.asList("A", "B", "C");
List<CompletableFuture<String>> futureList = dataList.stream()
.map(CompletableFutureWithStream::processData)
.collect(Collectors.toList());
CompletableFuture<Void> allFutures = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]));
allFutures.join();
List<String> results = futureList.stream()
.map(future -> {
try {
return future.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
return "Error";
}
})
.collect(Collectors.toList());
System.out.println(results);
}
}
在上述代码中，首先通过流对列表中的每个元素创建一个CompletableFuture任务，然后使用allOf等待所有任务完成，最后通过流获取每个任务的结果并收集到一个新的列表中。
通过以上高级应用和拓展，可以进一步发挥CompletableFuture.allOf方法在复杂异步任务场景中的强大功能，提升应用的性能和灵活性。
在实际应用中，应根据具体的业务需求和场景，合理选择和运用CompletableFuture.allOf方法，结合异常处理、性能优化等方面的考虑，编写高效、可靠的异步代码。同时，要不断学习和探索CompletableFuture的更多特性和用法，以应对日益复杂的并发编程挑战。