CompletableFuture学习笔记

参考资料：https://gitee.com/phui/share-concurrent，https://www.jianshu.com/p/6bac52527ca4

一、CompletableFuture介绍

使用Future获得异步执行结果时，要么调用阻塞方法get()，要么轮询看isDone()是否为true，这两种方法都不是很好，因为主线程也会被迫等待。

从Java 8开始引入了CompletableFuture，它针对Future做了改进，可以传入回调对象，当异步任务完成或者发生异常时，自动调用回调对象的回调方法。

二、基础用法

 1 public class SmallTool {
 2     public static void sleepMillis(long millis) {
 3         try {
 4             Thread.sleep(millis);
 5         } catch (InterruptedException e) {
 6             e.printStackTrace();
 7         }
 8     }
 9 
10     public static void printTimeAndThread(String tag) {
11         String result = new StringJoiner("\t|\t")
12                 .add(String.valueOf(System.currentTimeMillis()))
13                 .add(String.valueOf(Thread.currentThread().getId()))
14                 .add(Thread.currentThread().getName())
15                 .add(tag)
16                 .toString();
17         System.out.println(result);
18     }
19 }

工具

2.1 supplyAsync（开启异步任务）

 1 public class _01_supplyAsync {
 2     public static void main(String[] args) {
 3         SmallTool.printTimeAndThread("小白进入餐厅");
 4         SmallTool.printTimeAndThread("小白点了 番茄炒蛋 + 一碗米饭");
 5 
 6         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
 7             SmallTool.printTimeAndThread("厨师炒菜");
 8             SmallTool.sleepMillis(200);
 9             SmallTool.printTimeAndThread("厨师打饭");
10             SmallTool.sleepMillis(100);
11             return "番茄炒蛋 + 米饭 做好了";
12         });
13 
14         SmallTool.printTimeAndThread("小白在打王者");
15         SmallTool.printTimeAndThread(String.format("%s ,小白开吃", cf1.join()));
16     }
17 }

supplyAsync

执行结果

（Fork/Join框架介绍）

1640426058812    |    1    |    main    |    小白进入餐厅
1640426058812    |    1    |    main    |    小白点了 番茄炒蛋 + 一碗米饭
1640426058849    |    1    |    main    |    小白在打王者
1640426058850    |    11    |    ForkJoinPool.commonPool-worker-9    |    厨师炒菜
1640426059059    |    11    |    ForkJoinPool.commonPool-worker-9    |    厨师打饭
1640426059187    |    1    |    main    |    番茄炒蛋 + 米饭 做好了 ,小白开吃

result

2.2 thenCompose（连接异步任务）

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

 1 public class _02_thenCompose {
 2     public static void main(String[] args) {
 3         SmallTool.printTimeAndThread("小白进入餐厅");
 4         SmallTool.printTimeAndThread("小白点了 番茄炒蛋 + 一碗米饭");
 5 
 6         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
 7             SmallTool.printTimeAndThread("厨师炒菜");
 8             SmallTool.sleepMillis(200);
 9             return "番茄炒蛋";
10         }).thenCompose(dish -> CompletableFuture.supplyAsync(() -> {
11             SmallTool.printTimeAndThread("服务员打饭");
12             SmallTool.sleepMillis(100);
13             return dish + " + 米饭";
14         }));
15 
16         SmallTool.printTimeAndThread("小白在打王者");
17         SmallTool.printTimeAndThread(String.format("%s 好了,小白开吃", cf1.join()));
18     }
19 }

thenCompose

执行结果

1640426372421    |    1    |    main    |    小白进入餐厅
1640426372421    |    1    |    main    |    小白点了 番茄炒蛋 + 一碗米饭
1640426372461    |    11    |    ForkJoinPool.commonPool-worker-9    |    厨师炒菜
1640426372463    |    1    |    main    |    小白在打王者
1640426372673    |    12    |    ForkJoinPool.commonPool-worker-2    |    服务员打饭
1640426372803    |    1    |    main    |    番茄炒蛋 + 米饭 好了,小白开吃

result

2.3 thenCombine（合并异步任务）

