CompletableFuture 详解（JDK1.8新特性、并发编程）

深藏阁楼爱情的钟 2023-02-14 10:29 6阅读 0赞

### **CompletableFuture 详解** ###

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

**CompletableFuture 提供了四个静态方法来创建一个异步操作。**

> 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)

入参Runnable

> @FunctionalInterface  
> public interface Runnable \{  
>     public abstract void run();  
> \}

入参Supplier

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

executor参数指定使用的线程池，没有指定会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。

runAsync不支持返回值。

supplyAsync支持返回值。

**示例**

> CompletableFuture<Void> runAsync = CompletableFuture.*runAsync*( 
>         () -> System.*out*.println("\[1\]"+Thread.*currentThread*().getId() + "--" + Thread.*currentThread*().getName()));
> 
> 
> CompletableFuture<String> supplyAsync = CompletableFuture 
>         .*supplyAsync*(() -> \{ 
>             System.*out*.println("\[2\]"+Thread.*currentThread*().getId() + "--" + Thread.*currentThread*().getName()); 
>             return "hello"; 
>         \});
> 
> 
> System.*out*.println("\[3\]"+Thread.*currentThread*().getId() + "--" + Thread.*currentThread*().getName());
> 
> //获取返回值 
> System.*out*.println(runAsync.get()); 
> System.*out*.println(supplyAsync.get());

**结果**

> \[1\]50--ForkJoinPool.commonPool-worker-1
> 
> \[3\]30--http-nio-8080-exec-1
> 
> \[2\]50--ForkJoinPool.commonPool-worker-1
> 
> null
> 
> hello

**CompletableFuture 回调方法**

用于CompletableFuture的计算完成，或者抛出异常的时候，进行下一步操作，主要方法如下：

> 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)
> 
> 
> public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn)

## **1.exceptionally** ##

当原始的CompletableFuture抛出异常的时候，会进入exceptionally方法，用来处理异常的情况

入参Function

> @FunctionalInterface 
> public interface Function<T, R> \{ 
> 
>     R apply(T t);
> 
> ...
> 
> \}

## **2.whenComplete、whenCompleteAsync** ##

whenComplete入参BiConsumer<? super T,? super Throwable>它可以处理正常的计算结果，或者异常情况

第一个参数T:  CompletableFuture计算结果

第二个参数Throwable：CompletableFuture抛出的异常

> @FunctionalInterface 
> public interface BiConsumer<T, U> \{ 
> 
>     void accept(T t, U u);
> 
> ...
> 
> \}

带async后缀函数，由线程池里面拿一个空的线程或者新启一个线程来执行回调，与执行回调的runAsync或suppleAsync线程以及执行whenCompleteAsync事件注册的线程无关。

**示例**

>     CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> \{ 
>         System.out.println("\[1\]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName()); 
>         try \{ 
>             Thread.sleep(3000); 
>         \} catch (InterruptedException e) \{ 
>             e.printStackTrace(); 
>         \} 
>         return 1; 
>     \}); 
> 
>     supplyAsync.whenComplete((data, err) -> \{ 
>         System.out.println("\[2\]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName()); 
>         if (err != null) \{ 
>             System.out.println(err.getMessage()); 
>         \} 
>         data++; 
>         System.out.println("\[2\]"+data); 
>     \}); 
>     System.out.println("\[main\]"+supplyAsync.get()); 
>     System.out.println("\[main\]"+Thread.currentThread().getId() + "--" + Thread.currentThread().getName()); 
> \}

**结果**

> \[1\]51--ForkJoinPool.commonPool-worker-1
> 
> \[2\]51--ForkJoinPool.commonPool-worker-1
> 
> \[main\]1
> 
> \[main\]31--http-nio-8080-exec-1
> 
> \[2\]2

runAsync无返回值，whenComplete入参BiConsumer<? super T,? super Throwable>的第一个参数T的类型为Void

**注意**

执行whenComplete内容的线程不一定是执行runAsync或suppleAsync的线程，也可能是执行到whenComplete事件注册的线程。

以示例中线程为例：

当线程【id：51】执行完毕的时候，主线程还没有执行到whenComplete（即whenComplete事件还没有注册），那么线程【id：51】就不会去同步执行whenComplete的回调。这个时候哪个线程执行到了whenComplete的事件注册，就由哪个线程自己来同步执行whenComplete的事件内容，示例为主线程执行。

若线程主线程在线程【id：51】执行完毕前，执行到whenComplete的事件注册，则由线程【id：51】执行whenComplete内容

## **3.thenApply ** ##

转化上一个任务的返回结果

> 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)

入参Function

> @FunctionalInterface 
> public interface Function<T, R> \{ 
> 
>     R apply(T t);
> 
> ...
> 
> \}

与whenComplete的区别

whenComplete返回的CompletableFuture类型与上一个任务类型一致，且whenComplete执行完毕时两个任务返回值一致。

thenApply返回的CompletableFuture类型、返回值可不与上一个任务类型一致

whenComplete可处理异常情况，而thenApply异常时不执行

**示例**

> CompletableFuture<List<String>> supplyAsync1 = CompletableFuture.supplyAsync(() -> Lists.newArrayList("1")); 
> CompletableFuture<String> stringCompletableFuture = supplyAsync1.thenApply((a) -> a.get(0)); 
> CompletableFuture<List<String>> integerCompletableFuture = supplyAsync1.whenComplete((data, err) -> data.add(1,"2")); 
> Thread.sleep(1000); 
> System.out.println("supplyAsync1:"+supplyAsync1.get()); 
> System.out.println("stringCompletableFuture:"+stringCompletableFuture.get()); 
> System.out.println("integerCompletableFuture:"+integerCompletableFuture.get());

**结果**

