跳转至

Java CompletableFuture 示例：深入探索异步编程

简介

在现代的 Java 编程中，异步处理变得越来越重要，特别是在构建高性能、响应式的应用程序时。CompletableFuture 是 Java 8 引入的一个强大的类，它极大地简化了异步编程，允许我们以一种更直观、更灵活的方式处理异步任务及其结果。本文将通过丰富的示例详细介绍 CompletableFuture 的基础概念、使用方法、常见实践以及最佳实践，帮助读者更好地掌握这一特性。

目录

  1. 基础概念
  2. 使用方法
    • 创建 CompletableFuture
    • 执行异步任务
    • 获取任务结果
    • 处理任务完成
  3. 常见实践
    • 多任务并行处理
    • 任务依赖关系
  4. 最佳实践
    • 资源管理
    • 异常处理
  5. 小结
  6. 参考资料

基础概念

CompletableFuture 实现了 Future 接口和 CompletionStage 接口。Future 接口在 Java 早期版本就已存在，用于表示异步任务的结果，但它的功能相对有限，例如获取结果时可能需要阻塞等待。而 CompletionStage 接口提供了更丰富的方法来处理异步任务的完成，允许链式调用和组合多个异步操作。

CompletableFuture 可以由任务正常完成、异常完成或者被取消，并且可以在任务完成时执行后续的操作，这种灵活性使得它在异步编程中非常强大。

使用方法

创建 CompletableFuture

  1. 使用默认构造函数 java CompletableFuture<String> future = new CompletableFuture<>(); 这种方式创建的 CompletableFuture 不会自动执行任何任务，需要手动完成它。例如： java future.complete("Hello, CompletableFuture!");
  2. 使用静态工厂方法
    • CompletableFuture.supplyAsync(Supplier<T> supplier)：创建一个异步任务，该任务返回一个结果。 java CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { // 模拟耗时操作 try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "Task completed asynchronously"; });
    • CompletableFuture.runAsync(Runnable runnable)：创建一个异步任务，该任务不返回结果。 java CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> { // 模拟耗时操作 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("This is a void returning async task"); });

执行异步任务

上述静态工厂方法创建的 CompletableFuture 会自动在一个线程池中执行任务。默认情况下，使用的是 ForkJoinPool.commonPool()。如果需要使用自定义的线程池，可以使用带有 Executor 参数的方法：

Executor executor = Executors.newFixedThreadPool(5);
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    // 任务逻辑
    return "Task executed with custom executor";
}, executor);

获取任务结果

  1. 阻塞获取结果
    • T get()：等待任务完成并返回结果，如果任务还未完成，调用线程将被阻塞。 java try { String result = future1.get(); System.out.println(result); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); }
    • T get(long timeout, TimeUnit unit)：等待指定时间后如果任务仍未完成，将抛出 TimeoutException。 java try { String result = future1.get(1, TimeUnit.SECONDS); System.out.println(result); } catch (InterruptedException | ExecutionException | TimeoutException e) { e.printStackTrace(); }
  2. 非阻塞获取结果
    • boolean isDone()：检查任务是否已完成。
    • T join()：与 get() 类似，但不会抛出 InterruptedException 和 ExecutionException，而是将异常包装在 CompletionException 中。如果任务正常完成，直接返回结果。 java String result = future1.join(); System.out.println(result);

处理任务完成

  1. thenApply：任务完成后对结果进行转换。 java CompletableFuture<String> future4 = future1.thenApply(s -> s + " and transformed"); try { String result = future4.get(); System.out.println(result); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); }
  2. thenAccept：任务完成后消费结果，但不返回新的结果。 java future1.thenAccept(s -> System.out.println("Consumed result: " + s));
  3. thenRun：任务完成后执行一个无参的 Runnable。 java future1.thenRun(() -> System.out.println("Task has completed"));

常见实践

多任务并行处理

假设我们有多个独立的任务需要并行执行，然后汇总结果。

CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> "Task 1 result");
CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> "Task 2 result");
CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> "Task 3 result");

CompletableFuture<Void> allTasks = CompletableFuture.allOf(task1, task2, task3);

allTasks.join(); // 等待所有任务完成

String combinedResult = task1.join() + " " + task2.join() + " " + task3.join();
System.out.println(combinedResult);

任务依赖关系

有时一个任务需要依赖另一个任务的结果。

CompletableFuture<String> task4 = CompletableFuture.supplyAsync(() -> "Task 4 result");
CompletableFuture<String> dependentTask = task4.thenApply(s -> s + " processed further");

try {
    String result = dependentTask.get();
    System.out.println(result);
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

最佳实践

资源管理

在使用自定义线程池时，要确保正确管理线程池资源。例如，在应用程序关闭时，正确关闭线程池。

Executor executor = Executors.newFixedThreadPool(5);
// 使用 executor 创建 CompletableFuture

// 在应用程序关闭时
executor.shutdown();
try {
    if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
        executor.shutdownNow();
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            System.err.println("Pool did not terminate");
        }
    }
} catch (InterruptedException ie) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}

异常处理

在异步任务中，异常处理非常重要。可以使用 exceptionally 方法来处理异常。

CompletableFuture<String> future5 = CompletableFuture.supplyAsync(() -> {
    if (Math.random() < 0.5) {
        throw new RuntimeException("Task failed");
    }
    return "Task success";
}).exceptionally(ex -> {
    System.out.println("Caught exception: " + ex.getMessage());
    return "Default value";
});

try {
    String result = future5.get();
    System.out.println(result);
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

小结

CompletableFuture 为 Java 开发者提供了强大的异步编程能力，通过清晰的创建、执行、获取结果和处理完成的方式，以及丰富的组合操作，能够有效提高应用程序的性能和响应性。在实际应用中，遵循最佳实践，如正确的资源管理和异常处理，能够确保异步代码的稳定性和可靠性。希望通过本文的介绍和示例，读者能够更好地掌握并应用 CompletableFuture 进行异步编程。

参考资料