Java CompletableFuture 使用方法与最佳实践 – wiki基地

---

Java CompletableFuture 深度解析：从入门到精通

在现代软件开发中，尤其是在构建高并发、响应式的应用程序时，异步编程模型变得至关重要。Java 通过 java.util.concurrent.Future 接口为异步计算提供了一种基础机制，但 Future 本身存在一些局限性，例如无法方便地组合多个任务、缺乏非阻塞的回调机制以及异常处理相对繁琐。为了克服这些限制，Java 8 引入了 java.util.concurrent.CompletableFuture，它不仅实现了 Future 和 CompletionStage 接口，还提供了一套强大而灵活的工具，用于构建复杂的异步工作流。

本文将深入探讨 CompletableFuture 的核心概念、使用方法、组合技巧、异常处理机制以及最佳实践，旨在帮助开发者充分利用其潜力，编写出更高效、更健壮、更易于维护的异步 Java 代码。

1. CompletableFuture 概述：为何选择它？

CompletableFuture<T> 代表一个可能尚未完成的异步计算的结果。与传统的 Future 不同，CompletableFuture 具备以下关键优势：

• 非阻塞性: CompletableFuture 鼓励使用回调函数（注册在 CompletionStage 接口中的方法），而不是依赖阻塞的 get() 方法来等待结果。这使得线程可以被释放去处理其他任务，从而提高系统吞吐量和响应性。
• 可组合性 (Composability): 它提供了丰富的 API 来链接、组合和编排多个异步任务。你可以轻松地定义任务之间的依赖关系（顺序执行、并行执行、等待多个或任意一个完成等），构建复杂的异步处理流水线。
• 函数式编程风格: 大量使用了 Lambda 表达式和函数式接口（如 Function, Consumer, Supplier, BiFunction 等），使得代码更简洁、更具表达力。
• 显式完成: 你可以手动创建一个 CompletableFuture 并在稍后某个时间点通过 complete() 或 completeExceptionally() 方法显式地完成它，这对于集成基于事件或回调的 API 非常有用。
• 改进的异常处理: 提供了专门的方法（如 exceptionally, handle, whenComplete）来优雅地处理异步计算过程中可能出现的异常。

2. 创建 CompletableFuture 实例

创建 CompletableFuture 主要有以下几种方式：

• 使用 runAsync() – 执行无返回值的任务:
  如果你需要异步执行一个不返回任何结果的操作（实现了 Runnable 接口），可以使用 runAsync()。

  “`java
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.TimeUnit;

  public class CompletableFutureRunAsync {
  public static void main(String[] args) throws InterruptedException {
  System.out.println(“Main thread: ” + Thread.currentThread().getName());

      // 使用默认的 ForkJoinPool.commonPool()
      CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
          System.out.println("Task 1 running in thread: " + Thread.currentThread().getName());
          try {
              TimeUnit.SECONDS.sleep(1);
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
          }
          System.out.println("Task 1 finished.");
      });
  
      // 使用自定义的 Executor
      ExecutorService customExecutor = Executors.newFixedThreadPool(2);
      CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
          System.out.println("Task 2 running in thread: " + Thread.currentThread().getName());
          try {
              TimeUnit.SECONDS.sleep(1);
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
          }
          System.out.println("Task 2 finished.");
      }, customExecutor);
  
      // 等待任务完成 (仅为演示，实际应用中应避免阻塞)
      future1.join();
      future2.join();
  
      System.out.println("Main thread finished.");
      customExecutor.shutdown(); // 关闭自定义线程池
  }
  

  }
  ``runAsync有两个重载版本：一个使用默认的ForkJoinPool.commonPool()作为执行器，另一个允许你指定自定义的Executor`。

• 使用 supplyAsync() – 执行有返回值的任务:
  如果你的异步任务需要返回一个结果（实现了 Supplier<U> 接口），应该使用 supplyAsync()。

  “`java
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.TimeUnit;

  public class CompletableFutureSupplyAsync {
  public static void main(String[] args) throws Exception {
  System.out.println(“Main thread: ” + Thread.currentThread().getName());

      // 使用默认的 ForkJoinPool.commonPool()
      CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
          System.out.println("Task 1 (supply) running in thread: " + Thread.currentThread().getName());
          try {
              TimeUnit.SECONDS.sleep(1);
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              return "Task 1 Interrupted";
          }
          return "Result from Task 1";
      });
  
      // 使用自定义的 Executor
      ExecutorService customExecutor = Executors.newFixedThreadPool(2);
      CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
          System.out.println("Task 2 (supply) running in thread: " + Thread.currentThread().getName());
          try {
              TimeUnit.SECONDS.sleep(1);
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              return -1;
          }
          return 123;
      }, customExecutor);
  
      // 获取结果 (join() 会阻塞)
      String result1 = future1.join();
      Integer result2 = future2.join();
  
      System.out.println("Result 1: " + result1);
      System.out.println("Result 2: " + result2);
  
      System.out.println("Main thread finished.");
      customExecutor.shutdown();
  }
  

  }
  ``
与runAsync类似，supplyAsync` 也有使用默认线程池和自定义线程池的两个版本。