> supplyAsync1:\[1, 2\]
> 
> stringCompletableFuture:1
> 
> integerCompletableFuture:\[1, 2\]

## **4.thenAccept** ##

接收任务的处理结果，并消费处理，无返回结果。

> public CompletionStage<Void> thenAccept(Consumer<? super T> action); 
> public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action); 
> public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);

入参Consumer

> @FunctionalInterface 
> public interface Consumer<T> \{ 
> 
>     void accept(T t);
> 
> ...
> 
> ｝

**示例**

> CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> 1); 
> CompletableFuture<Void> voidCompletableFuture = supplyAsync.thenAccept(data -> System.out.println(data));

## **5.handle** ##

对上一个任务的结果进行处理。  
handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行，还可以处理异常的任务。thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法。

> public CompletionStage handle(BiFunction<? super T, Throwable, ? extends U> fn); 
> public CompletionStage handleAsync(BiFunction<? super T, Throwable, ? extends U> fn); 
> public CompletionStage handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);

入参BiFunction

> @FunctionalInterface 
> public interface BiFunction<T, U, R> \{ 
> 
>     R apply(T t, U u);
> 
> ...
> 
> \}

**h****andle****与****whenComplete****的区别**

1.都是对结果进行处理,handle有返回值,whenComplete没有返回值

2.上一个任务异常时whenComplete可判断异常但不处理异常（仍然会抛异常）,而handle可实现exceptionally的功能：对异常进行"美化"（不再抛异常）

## **6、thenRun**  ##

跟 thenAccept 方法不一样的是，不关心任务的处理结果。只要上面的任务执行完成，就开始执行 thenAccept 。

> public CompletionStage<Void> thenRun(Runnable action); 
> public CompletionStage<Void> thenRunAsync(Runnable action); 
> public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

入参Runnable

> @FunctionalInterface  
> public interface Runnable \{  
>     public abstract void run();  
> \}

该方法同 thenAccept 方法类似。不同的是上个任务处理完成后，并不会把计算的结果传给 thenRun 方法。只是处理完任务后，执行 thenAccept 的后续操作。

## **7、thenCombine 合并任务** ##

thenCombine 会把两个 CompletionStage 的任务都执行完成后，把两个任务的结果一块交给 thenCombine 来处理（类似thenApply转化操作）。

> public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
> 
> 
> public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
> 
> 
> public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn,Executor executor);

入参

参数一：CompletionStage<? extends U> other

> Public  class  CompletableFuture<T>  implements Future<T>, CompletionStage<T>

参数二：BiFunction

> @FunctionalInterface 
> public interface BiFunction<T, U, R> \{ 
>     R apply(T t, U u);
> 
> ...
> 
> ｝

## **8、thenAcceptBoth** ##

当两个CompletionStage都执行完成后，接收两个任务的处理结果，并消费处理，无返回结果（类似thenAccept）。

> public CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
> 
> 
> public CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
> 
> 
> public CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action,     Executor executor);

入参

参数一：CompletionStage<? extends U> other

> Public  class  CompletableFuture<T>  implements Future<T>, CompletionStage<T>

参数二：BiConsumer

> @FunctionalInterface 
> public interface BiConsumer<T, U> \{ 
> 
>     void accept(T t, U u);
> 
> ...
> 
> \}

## **9、applyToEither 方法** ##

两个CompletionStage，谁执行返回的结果快，就用哪个CompletionStage的结果进行下一步的转化操作（类似thenApply）。

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

入参二：Function

> @FunctionalInterface 
> public interface Function<T, R> \{ 
> 
>     R apply(T t);
> 
> ...
> 
> \}

## **10、acceptEither 方法** ##

两个CompletionStage，谁执行返回的结果快，就用哪个CompletionStage的结果进行下一步的消耗操作（类似thenAccept），无返回值。

> public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action);
> 
> 
> public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action);
> 
> 
> public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action,Executor executor);

入参二：Consumer

> @FunctionalInterface 
> public interface Consumer<T> \{ 
> 
>     void accept(T t);
> 
> ...
> 
> ｝

## **11、runAfterEither 方法** ##

两个CompletionStage，任何一个完成了都会执行下一步的操作（类似thenRun ）

> public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);
> 
> public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);
> 
> 
> public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor executor);

入参二：Runnable

> @FunctionalInterface  
> public interface Runnable \{  
>     public abstract void run();  
> \}

## **12、runAfterBoth** ##

两个CompletionStage，都完成了计算才会执行下一步的操作（类似thenRun ）

> public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable action);
> 
> public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);
> 
> 
> public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor);

入参二：Runnable

> @FunctionalInterface  
> public interface Runnable \{  
>     public abstract void run();  
> \}

## **13、thenCompose 方法** ##

thenCompose 方法允许你对两个 CompletionStage 进行流水线操作，第一个操作完成时，将其结果作为参数传递给第二个操作。

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

入参Function

> @FunctionalInterface 
> public interface Function<T, R> \{ 
> 
>     R apply(T t);
> 
> ...
> 
> \}

Function<? super T, ? extends CompletionStage> fn

返回值为CompletionStage类型

**示例**

> CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> 1); 
> CompletableFuture<String> future = supplyAsync 
>         .thenCompose(integer -> CompletableFuture.supplyAsync(() -> "结果为：" + integer \* 10));

**thenApply()和thenCompose（）的区别：**

thenapply（）是返回的是非CompletableFuture类型：它的功能相当于将CompletableFuture<T>转换成CompletableFuture。

thenCompose（）用来连接两个CompletableFuture，返回值是新的CompletableFuture：

总结：thenApply（）转换的是泛型中的类型，是同一个CompletableFuture；  
thenCompose（）用来连接两个CompletableFuture，是生成一个新的CompletableFuture。