CompletableFuture学习 

1
2
3
4
感谢
https://www.zhihu.com/question/318472639/answer/652254785
https://zhuanlan.zhihu.com/p/344431341
https://www.bilibili.com/video/BV1ui4y1T7xf

0、准备

ListenableFuture 是由 Google Guava工具包提供的 Future 扩展类，随后，JDK 在 1.8 版本中马上也提供了类似这样的类，在 9 中进一步增强，这就是 CompletableFuture。CompletableFuture 同样实现了 Future 接口，并在此基础上进行了丰富的扩展，弥补了 Future 的局限性，是对 Future 的扩展和增强，同时 CompletableFuture 实现了对任务编排的能力。

1、get/join

1.1、get

get 有两种类型的重载，继承自父接口 Future 中两种方法，详情可见 Future 接口相关内容。

  • 无参数 get：获得异步执行结果，如果任务一直不返回会一直阻塞，一般不建议使用；
1
public T get()
1
2
3
4
// 这种写法主线程会一直阻塞，因为这种方式创建的future从未完成，
// 可以打个断点看看，future的状态一直是not completed。
CompletableFuture<Integer> future = new CompletableFuture<>();
Integer integer = future.get();
  • 有参数 get：获得异步执行结果，指定超时时间，超时抛出异常；
1
public T get(long timeout, TimeUnit unit)

demo（github）

1.2、getNow

获取任务的执行结果，但不会产生阻塞。如果任务还没执行完成，那么就会返回传入的 valueIfAbsent 参数值，如果执行完成了，就会返回任务执行的结果。

1
public T getNow(T valueIfAbsent)

demo

1.3、join

  • 返回的类型就是 CompletableFuture 的泛型，即 supplier 的返回值；
  • 实际上就是 Future 中 get() 的升级版，等待任务结束，返回任务的结果；
  • 区别：get() 会抛出检查时的异常，可被捕获进行处理或直接抛出，join() 会抛出未经检查的异常。
1
public T join()

2、supplyAsync/runAsync

在学习异步任务的编排前，首先准备打印当前时间和线程名的工具类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class SmallTool {
    public static void sleepMillis(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public static void printTimeAndThread(String tag) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        String result = new StringJoiner("\t|\t")
                .add(sdf.format(new Date()))
                .add(String.valueOf(Thread.currentThread().getId()))
                .add(Thread.currentThread().getName())
                .add(tag).toString();
        System.out.println(result);
    }
}

功能：执行异步任务，可以初始化 cf。

区别：supplyAsync 用于有返回值的任务，runAsync 用于没有返回值的任务，不需要返回值时使用 runAsync 即可。

1
2
3
4
5
6
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) {
    return asyncSupplyStage(screenExecutor(executor), supplier);
}

参数：

  • 供应者 supplier，是一个函数式接口；（详见函数式接口）

  • supplyAsync 后括号中的 supplier 会去另一个线程中执行；

  • Executor 参数可以手动指定线程池，否则默认使用 ForkJoinPool.commonPool() 系统级公共线程池

注意：这些线程都是 Daemon 线程，主线程结束 Daemon 线程不结束，只有 JVM 关闭时，生命周期才终止。

demo

可以看到，在小白点餐后，厨师的操作和小白后续的操作几乎是同时进行的。

res

3、thenCompose

thenCompose：连接，在异步执行线程1的基础上再开一个新的线程，使用 thenCompose 会将之前线程的结果交给下一个的线程，前一个线程运行完了有结果后下一个线程才能触发。

1
2
3
public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
    return uniComposeStage(null, fn);
}

参数：

  • Function：函数式接口，传入T，返回 R；
  • CompletionStage：CompletableFuture 的父接口，使用 CompletableFuture 替代即可。

demo

在下面例子中入参 dish 是厨师任务的返回值，只有厨师炒菜完成后服务员才会开始打饭，然后进入第二个 CompletableFuture，再开启一个线程运行后返回，观察运行结果也可知，炒菜和打饭不在同一个线程上运行。

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

4、thenCombine

thenCombine：合并，将任务1和任务2一起执行，然后将任务1和任务2的返回结果用作任务3的输入。

demo

demo 中 dish 和 rice 作为线程 3 的输入，只有等线程 1 和线程 2 运行完毕后线程 3 才能开始运行，观察运行结果可知，线程 2 和线程 3 使用的使用同一个线程。

res

thenAcceptBoth/runAfterBoth

thenAcceptBoth/runAfterBoth 与 thenCombine相比的区别？

  • thenCombine：需要前边任务的结果，有返回值；

  • thenAcceptBoth：得到前边两个任务的参数之后直接内部消化，没有返回值；

  • runAfterBoth：不需要前边任务的结果也不需要返回值。