• 使用 CompletableFuture.completedFuture() – 创建已完成的 Future:
  有时你需要一个立即返回预定结果的 CompletableFuture，这在测试或某些需要 Future 作为起点的链式调用中很有用。

  java
CompletableFuture<String> alreadyCompleted = CompletableFuture.completedFuture("Immediate Result");
System.out.println("Is done? " + alreadyCompleted.isDone()); // true
System.out.println("Result: " + alreadyCompleted.get()); // "Immediate Result" (不会阻塞)

• 手动创建和完成:
  对于需要集成非 Future 基础的异步 API 的情况，可以先创建一个空的 CompletableFuture，然后在回调中手动完成它。

  “`java
  CompletableFuture manualFuture = new CompletableFuture<>();

  // 模拟一个异步操作完成后的回调
  new Thread(() -> {
  try {
  TimeUnit.SECONDS.sleep(2);
  // 模拟成功
  manualFuture.complete(“Result from manual operation”);
  // 或者模拟失败
  // manualFuture.completeExceptionally(new RuntimeException(“Operation failed”));
  } catch (InterruptedException e) {
  manualFuture.completeExceptionally(e);
  }
  }).start();

  // 后续可以链式调用或获取结果
  manualFuture.thenAccept(result -> System.out.println(“Manual future completed with: ” + result));

  // 阻塞等待 (仅为演示)
  System.out.println(“Waiting for manual future…”);
  System.out.println(manualFuture.join());
  “`

3. 处理 CompletableFuture 的结果 (回调函数)

CompletableFuture 的强大之处在于其丰富的回调方法，允许你在 Future 完成时（成功或失败）执行相应的操作，而无需阻塞当前线程。主要的回调方法分为三类：thenApply, thenAccept, 和 thenRun。

• thenApply(Function<? super T,? extends U> fn):
  当 CompletableFuture 成功完成时，将其结果作为输入传递给指定的 Function，并将该 Function 的返回值包装成一个新的 CompletableFuture<U> 返回。这用于转换结果。

  “`java
  CompletableFuture originalFuture = CompletableFuture.supplyAsync(() -> “Hello”);

  CompletableFuture transformedFuture = originalFuture.thenApply(result -> {
  System.out.println(“Applying transformation in thread: ” + Thread.currentThread().getName());
  return result.length(); // 转换 “Hello” 为 5
  });

  System.out.println(“Transformed result: ” + transformedFuture.join()); // 5
  “`

• thenAccept(Consumer<? super T> action):
  当 CompletableFuture 成功完成时，将其结果传递给指定的 Consumer 进行消费（执行某个操作，无返回值）。返回 CompletableFuture<Void>。

  “`java
  CompletableFuture greetingFuture = CompletableFuture.supplyAsync(() -> “World”);

  CompletableFuture acceptedFuture = greetingFuture.thenAccept(result -> {
  System.out.println(“Accepting result in thread: ” + Thread.currentThread().getName());
  System.out.println(“Greeting: ” + result); // 消费结果 “World”
  });

  acceptedFuture.join(); // 等待消费操作完成
  “`

• thenRun(Runnable action):
  当 CompletableFuture 成功完成时，执行指定的 Runnable 任务，该任务不接收结果，也不返回任何值。返回 CompletableFuture<Void>。

  “`java
  CompletableFuture taskFuture = CompletableFuture.supplyAsync(() -> {
  // some operation
  return “Done”;
  });

  CompletableFuture runFuture = taskFuture.thenRun(() -> {
  System.out.println(“Running action after completion in thread: ” + Thread.currentThread().getName());
  System.out.println(“Previous task completed successfully.”);
  });

  runFuture.join(); // 等待后续操作完成
  “`

关于 *Async 变体:

上述每个回调方法 (thenApply, thenAccept, thenRun) 都有一个对应的 *Async 版本（例如 thenApplyAsync）。它们的区别在于回调任务执行的线程：

• 不带 Async 后缀的方法: 回调可能在完成前一个阶段任务的线程中执行，也可能在调用回调注册方法（如 thenApply）的线程中执行。具体取决于前一个阶段是否已经完成。
• 带 Async 后缀的方法 (无 Executor 参数): 回调任务会被提交到默认的 ForkJoinPool.commonPool() 中异步执行。
• 带 Async 后缀的方法 (带 Executor 参数): 回调任务会被提交到你指定的 Executor 中异步执行。

何时使用 Async 版本？

• 当回调任务是 CPU 密集型或阻塞型操作时，应使用 Async 版本，并最好指定一个独立的 Executor，以避免阻塞 ForkJoinPool.commonPool() 或执行前序任务的线程。
• 当回调任务非常轻量且快速时，可以直接使用非 Async 版本，以减少线程切换的开销。

4. 组合 CompletableFuture

组合是 CompletableFuture 最强大的特性之一，允许你构建复杂的异步工作流。

• thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) – 链式依赖:
  当你需要在一个 CompletableFuture 完成后，使用其结果启动另一个异步操作（返回 CompletionStage，通常是另一个 CompletableFuture）时，使用 thenCompose。它类似于 flatMap 操作，避免了 CompletableFuture<CompletableFuture<U>> 这样的嵌套结构。

  “`java
  CompletableFuture getUserFuture = CompletableFuture.supplyAsync(() -> “UserId-123”);

  CompletableFuture getOrderFuture = getUserFuture.thenCompose(userId ->
  // 根据 userId 异步获取订单信息
  CompletableFuture.supplyAsync(() -> {
  System.out.println(“Fetching order for user: ” + userId);
  // 模拟网络调用
  try { TimeUnit.MILLISECONDS.sleep(500); } catch (InterruptedException e) {}
  return “Order details for ” + userId;
  })
  );

  System.out.println(getOrderFuture.join()); // “Order details for UserId-123”
  ``
注意thenCompose的参数函数返回的是一个CompletionStage`。

• thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) – 合并两个独立结果:
  当你有两个独立的 CompletableFuture，并且想在它们都完成后，将它们的结果合并（使用 BiFunction）产生一个新的结果时，使用 thenCombine。

  “`java
  CompletableFuture weightFuture = CompletableFuture.supplyAsync(() -> 70.5); // kg
  CompletableFuture heightFuture = CompletableFuture.supplyAsync(() -> 1.75); // meters

  CompletableFuture bmiFuture = weightFuture.thenCombine(heightFuture, (weight, height) -> {
  System.out.println(“Calculating BMI in thread: ” + Thread.currentThread().getName());
  return weight / (height * height);
  });

  System.out.println(“BMI: ” + bmiFuture.join());
  “`

• thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) – 消费两个独立结果:
  类似于 thenCombine，但在两个 CompletableFuture 都完成后，使用它们的结执行一个 BiConsumer 操作（无返回值）。

  “`java
  CompletableFuture userFuture = CompletableFuture.supplyAsync(() -> “Alice”);
  CompletableFuture productFuture = CompletableFuture.supplyAsync(() -> “Laptop”);

  CompletableFuture combinedAction = userFuture.thenAcceptBoth(productFuture, (user, product) -> {
  System.out.println(user + ” purchased a ” + product);
  });

  combinedAction.join(); // “Alice purchased a Laptop”
  “`

• runAfterBoth(CompletionStage<?> other, Runnable action) – 两个都完成后执行动作:
  在两个 CompletableFuture 都完成后，执行一个 Runnable 任务，不关心它们的结果。

• allOf(CompletableFuture<?>... cfs) – 等待所有 Future 完成:
  接收一个可变参数的 CompletableFuture 数组，返回一个新的 CompletableFuture<Void>，它在所有输入的 Future 都完成后才完成。注意： allOf 返回的 CompletableFuture<Void> 本身不携带任何结果。你需要再次链式调用或手动遍历原始 Future 列表来获取它们各自的结果。

  “`java
  CompletableFuture f1 = CompletableFuture.supplyAsync(() -> “Result 1”);
  CompletableFuture f2 = CompletableFuture.supplyAsync(() -> 2);
  CompletableFuture f3 = CompletableFuture.supplyAsync(() -> true);

  CompletableFuture allFutures = CompletableFuture.allOf(f1, f2, f3);

  // 等待 allFutures 完成
  allFutures.join(); // 阻塞直到 f1, f2, f3 都完成

  // allOf 完成后，可以安全地获取各个 Future 的结果 (它们保证已完成)
  System.out.println(“f1 result: ” + f1.join()); // “Result 1”
  System.out.println(“f2 result: ” + f2.join()); // 2
  System.out.println(“f3 result: ” + f3.join()); // true

  // 或者使用流处理获取所有结果 (更优雅的方式)
  CompletableFuture.allOf(f1, f2, f3).thenRun(() -> {
  try {
  String r1 = f1.get(); // 非阻塞，因为 allOf 保证了完成
  Integer r2 = f2.get();
  Boolean r3 = f3.get();
  System.out.println(“All results via thenRun: ” + r1 + “, ” + r2 + “, ” + r3);
  } catch (Exception e) { // InterruptedException, ExecutionException
  e.printStackTrace();
  }
  }).join();
  “`

• anyOf(CompletableFuture<?>... cfs) – 等待任意一个 Future 完成:
  接收一个可变参数的 CompletableFuture 数组，返回一个新的 CompletableFuture<Object>，它在输入的 Future 中任意一个完成后就立即完成，其结果是第一个完成的 Future 的结果。

  “`java
  CompletableFuture slowTask = CompletableFuture.supplyAsync(() -> {
  try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) {}
  return “Slow Result”;
  });
  CompletableFuture fastTask = CompletableFuture.supplyAsync(() -> {
  try { TimeUnit.MILLISECONDS.sleep(500); } catch (InterruptedException e) {}
  return “Fast Result”;
  });

  CompletableFuture