MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java CompletableFuture allOf协调多个异步任务的执行

2021-06-265.6k 阅读

Java CompletableFuture allOf 协调多个异步任务的执行

CompletableFuture 简介

在现代Java编程中，处理异步任务是一项常见且重要的工作。CompletableFuture 是Java 8引入的一个强大工具，用于处理异步计算。它实现了 Future 接口，同时提供了许多增强功能，使得异步编程更加灵活和高效。CompletableFuture 不仅可以表示一个异步操作的结果，还允许在异步操作完成时执行回调函数，并且支持链式调用和组合多个异步操作。

allOf 方法概述

CompletableFuture 的 allOf 方法是一个静态方法，它接受一个可变参数列表，其中每个参数都是 CompletableFuture 类型的对象。allOf 方法的作用是等待所有给定的 CompletableFuture 都完成。当所有的 CompletableFuture 都完成时，返回的 CompletableFuture 也会完成，其结果为 null。

allOf 方法的语法

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

简单示例：多个异步任务并行执行并等待全部完成

假设我们有三个简单的异步任务，每个任务模拟一个耗时操作，例如网络请求或者文件读取。我们可以使用 CompletableFuture 的 supplyAsync 方法创建异步任务，并使用 allOf 方法等待所有任务完成。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class AllOfExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task 1 completed";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task 2 completed";
        });

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task 3 completed";
        });

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

        allFutures.join();

        try {
            System.out.println(future1.get());
            System.out.println(future2.get());
            System.out.println(future3.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中：

  1. 使用 CompletableFuture.supplyAsync 创建了三个异步任务 future1、future2 和 future3，每个任务都模拟了不同时长的耗时操作，并返回一个字符串结果。
  2. 使用 CompletableFuture.allOf 方法将这三个 CompletableFuture 组合起来，返回一个新的 CompletableFuture<Void>，名为 allFutures。
  3. 调用 allFutures.join() 方法等待所有的异步任务完成。
  4. 最后通过 future1.get()、future2.get() 和 future3.get() 获取每个任务的结果并打印。

allOf 方法的本质

从本质上讲，allOf 方法实现了一种并行任务的协调机制。它通过维护一个内部计数器来跟踪所有传入的 CompletableFuture 的完成状态。每当一个 CompletableFuture 完成时，计数器减一。当计数器变为零时，意味着所有的 CompletableFuture 都已完成，此时返回的 CompletableFuture 也会完成。

allOf 方法的异常处理

在实际应用中，异步任务可能会抛出异常。当使用 allOf 方法时，如果其中任何一个 CompletableFuture 抛出异常，allOf 返回的 CompletableFuture 也会异常完成。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class AllOfExceptionExample {
    public static void main(String[] args) {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("Task 1 failed");
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            return "Task 2 completed";
        });

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

        allFutures.join();

        try {
            System.out.println(future1.get());
            System.out.println(future2.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，future1 抛出了一个运行时异常。当调用 allFutures.join() 时，虽然 future2 正常完成，但由于 future1 抛出异常，allFutures 也会异常完成。在获取 future1.get() 时会抛出 ExecutionException，其内部包含了 future1 抛出的原始异常 RuntimeException("Task 1 failed")。

更复杂的场景：异步任务的依赖与并行

有时候，我们可能有一些异步任务之间存在依赖关系，同时又希望部分任务能够并行执行。CompletableFuture 的 allOf 方法可以与其他方法（如 thenApply、thenCompose 等）结合使用来满足这种需求。

假设我们有以下场景：任务A和任务B可以并行执行，任务C需要在任务A和任务B都完成后执行。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class DependentTasksExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Result of Task A";
        });

        CompletableFuture<String> futureB = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Result of Task B";
        });

        CompletableFuture<Void> allAB = CompletableFuture.allOf(futureA, futureB);

        CompletableFuture<String> futureC = allAB.thenApply(v -> {
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                return "Task C combined: " + futureA.get() + " and " + futureB.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        });

        System.out.println(futureC.get());
    }
}

在上述代码中：

  1. futureA 和 futureB 并行执行。
  2. allAB 使用 allOf 方法等待 futureA 和 futureB 都完成。
  3. futureC 通过 allAB.thenApply 方法在 futureA 和 futureB 完成后执行，并且在 futureC 中获取 futureA 和 futureB 的结果并进行组合。

allOf 方法与线程池