thenCombine 会把两个 CompletionStage 的任务都执行完成后，把两个任务的结果一块交给 thenCombine 来处理。

 1 public class _03_thenCombine {
 2     public static void main(String[] args) {
 3         SmallTool.printTimeAndThread("小白进入餐厅");
 4         SmallTool.printTimeAndThread("小白点了 番茄炒蛋 + 一碗米饭");
 5 
 6         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
 7             SmallTool.printTimeAndThread("厨师炒菜");
 8             SmallTool.sleepMillis(200);
 9             return "番茄炒蛋";
10         }).thenCombine(CompletableFuture.supplyAsync(() -> {
11             SmallTool.printTimeAndThread("服务员蒸饭");
12             SmallTool.sleepMillis(300);
13             return "米饭";
14         }), (dish, rice) -> {
15             SmallTool.printTimeAndThread("服务员打饭");
16             SmallTool.sleepMillis(100);
17             return String.format("%s + %s 好了", dish, rice);
18         });
19 
20         SmallTool.printTimeAndThread("小白在打王者");
21         SmallTool.printTimeAndThread(String.format("%s ,小白开吃", cf1.join()));
22     }
23 
24     /**
25      * 用 applyAsync 也能实现
26      */
27     private static void applyAsync() {
28         SmallTool.printTimeAndThread("小白进入餐厅");
29         SmallTool.printTimeAndThread("小白点了 番茄炒蛋 + 一碗米饭");
30 
31         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
32             SmallTool.printTimeAndThread("厨师炒菜");
33             SmallTool.sleepMillis(200);
34             return "番茄炒蛋";
35         });
36         CompletableFuture race = CompletableFuture.supplyAsync(() -> {
37             SmallTool.printTimeAndThread("服务员蒸饭");
38             SmallTool.sleepMillis(300);
39             return "米饭";
40         });
41         SmallTool.printTimeAndThread("小白在打王者");
42 
43         String result = String.format("%s + %s 好了", cf1.join(), race.join());
44         SmallTool.printTimeAndThread("服务员打饭");
45         SmallTool.sleepMillis(100);
46 
47         SmallTool.printTimeAndThread(String.format("%s ,小白开吃", result));
48     }
49 }

thenCombine

执行结果

1640426656536    |    1    |    main    |    小白进入餐厅
1640426656537    |    1    |    main    |    小白点了 番茄炒蛋 + 一碗米饭
1640426656581    |    11    |    ForkJoinPool.commonPool-worker-9    |    厨师炒菜
1640426656582    |    12    |    ForkJoinPool.commonPool-worker-2    |    服务员蒸饭
1640426656586    |    1    |    main    |    小白在打王者
1640426656883    |    12    |    ForkJoinPool.commonPool-worker-2    |    服务员打饭
1640426657033    |    1    |    main    |    番茄炒蛋 + 米饭 好了 ,小白开吃

result

2.4 thenApply（任务的后置处理）

当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。

thenApply是执行当前任务的线程执行继续执行 thenApply的任务。

thenApplyAsync是把 thenApplyAsync这个任务继续提交给线程池来进行执行。

public class _01_thenApply {
    public static void main(String[] args) {
        SmallTool.printTimeAndThread("小白吃好了");
        SmallTool.printTimeAndThread("小白 结账、要求开发票");

        CompletableFuture invoice = CompletableFuture.supplyAsync(() -> {
            SmallTool.printTimeAndThread("服务员收款 500元");
            SmallTool.sleepMillis(100);
            return "500";
        }).thenApplyAsync(money -> {
            SmallTool.printTimeAndThread(String.format("服务员开发票 面额 %s元", money));
            SmallTool.sleepMillis(200);
            return String.format("%s元发票", money);
        });

        SmallTool.printTimeAndThread("小白 接到朋友的电话，想一起打游戏");

        SmallTool.printTimeAndThread(String.format("小白拿到%s，准备回家", invoice.join()));
    }
}

thenApply

执行结果

1640427311931    |    1    |    main    |    小白吃好了
1640427311932    |    1    |    main    |    小白 结账、要求开发票
1640427311974    |    11    |    ForkJoinPool.commonPool-worker-9    |    服务员收款 500元
1640427311974    |    1    |    main    |    小白 接到朋友的电话，想一起打游戏
1640427312100    |    11    |    ForkJoinPool.commonPool-worker-9    |    服务员开发票 面额 500元
1640427312307    |    1    |    main    |    小白拿到500元发票，准备回家

result

2.5 applyToEither(获取最先完成的任务)

两个CompletionStage，返回执行快的结果，用快的CompletionStage的结果进行下一步的转化操作。

 1 public class _02_applyToEither {
 2     public static void main(String[] args) {
 3         SmallTool.printTimeAndThread("小白走出餐厅，来到公交站");
 4         SmallTool.printTimeAndThread("等待 700路 或者 800路 公交到来");
 5 
 6         CompletableFuture bus = CompletableFuture.supplyAsync(() -> {
 7             SmallTool.printTimeAndThread("700路公交正在赶来");
 8             SmallTool.sleepMillis(100);
 9             return "700路到了";
10         }).applyToEither(CompletableFuture.supplyAsync(() -> {
11             SmallTool.printTimeAndThread("800路公交正在赶来");
12             SmallTool.sleepMillis(200);
13             return "800路到了";
14         }), firstComeBus -> firstComeBus);
15 
16         SmallTool.printTimeAndThread(String.format("%s,小白坐车回家", bus.join()));
17     }
18 }

