【Java异步编程】CompletableFuture实现：异步任务的串行执行

在 Java 异步编程中，CompletableFuture 类提供了多种方法来处理任务的依赖和执行顺序。理解这些方法如何协同工作，可以帮助我们更高效地处理复杂的异步操作。

一. thenApply()：转换计算结果

thenApply() 方法允许我们在前一个 CompletableFuture 执行完成后，基于其计算结果执行转换操作，并返回一个新的 CompletableFuture。这个方法非常适合用于链式转换操作，例如数据转换或映射。

1. 一个线程中执行或多个线程中执行

三个重载方法如下：

     //后一个任务与前一个任务在同一个线程中执行public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)//后一个任务与前一个任务不在同一个线程中执行public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)//后一个任务在指定的executor线程池中执行 public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

可以看到thenApply可以将前后任务串到一个线程中执行，或者异步执行。

 

2. 使用场景说明

假设你从一个异步任务中获取了一个整数结果，而你需要对其进行一些计算或转换。通过 thenApply()，你可以轻松地对该结果进行处理。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 2);
future.thenApply(result -> result * 2)  // result = 2, 返回结果是 4.thenAccept(result -> System.out.println(result));  // 输出 4> **分析**：
> 
> - `CompletableFuture.supplyAsync(() -> 2)`：异步执行任务，返回结果 2。
> - `thenApply(result -> result * 2)`：对结果进行转换，返回 4。
> - `thenAccept(result -> System.out.println(result))`：消费结果，打印 4。
> 
> `thenApply()` 适用于当我们需要基于前一个任务的计算结果执行某种转换时，它帮助我们创建一个新的异步任务，并将处理后的结果返回。

 
再看一个例子：

@Test  
public void thenApplyDemo() throws Exception {  CompletableFuture<Long> future = CompletableFuture.supplyAsync(new Supplier<Long>() {  @Override  public Long get() {  long firstStep = 10L + 10L;  Print.tco("firstStep outcome is " + firstStep);  return firstStep;  }  }).thenApplyAsync(new Function<Long, Long>() {  @Override  public Long apply(Long firstStepOutCome) {  long secondStep = firstStepOutCome * 2;  Print.tco("secondStep outcome is " + secondStep);  return secondStep;  }  });  long result = future.get();  Print.tco(" future is " + future);  Print.tco(" outcome is " + result);  
}[ForkJoinPool.commonPool-worker-9]：firstStep outcome is 20
[ForkJoinPool.commonPool-worker-9]：secondStep outcome is 40
[main]： future is java.util.concurrent.CompletableFuture@5b37e0d2[Completed normally]
[main]： outcome is 40
JVM退出钩子(定时和顺序任务线程池) starting.... 
JVM退出钩子(定时和顺序任务线程池)  耗时(ms): 0

 

二. thenRun()：执行无返回值的操作

与 thenApply() 不同，thenRun() 方法用于在前一个 CompletableFuture 执行完毕后，执行一个没有返回值的操作。通常用于执行副作用操作，比如打印日志、更新状态等。

1. 语法说明

前后任务是否在一个线程中执行

     //后一个任务与前一个任务在同一个线程中执行public CompletionStage<Void> thenRun(Runnable action);//后一个任务与前一个任务不在同一个线程中执行public CompletionStage<Void> thenRunAsync(Runnable action);//后一个任务在executor线程池中执行public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

thenRun系列方法中的action参数是Runnable类型的，所以thenRun()既不能接收参数又不支持返回值。

 

2. 使用场景说明

假设你只关心任务完成后的某个动作，但不需要使用前一个任务的计算结果。thenRun() 正是为这种场景设计的。

CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> 2);
future.thenRun(() -> System.out.println("Task finished"));  // 输出 "Task finished"**分析**：- `CompletableFuture.supplyAsync(() -> 2)`：异步任务返回结果 2，但我们并不关心它。
- `thenRun(() -> System.out.println("Task finished"))`：
- 在任务完成后执行无返回值的副作用操作，输出 "Task finished"。

thenRun() 不需要前一个任务的结果，只是执行一个副作用操作。因此，它常用于在任务完成后进行一些收尾工作，比如清理资源、记录日志等。

 

三. thenAccept()：消费计算结果

thenAccept() 方法与 thenApply() 类似，不同之处在于它不返回任何值，只是消费前一个 CompletableFuture 的结果。它通常用于当你只关心处理结果，而不需要转换它时。

1. 语法说明

a. 前后任务是否在一个线程中执行

     //后一个任务与前一个任务在同一个线程中执行public CompletionStage<Void> thenAccept(Consumer<? super T> action);//后一个任务与前一个任务不在同一个线程中执行public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);//后一个任务在指定的executor线程池中执行public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);

 

b. 要点说明

     @FunctionalInterfacepublic interface Consumer<T> {void accept(T t);}

