什么，你还不会用CompletableFuture？

痛定思痛。 2022-12-04 03:25 285阅读 0赞

如果对JAVA8中的Supplier、Function等类不熟悉的话，可以先移步我的另外一篇文章[快速理解Consumer、Supplier、Predicate与Function][Consumer_Supplier_Predicate_Function]

上一篇我们讲了Future机制，有兴趣的可以参考[谈谈Runnable、Future、Callable、FutureTask之间的关系][Runnable_Future_Callable_FutureTask]

但Future机制，还不那么灵活，比如怎么去利用Future机制描述两个任务串行执行，又或是两个任务并行执行，又或是只关心最先执行结束的任务结果。

Future机制在一定程度上都无法快速地满足以上需求，CompletableFuture便应运而生了。

本片会介绍CompletableFuture的api，并用一些示例演示如何去使用。

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

## 1. 创建一个异步任务 ##

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

supplyAsync与runAsync的区别在于：supplyAsync有返回值，而runAsync没有返回值

带Executor参数的构造函数，则使用线程池中的线程执行异步任务（线程池可以参考[说说线程池][Link 1]）

不带Executor参数的构造函数，则使用ForkJoinPool.commonPool()中的线程执行异步任务（Fork/Join框架可以参考[谈谈并行流parallelStream][parallelStream]）

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

### 1.1 示例：使用supplyAsync创建一个有返回值的异步任务 ###

public class Case1 {
 
 public static void main(String[] args) throws Exception {
 
 CompletableFuture<Integer> completableFuture=CompletableFuture.supplyAsync(()->{
 try {
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return 1;
 });
 //该方法会一直阻塞
 Integer result = completableFuture.get();
 System.out.println(result);
 }
 
 }

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

## 2. 异步任务的回调 ##

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

whenComplete开头的方法在计算任务完成（包括正常完成与出现异常）之后会回调

而exceptionally则只会在计算任务出现异常时才会被回调

如何确定哪个线程去回调whenComplete，比较复杂，先略过。

而回调whenCompleteAsync的线程比较简单，随便拿一个空闲的线程即可，后缀是Async的方法同理。

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

### 2.1 示例：计算出现异常，使用whenComplete与exceptionally进行处理 ###

package com.qcy.testCompleteableFuture;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.stream.IntStream;
 
 /**
 * @author qcy
 * @create 2020/09/07 17:40:44
 */
 public class Case2 {
 
 public static void main(String[] args) throws Exception {
 
 CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
 try {
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("执行supplyAsync的线程:" + Thread.currentThread().getName());
 int i = 1 / 0;
 return 1;
 });
 
 completableFuture.whenComplete(new BiConsumer<Integer, Throwable>() {
 @Override
 public void accept(Integer integer, Throwable throwable) {
 System.out.println("执行whenComplete的线程:" + Thread.currentThread().getName());
 if (throwable == null) {
 System.out.println("计算未出现异常,结果:" + integer);
 }
 }
 });
 
 completableFuture.exceptionally(new Function<Throwable, Integer>() {
 @Override
 public Integer apply(Throwable throwable) {
 //出现异常时,则返回一个默认值
 System.out.println("计算出现异常,信息:" + throwable.getMessage());
 return -1;
 }
 });
 
 System.out.println(completableFuture.get());
 }
 
 }

输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70][]

当然，CompletableFuture内的各种方法是支持链式调用与Lambda表达式的，我们进行如下改写：

public static void main(String[] args) throws Exception {
 
 CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
 try {
 Thread.sleep(2000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("执行supplyAsync的线程:" + Thread.currentThread().getName());
 int i = 1 / 0;
 return 1;
 }).whenComplete((integer, throwable) -> {
 System.out.println("执行whenComplete的线程:" + Thread.currentThread().getName());
 if (throwable == null) {
 System.out.println("计算未出现异常,结果:" + integer);
 }
 }).exceptionally(throwable -> {
 //出现异常时,则返回一个默认值
 System.out.println("计算出现异常,信息:" + throwable.getMessage());
 return -1;
 });
 
 System.out.println("计算结果:" + completableFuture.get());
 }

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