applyToEither

执行结果

1640427683299    |    1    |    main    |    小白走出餐厅，来到公交站
1640427683300    |    1    |    main    |    等待 700路 或者 800路 公交到来
1640427683363    |    11    |    ForkJoinPool.commonPool-worker-9    |    700路公交正在赶来
1640427683363    |    12    |    ForkJoinPool.commonPool-worker-2    |    800路公交正在赶来
1640427683498    |    1    |    main    |    700路到了,小白坐车回家

result

2.6 exceptionally（异常处理）

当CompletableFuture的抛出异常的时候，可以执行特定的Action。

 1 public class _03_exceptionally {
 2     public static void main(String[] args) {
 3         SmallTool.printTimeAndThread("张三走出餐厅，来到公交站");
 4         SmallTool.printTimeAndThread("等待 700路 或者 800路 公交到来");
 5 
 6         CompletableFuture bus = CompletableFuture.supplyAsync(() -> {
 7             SmallTool.printTimeAndThread("700路公交正在赶来");
 8             SmallTool.sleepMillis(100);
 9             return "700路到了";
10         }).applyToEither(CompletableFuture.supplyAsync(() -> {
11             SmallTool.printTimeAndThread("800路公交正在赶来");
12             SmallTool.sleepMillis(200);
13             return "800路到了";
14         }), firstComeBus -> {
15             SmallTool.printTimeAndThread(firstComeBus);
16             if (firstComeBus.startsWith("700")) {
17                 throw new RuntimeException("撞树了……");
18             }
19             return firstComeBus;
20         }).exceptionally(e -> {
21             SmallTool.printTimeAndThread(e.getMessage());
22             SmallTool.printTimeAndThread("小白叫出租车");
23             return "出租车 叫到了";
24         });
25 
26         SmallTool.printTimeAndThread(String.format("%s,小白坐车回家", bus.join()));
27     }
28 }

exceptionally

三、API使用详解

3.1 runAsync 和 supplyAsync（开启异步任务）

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

runAsync方法不提供返回值。
supplyAsync提供返回值。

没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。

public static CompletableFuture runAsync(Runnable runnable)
public static CompletableFuture runAsync(Runnable runnable, Executor executor)
public static  CompletableFuture supplyAsync(Supplier supplier)
public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor)

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

thenCompose前面和后面的两段代码是两个任务（厨师、服务员B）。区别如下：

thenCompose ：是执行当前任务的线程执行继续执行。（效果如厨师和服务员A在同一个任务。）
thenComposeAsync：是把 thenComposeAsync这个任务继续提交给线程池来进行执行。厨师和服务员A在两个任务

 1 public class _02_thenCompose {
 2     public static void main(String[] args) {
 3         CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
 4             SmallTool.printTimeAndThread("厨师炒菜");
 5             SmallTool.sleepMillis(200);
 6             return "番茄炒蛋";
 7         }).thenCompose(dish -> {
 8             SmallTool.printTimeAndThread("服务员A 准备打饭，但是被领导叫走，打饭交接给服务员B");
 9 
10             return CompletableFuture.supplyAsync(() -> {
11                 SmallTool.printTimeAndThread("服务员B 打饭");
12                 SmallTool.sleepMillis(100);
13                 return dish + " + 米饭";
14             });
15         });
16 
17         SmallTool.printTimeAndThread(cf1.join()+"好了，开饭");
18     }
19 }

example

thenCompose执行结果

1640438004703    |    12    |    ForkJoinPool.commonPool-worker-9    |    厨师炒菜
1640438006715    |    12    |    ForkJoinPool.commonPool-worker-9    |    服务员A 准备打饭，但是被领导叫走，打饭交接给服务员B
1640438006717    |    13    |    ForkJoinPool.commonPool-worker-2    |    服务员B 打饭
1640438006827    |    1    |    main    |    番茄炒蛋 + 米饭好了，开饭

thenCompose执行结果

thenComposeAsync执行结果