在实际应用中，我们通常会使用线程池来管理异步任务的执行。CompletableFuture 的 supplyAsync 方法有一个重载版本，可以接受一个 Executor 参数，用于指定任务执行的线程池。

import java.util.concurrent.*;

public class AllOfWithThreadPoolExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(3);

        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task 1 completed";
        }, executor);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task 2 completed";
        }, executor);

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task 3 completed";
        }, executor);

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

        allFutures.join();

        try {
            System.out.println(future1.get());
            System.out.println(future2.get());
            System.out.println(future3.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
        }
    }
}

在上述代码中，我们创建了一个固定大小为3的线程池 executor，并将其传递给 supplyAsync 方法，这样所有的异步任务都会在这个线程池中执行。最后在程序结束时，调用 executor.shutdown() 关闭线程池。

allOf 方法在实际项目中的应用场景

  1. 批处理操作：在处理大量数据时，可能需要对数据进行分组并并行处理。例如，在数据分析项目中，将数据分成多个批次，每个批次的数据处理作为一个异步任务，使用 allOf 方法等待所有批次处理完成后再进行汇总分析。
  2. 微服务调用：在微服务架构中，一个业务操作可能需要调用多个不同的微服务接口。这些接口调用可以并行发起，使用 allOf 方法等待所有接口调用完成后，再对返回结果进行整合和处理。
  3. 资源加载与初始化：在应用程序启动时，可能需要加载多个资源，如配置文件、数据库连接池初始化、缓存预热等。这些资源加载操作可以异步进行，使用 allOf 方法确保所有资源都加载完成后，应用程序再正式启动。

性能考虑

虽然 CompletableFuture 的 allOf 方法提供了强大的异步任务协调能力，但在使用时也需要考虑性能问题。

  1. 线程资源消耗：如果创建过多的异步任务并使用 allOf 方法等待它们完成，可能会消耗大量的线程资源。尤其是在没有合理配置线程池的情况下，可能导致线程耗尽，影响系统性能。因此，需要根据实际情况合理设置线程池的大小。
  2. 任务执行顺序：虽然 allOf 方法等待所有任务完成，但任务完成的顺序是不确定的。如果在某些场景下需要按照特定顺序处理任务结果，需要额外的逻辑来确保顺序。
  3. 异常处理开销：在处理异步任务异常时，捕获和处理异常可能会带来一定的性能开销。尤其是在大量异步任务并发执行的情况下，合理的异常处理策略可以避免性能瓶颈。

总结与注意事项

CompletableFuture 的 allOf 方法是Java异步编程中的一个重要工具，它允许我们方便地协调多个异步任务的执行，等待所有任务完成后再进行下一步操作。在使用时，需要注意异常处理、线程池的合理配置以及性能优化等方面。通过合理运用 allOf 方法，可以显著提高程序的并发性能和响应速度，特别是在处理复杂的异步操作场景中。同时，结合 CompletableFuture 的其他方法，如 thenApply、thenCompose 等，可以构建出更加灵活和高效的异步编程模型。在实际项目中，根据具体的业务需求和场景，合理选择和使用这些功能，能够提升系统的整体性能和稳定性。

在使用 allOf 方法时，以下几点需要特别注意：

  1. 任务数量限制：虽然 allOf 方法可以接受任意数量的 CompletableFuture 对象作为参数，但如果任务数量过多，可能会导致内存消耗过大或者系统性能下降。在实际应用中，需要根据系统资源和性能要求，合理控制任务数量。
  2. 结果获取方式：allOf 方法返回的 CompletableFuture 的结果为 null。如果需要获取每个子任务的结果，需要单独处理每个 CompletableFuture，如上述示例中通过 future1.get()、future2.get() 等方式获取结果。
  3. 异常处理策略：由于任何一个子任务抛出异常都会导致 allOf 返回的 CompletableFuture 异常完成，因此在处理异常时，需要考虑如何统一捕获和处理所有子任务可能抛出的异常，以确保程序的健壮性。

通过深入理解和合理运用 CompletableFuture 的 allOf 方法，开发人员可以更加高效地编写异步代码，充分利用多核处理器的性能优势，提升Java应用程序的并发处理能力。无论是在大型企业级应用开发，还是在小型项目中，掌握这一技术都能够为解决异步编程问题提供有力的支持。

希望通过以上详细的讲解和丰富的示例，能让你对 Java CompletableFuture allOf 协调多个异步任务的执行 有更深入的理解和掌握，在实际编程中能够灵活运用这一强大的功能。