5、thenApply/thenApplyAsync

thenApply：将前面异步任务的结果交给后面的 Function，不会处理异常，异常会被直接抛出，交给上一层处理。

  • thenApply 使用同一线程运行前后两个任务；
  • thenApplyAsync 使用不同的线程运行前后两个任务，类似之前的 thenCompose。

demo

res

thenApply：会在相同的线程中执行

thenApplyAsync：会在不同的线程中执行

thenAccept/thenRun

thenAccept/thenRun 与 thenApply 相比的区别？

  • thenApply 接收前边任务的参数，有返回值；

  • thenAccept 会接收前边任务的参数，但是没有返回值；

  • thenRun 不接收前边任务的参数，也没有返回值；

6、applyToEither

applyToEither：两个任务一起运行，谁先运行完返回谁。

demo

demo 中 800 路的线程运行时间更快一些，会提前完成。

res

acceptEither/runAfterEither

acceptEither/runAfterEither 和 applyToEither 的区别？

  • applyToEither：获取前两个任务最先完成的任务的结果，处理后得到返回值；
  • acceptEither：得到结果，没有返回值；
  • runAfterEither：不关心结果，也不关心返回值。

7、exceptionally

exceptionally：在链式调用中如果出现异常，就会调用 exceptionally 块中的内容。

exceptionally 不一定放在最后，放在前面也可以。

demo

运行结果

testEnd：exceptionally 放在程序最后

在程序中，700 路会撞车，并且提前到，小白一定会上 700；

如果调整时间，让 800 提前到，小白就会上 800，就不会出现异常。

testMiddle：exceptionally 放在程序中间

handle/whenComplete

handle/whenComplete 与 exceptionally 的区别？

  • handle：参数是 BiFunction
    • 如果前面的程序正常执行，那么会接收到正常执行的结果；
    • 如果出现异常，就会接收到异常；
    • 无论正常或异常，都会返回一个结果，让后面的程序继续执行。
  • whenComplete：与 handle 类似，没有返回值。
    • whenCompleteAsync：使用异步处理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 开启一个异步方法
CompletableFuture<List> future = CompletableFuture.supplyAsync(() -> {
    List<String> list = new ArrayList<>();
    list.add("语文");
    list.add("数学");
    // 获取得到今天的所有课程
    return list;
});
// 使用handle()方法接收list数据和error异常
CompletableFuture<Integer> future2 = future.handle((list,error)-> {
    // 如果报错，就打印出异常
    error.printStackTrace();
    // 如果不报错，返回一个包含Integer的全新的CompletableFuture
    return list.size();
    // 注意这里的两个CompletableFuture包含的返回类型不同
});

8、complete/completeExceptionally

complete：主动触发当前异步任务的完成。

  • 调用此方法时如果你的任务已经完成，那么方法就会返回 false；
  • 如果任务没完成，就会返回 true，并且其它线程获取到的任务的结果就是 complete 的参数值。

completeExceptionally：跟 complete 的作用差不多，complete 是正常结束任务，返回结果，而completeExceptionally 就是触发任务执行的异常。

1
2
3
4
5
6
7
8
9
10
11
12
public class _07_complete {

    public void completeTest() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = new CompletableFuture<>();
        try {
            future.complete("test success");
        } catch (Exception e) {
            future.completeExceptionally(e);
        }         
        System.out.println(future.get());
    }
}

9、allOf/anyOf

allOf：多个任务都执行完成后才会执行。

  • 只有一个任务执行异常，则返回的 CompletableFuture 执行 get 方法时会抛出异常；
  • 如果都是正常执行，则 get 返回 null。

anyOf：多个任务中只要有一个任务执行完成，get 就会返回已经执行完成任务的结果。

  • 成功执行就返回成功执行的结果；
  • 有异常就返回异常。

举例如下：

  • allOf 等待所有任务执行完成才执行 cfAll，如果有一个任务异常终止，则 cfAll.get() 时会抛出异常，都是正常执行，cfAll.get() 返回 null；
  • anyOf 是只有一个任务执行完成，无论是正常执行或者执行异常，都会执行 cfAny，cfAny.get()的结果就是已执行完成的任务的执行结果。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
public class _08_allOfAnyOf {