1. Consumer<T>接口的accept()方法可以接收一个参数，但是不支持返回值，所以thenAccept()可以将前一个任务的结果及该阶段性的结果通过void accept(T t)方法传递到下一个任务。

2. Consumer<T>接口的accept()方法没有返回值，所以thenAccept()方法也不能提供第二个任务的执行结果。

 

2. 使用场景说明

假设你从异步任务中获取了一个值，你只需要打印它或记录它，而不需要对其进行转换或进一步操作。thenAccept() 就是为这种需求设计的。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 2);
future.thenAccept(result -> System.out.println("Result: " + result));  // 输出 "Result: 2"**分析**：
- `CompletableFuture.supplyAsync(() -> 2)`：异步执行任务，返回结果 2。
- `thenAccept(result -> System.out.println("Result: " + result))`：消费结果，打印 "Result: 2"。

 

 

四. thenCompose()：处理嵌套异步任务

thenCompose() 是一种用于组合多个异步任务的方法。当你需要基于前一个任务的结果返回另一个 CompletableFuture 时，thenCompose() 是最适合的选择。它避免了“回调地狱”（Callback Hell），允许我们将多个异步操作串联在一起。

 

1. 语法说明

     public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn);public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) ;public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) ;

 

2. 场景说明

假设你需要先执行一个异步任务，获得它的结果后，再执行另一个基于该结果的异步任务。thenCompose() 就是解决这个问题的理想工具。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 2);
future.thenCompose(result -> CompletableFuture.supplyAsync(() -> result * 2)).thenAccept(result -> System.out.println(result));  // 输出 4**分析**：- `CompletableFuture.supplyAsync(() -> 2)`：异步任务返回结果 2。
- `thenCompose(result -> CompletableFuture.supplyAsync(() -> result * 2))`：
- 基于前一个任务的结果，执行另一个异步任务，返回新的 `CompletableFuture`，结果是 4。
- `thenAccept(result -> System.out.println(result))`：消费最终结果，输出 4。

通过 thenCompose()，我们可以将多个异步任务串联在一起，并将前一个任务的结果传递给下一个任务。它使得我们能够清晰地处理依赖链，并避免嵌套的回调函数。

 

例子2：

@Test  
public void thenComposeDemo() throws Exception {  CompletableFuture<Long> future = CompletableFuture.supplyAsync(new Supplier<Long>() {  @Override  public Long get() {  long firstStep = 10L + 10L;  Print.tco("firstStep outcome is " + firstStep);  return firstStep;  }  }).thenCompose(new Function<Long, CompletionStage<Long>>() {  @Override  public CompletionStage<Long> apply(Long firstStepOutCome) {  //组装了第二个子任务  return CompletableFuture.supplyAsync(new Supplier<Long>() {  @Override  public Long get() {  long secondStep = firstStepOutCome * 2;  Print.tco("secondStep outcome is " + secondStep);  return secondStep;  }  });  }  });  long result = future.get();  Print.tco(" outcome is " + result);  
}

 

五. 总结

1. thenCompose与thenApply的区别

1. thenCompose()返回的是包装了普通异步方法的CompletionStage任务实例，通过该实例还可以进行下一轮CompletionStage任务的调度和执行，比如可以持续进行CompletionStage链式（或者流式）调用。

2. thenApply()的返回值则简单多了，直接就是第二个任务的普通异步方法的执行结果，它的返回类型与第二步执行的普通异步方法的返回类型相同，通过thenApply()所返回的值不能进行下一轮CompletionStage链式（或者流式）调用。

特性	thenApply()	thenCompose()
返回值	返回一个新的 CompletableFuture<T>，其中 T 是转换后的结果类型。	返回一个新的 CompletableFuture<U>，其中 U 是通过前一个 CompletableFuture 的结果生成的另一个 CompletableFuture。
返回类型	返回一个单一的值（即结果的转换）。	返回一个嵌套的 CompletableFuture，通常用于链式异步任务。
作用	对前一个 CompletableFuture 的结果进行转换并返回新的结果。	使用前一个任务的结果去执行另一个异步任务，并将该异步任务的结果返回。
适用场景	适用于需要对结果进行处理、转换或者映射时。	适用于需要处理嵌套异步任务（即结果依赖于另一个异步任务）时。

 

2. 四者区别

CompletableFuture 提供了丰富的方法来管理异步任务之间的关系。通过理解和使用 thenApply()、thenRun()、thenAccept() 和 thenCompose()，我们可以灵活地控制任务的执行顺序、结果的传递和副作用的执行。

• thenApply()：用于基于前一个任务的结果进行转换，并返回新的 CompletableFuture。
• thenRun()：用于执行不依赖于结果的操作，常用于副作用处理。
• thenAccept()：用于消费前一个任务的结果，通常用于打印或记录日志。
• thenCompose()：用于处理依赖于前一个任务结果的嵌套异步任务。