CompletableFuture探索之旅-京东云开发者社区

### 1 背景

最近看了一篇关于接口优化的文章，其中在处理顺序调用复杂逻辑时，建议使用CompletableFuture,什么是CompletableFuture，他能做什么？为什么用它来处理顺序逻辑？让我们一起来探索一下。

### 2 CompletableFuture

CompletableFuture是1.8新加的类：

```java
public class CompletableFuture<T> implements Future<T>, CompletionStage<T>
```

如图代码所示CompletableFuture实现了Future和CompletionStage。

#### 2.1 Future

Future 是JDK1.5添加的接口。Future表示异步计算的结果。 提供了一些方法来检查计算是否完成、等待其完成以及检索计算的结果。

```java
public interface Future<V> {

boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}
```

如图所示：Future定义了4个方法：

- cancel()用来取消任务，如果无法取消返回false。
- isCancelled()如果任务在完成之前就取消了，返回true。
- isDone()如果任务已经完成，返回true。
- V get() 等待任务完成后，获取任务的结果，如果异常抛出相应异常。

#### 2.2 CompletionStage

CompletionStage是1.8新增的接口。CompletionStage的方法服务与异步任务的不同阶段，它在另一个CompletionStage完成时执行一个动作或计算一个值。 一个阶段在其计算结束后完成，但这可能反过来触发其他相关阶段。 这个接口中定义的功能只有几个基本的形式：

- 上一个阶段的结果作为指定函数的参数执行函数产生新的结果。
- 上一个阶段的结果作为指定操作的参数执行指定的操作。
- 不依据上一个阶段的执行结果，只要上一个阶段完成，就执行指定的操作。

CompletionStage 部分源码：

```java
public interface CompletionStage<T> {

public CompletionStage thenApply(Function<? super T,? extends U> fn);

public CompletionStage<Void> thenAccept(Consumer<? super T> action);

public CompletionStage<Void> thenRun(Runnable action);

}
```

- thenApply：返回一个新的CompletionStage，当该阶段正常完成时，该阶段的结果作为提供的函数的参数执行。
- thenAccept：返回一个新的CompletionStage，当该阶段正常完成时，该阶段的结果作为提供的动作的参数执行。
- thenRun：返回一个新的CompletionStage，当该阶段正常完成时，执行给定的动作。

CompletionStage的阶段依赖可以是多个阶段，即一个阶段可以等待两个阶段的结果，也可以两个阶段等待一个阶段的结果。

#### 2.3 CompletableFuture

CompletableFuture实现了Future和CompletionStage接口，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合 CompletableFuture 的方法。

##### 2.3.1 创建

可以通过runAsync和supplyAsync创建CompletableFuture。supplyAsync与runAsync不同在与前者异步返回一个结果,后者是void。

```java
 public static CompletableFuture<Void> runAsync(Runnable runnable) 
 public static CompletableFuture<Void> runAsync(Runnable runnable,
 Executor executor) 
 public static CompletableFuture supplyAsync(Supplier supplier)
 public static CompletableFuture supplyAsync(Supplier supplier,
 Executor executor) 
```

Supplier是1.8定义的接口，表示结果的提供者。

```java
@FunctionalInterface
public interface Supplier<T> {
 T get();
}
```

##### 2.3.2 结果完成时处理

whenComplete、whenCompleteAsync计算完成时触发，方法不以Async结尾，意味着Action使用相同的线程执行，而Async可能会使用其它的线程去执行。

```java
 public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) 
 public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action)
 public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) 
```

BiConsumer是1.8新增的接口，表示接受两个输入参数但不返回结果的操作。

```java
@FunctionalInterface
public interface BiConsumer<T, U> {
 void accept(T t, U u);
 default BiConsumer<T, U> andThen(BiConsumer<? super T, ? super U> after) {
 Objects.requireNonNull(after);
 return (l, r) -> {
 accept(l, r);
 after.accept(l, r);
 };
 }
}
```

##### 2.3.3 任务完成后执行