## 3. 任务串行化执行 ##

public CompletableFuture thenApply(Function<? super T,? extends U> fn);
 
 public CompletableFuture<Void> thenRun(Runnable action);
 
 public CompletableFuture<Void> thenAccept(Consumer<? super T> action);
 
 public CompletableFuture handle(BiFunction<? super T, Throwable, ? extends U> fn);
 
 public CompletableFuture thenCompose(Function<? super T, ? extends CompletionStage> fn);

thenApply，依赖上一次任务执行的结果，参数中的Function<? super T,? extends U>，T代表上一次任务返回值的类型，U代表当前任务返回值的类型，当上一个任务没有出现异常时，thenApply才会被调用

thenRun，不需要知道上一个任务的返回结果，只是在上一个任务执行完成之后开始执行Runnable

thenAccept，依赖上一次任务的执行结果，因为入参是Consumer，所以不返回任何值。

handle和thenApply相似，不过当上一个任务出现异常时，能够执行handle，却不会去执行thenApply

thenCompose，传入一次任务执行的结果，返回一个新的CompleteableFuture对象

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

### 3.1 示例：使用串行化任务分解两数相乘并输出 ###

package com.qcy.testCompleteableFuture;
    
    import java.util.concurrent.CompletableFuture;
    
    /**
     * @author qcy
     * @create 2020/09/07 17:40:44
     */
    public class Case4 {
    
        public static void main(String[] args) {
            
            CompletableFuture.supplyAsync(() -> 2)
                    .thenApply(num -> num * 3)
                    .thenAccept(System.out::print);
        }
    
    }

很显然，输出为6

### 3.2 示例：使用串行化任务并且模拟出现异常 ###

package com.qcy.testCompleteableFuture;
    
    import java.util.concurrent.CompletableFuture;
    import java.util.function.BiFunction;
    
    /**
     * @author qcy
     * @create 2020/09/07 17:40:44
     */
    public class Case4 {
    
        public static void main(String[] args) {
    
            CompletableFuture.supplyAsync(() -> 2)
                    .thenApply(num -> num / 0)
                    .thenApply(result -> result * 3)
                    .handle((integer, throwable) -> {
                        if (throwable == null) {
                            return integer;
                        } else {
                            throwable.printStackTrace();
                            return -1;
                        }
                    }).thenAccept(System.out::print);
        }
    
    }

最终会输出-1

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

## 4. 任务同时执行，且都需要执行完成 ##

public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,
 BiFunction<? super T,? super U,? extends V> fn);
 
 public CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,
 BiConsumer<? super T, ? super U> action);
 
 public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,Runnable action);
 
 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs);

thenCombine，**合并两个任务，两个任务可以同时执行，都执行成功后，执行最后的BiFunction操作**。其中T代表第一个任务的执行结果类型，U代表第二个任务的执行结果类型，V代表合并的结果类型

thenAcceptBoth，和thenCombine特性用法都及其相似，唯一的区别在于thenAcceptBoth进行一个消费，没有返回值

runAfterBoth，两个任务都执行完成后，但不关心他们的返回结构，然后去执行一个Runnable。

allOf，当所有的任务都执行完成后，返回一个CompletableFuture

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

### 4.1 示例：使用thenCombine合并任务 ###

package com.qcy.testCompleteableFuture;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
 /**
 * @author qcy
 * @create 2020/09/07 17:40:44
 */
 public class Case5 {
 
 public static void main(String[] args) throws Exception {
 
 CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务1开始");
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务1结束");
 return 2;
 });
 
 CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务2开始");
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务2结束");
 return 3;
 });
 
 CompletableFuture<Integer> completableFuture = cf1.thenCombine(cf2, (result1, result2) -> result1 * result2);
 System.out.println("计算结果:" + completableFuture.get());
 }
 
 }

输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70 1][]

可以看到两个任务确实是同时执行的

当然，熟练了之后，直接使用链式操作，代码如下：