    public void testAllOfAnyOf() throws ExecutionException, InterruptedException {

        // cf1会提前完成
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            SmallTool.sleepMillis(10);
            SmallTool.printTimeAndThread("cf1 working...");
            return "cf1 任务完成";
        });

        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            SmallTool.sleepMillis(500);
            SmallTool.printTimeAndThread("cf2 working...");
            return "cf2 任务完成";
        });

        CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> {
            SmallTool.sleepMillis(500);
            SmallTool.printTimeAndThread("cf3 working...");
//        int i = 1 / 0;
            return "cf3 任务完成";
        });

        System.out.println("===========cfAll===========");
        /**
         * 有异常就抛异常
         * 没异常get()会返回null
         */
        CompletableFuture<Void> cfAll = CompletableFuture.allOf(cf1, cf2, cf3);
        System.out.println("cfAll->" + cfAll.get());
        System.out.println("===========cfAny===========");
        /**
         * cf3会出现异常，但是cf1提前完成，anyOf会提前返回结果
         * 此时程序直接结束，不会抛出异常
         */
        CompletableFuture<Object> cfAny = CompletableFuture.anyOf(cf1, cf2, cf3);
        System.out.println("cfAny->" + cfAny.get());

    }
}

10、规律

规律

1
2
3
xxx(arg)
xxxAsync(arg)
xxxAsync(arg,Executor)

  • 一般情况下，大多数方法的构造方法都会有一个带有两个参数的重载构造方法，会增加一个线程池 Executor 参数，如 supplyAsync：

    1
    2
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

  • 有方法 xxx，就会有 xxxAsync 方法，例如 thenCompose/thenComposeAsync、thenApply/thenApplyAsync、thenCombine/thenCombineAsync。

Async

  • 不加的话，在 CompletableFuture 看来，前后两段代码就是同一个任务，运行时会把后面的代码放到前一段代码的末尾封装成一个任务交给一个线程运行；

  • async：异步的，加上这个单词的话，CompletableFuture 会把前后两段代码当成是独立的两个任务，分别放到两个线程中执行。

  • 以下面的代码为例，在 thenComposeAsync 不再调用 CompletableFuture，而是直接返回 CompletableFuture，这样在 return 之前还可以进行其他的任务；

    • 使用 thenCompose，因为前后两段代码本身就不在一个线程中执行，使用 return 之后可以看得更加明显。线程 2 会被添加到线程 1 的末尾，划分到同一个任务中提交到线程池，而线程 3 是另一个任务，在整个程序中实际上有 2 个异步任务。
    • 使用 thenComposeAsync，线程 2 就是一个独立的任务，这样一来三个线程都是各自独立，每次任务结束后都会重新寻找空闲线程运行，在整个程序中实际上有 3 个异步任务。（由于线程复用，前后任务选择的可能是同一个线程）
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    public class ThenComposeAsyncTest {
        public static void main(String[] args) {
            CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
                SmallTool.printTimeAndThread("厨师炒菜");
                SmallTool.sleepMillis(200);
                return "番茄炒蛋";		// 线程1
            }).thenComposeAsync(dish -> {
                SmallTool.printTimeAndThread("服务员A 准备打饭，但是被领导叫走，打饭交接给服务员B");	// 线程2
                return CompletableFuture.supplyAsync(() -> {
                    SmallTool.printTimeAndThread("服务员B 打饭");
                    SmallTool.sleepMillis(100);
                    return dish + " + 米饭";		// 线程3
                });
            });
            SmallTool.printTimeAndThread(cf1.join()+"好了，开饭");
        }
    }

相同类型的方法

无入参，有出参 有入参，有出参 有入参，无出参 无入参，无出参
supplyAsync runAsync
thenCompose
thenCombine thenAcceptBoth runAfterBoth
thenApply thenAccept thenRun
applyToEither acceptEither runAfterEither
exceptionally/handle whenComplete

6、CompletableFuture 与 ListenableFuture 之间的转换

更新时间：2022-09-11 18:20

FutureUtil 工具类，实现 ListenableFuture 和 CompletableFuture 的相互转换：

测试



-------------Thanks for your time.-------------



× 目录
  1. 1. 0、准备
  2. 2. 1、get/join
    1. 2.1. 1.1、get
    2. 2.2. 1.2、getNow
    3. 2.3. 1.3、join
  3. 3. 2、supplyAsync/runAsync
  4. 4. 3、thenCompose
  5. 5. 4、thenCombine
    1. 5.0.1. thenAcceptBoth/runAfterBoth
  • 6. 5、thenApply/thenApplyAsync
    1. 6.0.1. thenAccept/thenRun
  • 7. 6、applyToEither
    1. 7.0.1. acceptEither/runAfterEither
  • 8. 7、exceptionally
    1. 8.0.1. handle/whenComplete
  • 9. 8、complete/completeExceptionally
  • 10. 9、allOf/anyOf
  • 11. 10、规律
  • 12. 6、CompletableFuture 与 ListenableFuture 之间的转换

    • tag:

      缺失模块。
      1、请确保node版本大于6.2
      2、在博客根目录（注意不是yilia根目录）执行以下命令：
      npm i hexo-generator-json-content --save

      3、在根目录_config.yml里添加配置： 

        jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true