thenApply/thenApplyAsync表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中。

```java
public CompletableFuture thenApply(Function<? super T,? extends U> fn);

public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn) ;

public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) ;
```

thenAccept同 thenApply 接收上一个任务的返回值作为参数，但是无返回值；thenRun 的方法没有入参，也没有返回值。

```java
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor) 
public CompletableFuture<Void> thenRun(Runnable action) 
public CompletableFuture<Void> thenRunAsync(Runnable action) 
public CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor) 
```

##### 2.3.4 CompletableFuture的组合

thenCombine/thenAcceptBoth/runAfterBoth这三个方法都是将两个CompletableFuture组合起来，只有这两个都正常执行完了才会执行某个任务，区别在于，thenCombine会将两个任务的执行结果作为方法入参传递到指定方法中，且该方法有返回值；thenAcceptBoth同样将两个任务的执行结果作为方法入参，但是无返回值；runAfterBoth没有入参，也没有返回值。

```java
public <U,V> CompletableFuture<V> thenCombine( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor) 
public CompletableFuture<Void> thenAcceptBoth( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) 
public CompletableFuture<Void> thenAcceptBothAsync( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
public CompletableFuture<Void> thenAcceptBothAsync( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor)
public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action) 
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) 
```

applyToEither/acceptEither/runAfterEither这三个方法都是将两个CompletableFuture组合起来，只要其中一个执行完了就会执行某个任务，其区别在于applyToEither会将已经执行完成的任务的执行结果作为方法入参，并有返回值；acceptEither同样将已经执行完成的任务的执行结果作为方法入参，但是没有返回值；runAfterEither没有方法入参，也没有返回值。注意两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。

```java
public CompletableFuture applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)
public CompletableFuture applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn)
public CompletableFuture applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor)
```

allOf/anyOf fallOf返回的CompletableFuture是多个任务都执行完成后才会执行，只有有一个任务执行异常，则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回null。anyOf方法是当任意一个CompletableFuture执行完后就会执行计算，计算的结果相同。

```java
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
 return andTree(cfs, 0, cfs.length - 1);
}
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
 return orTree(cfs, 0, cfs.length - 1);
 }
```

### 3 测试效果

#### 3.1 测试代码

```java
public class CompletableFutureTest {

public static String getOrderMain(){
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
        }
        return "OrderMainId";
    }

public static String getOrderResult(String orderId){
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
        }
        System.out.println("getOrderResult is doing");

return "ResultList";
    }

public static String getOrderDetail(String orderId){
        try {
            Thread.sleep(3500);
        } catch (InterruptedException e) {
        }
        System.out.println("getOrderDetail is doing");

return "DetailList";
    }

public static String getOrderAttr(String orderId){
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
        }
        System.out.println("getOrderAttr is doing");
        return "OrderAttrList";
    }

public static void main(String[] args) {
        long start = System.currentTimeMillis();
        String orderId = getOrderMain();
        getOrderResult(orderId);
        getOrderDetail(orderId);
        getOrderAttr(orderId);
        System.out.println("顺序调用耗时："+(System.currentTimeMillis()-start));

long start2 = System.currentTimeMillis();
        String orderId2 = getOrderMain();
        CompletableFuture.allOf(CompletableFuture.supplyAsync(()->getOrderResult(orderId2)),
                CompletableFuture.supplyAsync(()->getOrderDetail(orderId2)),CompletableFuture.supplyAsync(()->getOrderAttr(orderId2))).join();
        System.out.println("异步任务调用耗时："+(System.currentTimeMillis()-start2));
    }

}
```

#### 3.2 执行结果

![](//img1.jcloudcs.com/developer.jdcloud.com/c6dcba20-d2e1-47fa-9d6c-2ce93806178020220516145328.png)

### 4 小结

如果接口中的方法是顺序执行的，并且每个方法时间较长，可以通过CompletableFuture进行优化，提高接口效率。

------------

###### 自猿其说Tech-JDL京东物流技术与数据智能部
###### 作者：陈昌浩