package com.qcy.testCompleteableFuture;
 
 import java.util.concurrent.CompletableFuture;
 
 /**
 * @author qcy
 * @create 2020/09/07 17:40:44
 */
 public class Case6 {
 
 public static void main(String[] args) throws Exception {
 
 CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务1开始");
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务1结束");
 return 2;
 }).thenCombine(CompletableFuture.supplyAsync(() -> {
 System.out.println("任务2开始");
 try {
 Thread.sleep(2000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务2结束");
 return 3;
 }), (result1, result2) -> result1 * result2);
 
 System.out.println("计算结果:" + completableFuture.get());
 }
 
 }

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

## 5. 任务同时执行，且只取最先完成的那个任务 ##

public CompletableFuture applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn);
 
 public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action);
 
 public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,Runnable action);
 
 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs);

applyToEither，最新执行完任务，将其结果执行Function操作，其中T是最先执行完的任务结果类型，U是最后输出的类型

acceptEither，最新执行完的任务，将其结果执行消费操作

runAfterEither，任意一个任务执行完成之后，执行Runnable操作

anyOf，多个任务中，返回最先执行完成的CompletableFuture

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

### 5.1 示例：两个任务同时执行，打印最先完成的任务的结果 ###

package com.qcy.testCompleteableFuture;
 
 import java.util.concurrent.CompletableFuture;
 
 /**
 * @author qcy
 * @create 2020/09/07 17:40:44
 */
 public class Case7 {
 
 public static void main(String[] args) throws Exception {
 
 CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务1开始");
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务1结束");
 return 2;
 }).acceptEither(CompletableFuture.supplyAsync(() -> {
 System.out.println("任务2开始");
 try {
 Thread.sleep(2000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务2结束");
 return 3;
 }), result -> System.out.println(result));
 
 //等待CompletableFuture返回，防止主线程退出
 completableFuture.join();
 }
 
 }

输出：

![20200908140516142.png][]

可以看得到，任务2结束后，直接不再执行任务1的剩余代码

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

### 5.2 示例：多个任务同时执行，打印最先完成的任务的结果 ###

package com.qcy.testCompleteableFuture;
 
 import java.util.concurrent.CompletableFuture;
 
 /**
 * @author qcy
 * @create 2020/09/07 17:40:44
 */
 public class Case8 {
 
 public static void main(String[] args) throws Exception {
 
 CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务1开始");
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务1结束");
 return 2;
 });
 
 CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务2开始");
 try {
 Thread.sleep(2000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务2结束");
 return 3;
 });
 
 CompletableFuture<Integer> cf3 = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务3开始");
 try {
 Thread.sleep(4000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务3结束");
 return 4;
 });
 
 CompletableFuture<Object> firstCf = CompletableFuture.anyOf(cf1, cf2, cf3);
 System.out.println(firstCf.get());
 }
 
 }

输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70 2][]

[Consumer_Supplier_Predicate_Function]: https://blog.csdn.net/qq_33591903/article/details/102948344?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522159947293819725254062819%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=159947293819725254062819&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_v2~rank_blog_v1-1-102948344.pc_v2_rank_blog_v1&utm_term=Supplier&spm=1018.2118.3001.4187
[Runnable_Future_Callable_FutureTask]: https://blog.csdn.net/qq_33591903/article/details/108422098
[Link 1]: https://blog.csdn.net/qq_33591903/article/details/106239102?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522159947332319725222413888%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=159947332319725222413888&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_v2~rank_blog_v1-2-106239102.pc_v2_rank_blog_v1&utm_term=%E7%BA%BF%E7%A8%8B%E6%B1%A0&spm=1018.2118.3001.4187
[parallelStream]: https://blog.csdn.net/qq_33591903/article/details/107990065
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70]: /images/20221123/11b0dcbcef2c4a838f7c0d693f02c054.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70 1]: /images/20221123/c523905235d343f683c4bd0f06b89fc6.png
[20200908140516142.png]: /images/20221123/97e27ff2f809453c8b13d2b17afb2ec0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTkxOTAz_size_16_color_FFFFFF_t_70 2]: /images/20221123/a268c10e73f64623bac1e974168bb1f8.png