1640438345289    |    12    |    ForkJoinPool.commonPool-worker-9    |    厨师炒菜
1640438347301    |    13    |    ForkJoinPool.commonPool-worker-2    |    服务员A 准备打饭，但是被领导叫走，打饭交接给服务员B
1640438347304    |    12    |    ForkJoinPool.commonPool-worker-9    |    服务员B 打饭
1640438347410    |    1    |    main    |    番茄炒蛋 + 米饭好了，开饭

thenComposeAsync执行结果

3.3 thenCombine（合并异步任务）、thenAcceptBoth、runAfterBoth

thenCombine 会把两个 CompletionStage 的任务都执行完成后，把两个任务的结果一块交给 thenCombine 来处理。

thenCombine 可以得到前面两个任务的结果，经处理，返回一个结果
thenAcceptBoth可以得到前面两个任务的结果，经处理，无返回结果
runAfterBoth不关心前面两个任务的结果，经处理，无返回值

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

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

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

3.4 thenApply（任务的后置处理）、thenAccept、thenRun

当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。

thenApply是执行当前任务的线程执行继续执行。作用类似于把后面的代码块放到上一个任务的末尾。会交给一个线程去运行。

thenApplyAsync是把 thenApplyAsync这个任务继续提交给线程池来进行执行。将两部分代码当做两个独立任务。在第二个任务开始前，需要把第一个任务处理完，将第一个任务的结果交给第二个任务。

thenApply会接受上个任务的结果参数，经处理，返回结果
thenAccept会接受上个任务的结果参数，经处理，无返回值
thenRun不关心上个任务的结果参数，经处理，无返回值

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)
    
public CompletionStage thenAccept(Consumer<? super T> action);
public CompletionStage thenAcceptAsync(Consumer<? super T> action);
public CompletionStage thenAcceptAsync(Consumer<? super T> action,Executor executor);

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

Function<? super T,? extends U> T：上一个任务返回结果的类型 U：当前任务的返回值类型

3.5 applyToEither(获取最先完成的任务)、acceptEither 、runAfterEither

两个CompletionStage，返回执行快的结果，用快的CompletionStage的结果进行下一步的转化操作。

applyToEither获取前两个任务中最先执行完的任务结果，经处理，返回一个结果
acceptEither获取前两个任务中最先执行完的任务结果，经处理，无返回值
runAfterEither不关心前两个任务中最先执行完的任务结果，经处理，无返回值

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

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

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

3.6 exceptionally（异常处理）、handle、whenComplete

当CompletableFuture的计算结果完成，或者抛出异常的时候，可以执行特定的Action。

exceptionally处理前面任务的异常，并把异常修正为正常值
handle前面的程序发生异常，那么就接收到异常；前面的结果正常则正常处理。handle会返回一个结果，让后续方法正常执行。（handle 是执行任务完成时对结果的处理。handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行，还可以处理异常的任务。thenApply 只可以执行正常的任务，任务出现异常则不执行 thenApply 方法。）
whenComplete与handle相同，但无返回值。

public CompletableFuture exceptionally(Functionextends T> fn)
    
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);

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

Action的类型是BiConsumer<? super T,? super Throwable>它可以处理正常的计算结果，或者异常情况。

whenComplete 和 whenCompleteAsync 的区别： whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。 whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。

四、参考

https://www.jianshu.com/p/6bac52527ca4

https://gitee.com/phui/share-concurrent

java

CompletableFuture学习笔记

一、CompletableFuture介绍

二、基础用法

2.1 supplyAsync（开启异步任务）

2.2 thenCompose（连接异步任务）

2.3 thenCombine（合并异步任务）

2.4 thenApply（任务的后置处理）

2.5 applyToEither(获取最先完成的任务)

2.6 exceptionally（异常处理）

三、API使用详解

3.1 runAsync 和 supplyAsync（开启异步任务）

3.2 thenCompose（连接异步任务）

3.3 thenCombine（合并异步任务）、thenAcceptBoth、runAfterBoth

3.4 thenApply（任务的后置处理）、thenAccept、thenRun

3.5 applyToEither(获取最先完成的任务)、acceptEither 、runAfterEither

3.6 exceptionally（异常处理）、handle、whenComplete

四、参考

相关

JavaWeb的三大作用域

java中常用数组copy方法

6.Java方法

Java web开发：从零到 Run 一个现有的 Spring Boot 前后端分离项目（前端Vue）

javascript基础知识(29) 实现继承的方式

java 正则表达式

JAVA List<Object[]>取值问题

初次接触Java感受

99乘法表 java for循环

基于java.net.socket包的对象传递

Idea Java项目文件夹后面提示百分数怎么去掉

Java8新特性

标签