AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

简述CompletableFuture异步任务编排（上）

2022-12-17 381
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 简述CompletableFuture异步任务编排

前言

在之前的项目开发中，都没怎么使用过CompletableFuture的功能，只听说过和异步编程有关。为了能够在将来有需要的时候用得上，这两天花了点时间学习了一下，并简单地总结一下如何使用CompletableFuture完成异步任务编排。

先创建一个自定义的线程池，后续所有代码都会使用到：

private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new ThreadFactory() {
    private final AtomicInteger THREAD_NUM = new AtomicInteger(1);
    @Override
    public Thread newThread(Runnable r) {
      Thread t = new Thread(r);
//      设置为守护线程，main线程结束就跟着一起结束，否则main函数结束jvm还在
      t.setDaemon(true);
      t.setName("completable-future-test-Thread-" + THREAD_NUM.incrementAndGet());
      return t;
    }
  }, new ThreadPoolExecutor.AbortPolicy());
复制代码

同步串行

image.png

同步串行代表任务1、任务2、任务3按时间先后顺序执行，并且都是同一个线程来执行。

示例代码如下：

CompletableFuture
        .supplyAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task1";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return taskName;
            }, THREAD_POOL_EXECUTOR)
        .thenApply(
            (task1Result) -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task2";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println("拿到上一个任务的返回值：" + task1Result);
              System.out.println(taskName + "执行结束");
              return taskName;
            })
        .thenAccept(
             (task2Result) -> {
             Thread currentThread = Thread.currentThread();
             String ThreadName = currentThread.getName();
             String taskName = "task3";
             System.out.println(ThreadName + "开始执行任务：" + taskName);
             System.out.println("正在执行任务" + taskName);
             System.out.println("拿到上一个任务的返回值：" + task2Result);
             System.out.println(taskName + "执行结束");
           });
复制代码

执行结果：

completable-future-test-Thread-2开始执行任务：task1
正在执行任务task1
task1执行结束
completable-future-test-Thread-2开始执行任务：task2
正在执行任务task2
拿到上一个任务的返回值：task1
task2执行结束
completable-future-test-Thread-2开始执行任务：task3
正在执行任务task3
拿到上一个任务的返回值：task2
task3执行结束
复制代码

1.入口函数supplyAsync()代表一个异步的有返回值的函数，之所以异步，是与主线程区别，从线程池中的拿一个线程来执行。

2.thenApply()和thenAccept()没有Async，意味着是和前面的任务共用一个线程，从执行结果上我们也可以看到线程名称相同。

3.thenApply()需要接收上一个任务的返回值，并且自己也要有返回值。

4.thenAccept()需要接收上一个任务的返回值，但是它不需要返回值。

异步串行

image.png


异步串行代表任务1、任务2、任务3按时间先后顺序执行，并由不同的线程来执行。

示例代码如下：

CompletableFuture
        // 有返回值
        .supplyAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task1";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return taskName;
            }, THREAD_POOL_EXECUTOR)
        // 需要上一个任务的返回值，并且自身有返回值
        .thenApplyAsync(
            (task1Result) -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task2";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println("拿到上一个任务的返回值：" + task1Result);
              System.out.println(taskName + "执行结束");
              return taskName;
            }, THREAD_POOL_EXECUTOR)
        // 不需要上一个任务的返回值，自身也没有返回值
        .thenRunAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task3";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println("thenRunAsync()不需要上一个任务的返回值");
              System.out.println(taskName + "执行结束");
            }, THREAD_POOL_EXECUTOR);
复制代码

执行结果如下：

completable-future-test-Thread-2开始执行任务：task1
正在执行任务task1
task1执行结束
completable-future-test-Thread-3开始执行任务：task2
正在执行任务task2
拿到上一个任务的返回值：task1
task2执行结束
completable-future-test-Thread-4开始执行任务：task3
正在执行任务task3
thenRunAsync()不需要上一个任务的返回值
task3执行结束
复制代码

1.入口函数依然是supplyAsync()，需要传入一个有返回值的函数作为参数；如果想要没有返回值的函数传进来的话，可以使用CompletableFuture.runAsync();

2.thenApplyAsync()和thenRunAsync()分别表示里面的任务都是异步执行的，和执行前面的任务不是同一个线程；

3.thenRunAsync()需要传入一个既不需要参数，也没有返回值的任务；

并行任务

image.png

并行代表任务1、任务2、任务3没有依赖关系，分别由不同的线程执行；

示例代码如下：

CompletableFuture<String> future1 = CompletableFuture
        .supplyAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task1";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return taskName;
            }, THREAD_POOL_EXECUTOR);
    CompletableFuture<Void> future2 = CompletableFuture
        .runAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task2";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
            }, THREAD_POOL_EXECUTOR);
    CompletableFuture<String> future3 = CompletableFuture
        .supplyAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task3";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return taskName;
            }, THREAD_POOL_EXECUTOR);
复制代码

执行结果如下：

completable-future-test-Thread-4开始执行任务：task3
completable-future-test-Thread-2开始执行任务：task1
completable-future-test-Thread-3开始执行任务：task2
正在执行任务task3
task3执行结束
正在执行任务task2
正在执行任务task1
task2执行结束
task1执行结束
复制代码

一看执行结果，明显是乱序的，并且三个任务分别由三个线程执行，符合咱们的预期；注意异步的方法后面都是带有Async关键字的；

多任务结果合并计算

  • 两个任务结果的合并

image.png


任务3的执行依赖于任务1、任务2的返回值，并且任务1和任务3由同一个线程执行，任务2单独一个线程执行；

示例代码如下：

CompletableFuture
        // 任务1
        .supplyAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task1";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return taskName;
            }, THREAD_POOL_EXECUTOR)
        .thenCombine(
            CompletableFuture
                // 任务2
                .supplyAsync(
                    () -> {
                      Thread currentThread = Thread.currentThread();
                      String ThreadName = currentThread.getName();
                      String taskName = "task2";
                      System.out.println(ThreadName + "开始执行任务：" + taskName);
                      System.out.println("正在执行任务" + taskName);
                      System.out.println(taskName + "执行结束");
                      return taskName;
                    }, THREAD_POOL_EXECUTOR),
            // 任务3
            (task1Result, task2Result) -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task3";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("task1结果：" + task1Result + "\ttask2结果：" + task2Result);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return taskName;
            });
复制代码

执行结果如下：

completable-future-test-Thread-3开始执行任务：task2
completable-future-test-Thread-2开始执行任务：task1
正在执行任务task1
正在执行任务task2
task2执行结束
task1执行结束
completable-future-test-Thread-2开始执行任务：task3
task1结果：task1 task2结果：task2
正在执行任务task3
task3执行结束
复制代码

CompletableFuture提供了thenCombine()来合并另一个CompletableFuture的执行结果，所以thenCombine()需要两个参数，第一个参数是另一个CompletableFuture，第二个参数会收集前两个任务的返回值，类似下面这样：

(result1,result2)->{
  // 执行业务逻辑
  return result3;
}


image.png


如果小伙伴们想要实现任务3也是单独的线程执行的话，可以使用thenCombineAsync()这个方法。代码如下：

CompletableFuture
        // 任务1
        .supplyAsync(
            () -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task1";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return taskName;
            }, THREAD_POOL_EXECUTOR)
        .thenCombineAsync(
            CompletableFuture
                // 任务2
                .supplyAsync(
                    () -> {
                      Thread currentThread = Thread.currentThread();
                      String ThreadName = currentThread.getName();
                      String taskName = "task2";
                      System.out.println(ThreadName + "开始执行任务：" + taskName);
                      System.out.println("正在执行任务" + taskName);
                      System.out.println(taskName + "执行结束");
                      return 2;
                    }, THREAD_POOL_EXECUTOR),
            // 任务3
            (task1Result, task2Result) -> {
              Thread currentThread = Thread.currentThread();
              String ThreadName = currentThread.getName();
              String taskName = "task3";
              System.out.println(ThreadName + "开始执行任务：" + taskName);
              System.out.println("task1结果：" + task1Result + "\ttask2结果：" + task2Result);
              System.out.println("正在执行任务" + taskName);
              System.out.println(taskName + "执行结束");
              return 2L;
            }, THREAD_POOL_EXECUTOR);
复制代码

如果任务3中不需要返回结果，可以使用thenAcceptBoth()或thenAcceptBothAsync()，使用方式与thenCombineAsync()类似；


文章标签：
Java
dc73hre37f4my
目录
相关文章
菜菜的后端私房菜
|
2月前
|
Java API
异步任务编排神器CompletableFuture
【10月更文挑战第10天】CompletableFuture是JDK8并发包中引入的强大工具，用于处理复杂的异步任务编排。它提供了丰富的API，支持任务的串行、并行、组合及异常处理，适用于需要高效管理和协调多个异步操作的场景。例如，网页加载时需从多个服务异步获取数据，CompletableFuture可以有效提升性能和响应速度。使用时应注意异常处理和合理选择线程池，以确保程序稳定性和效率。
菜菜的后端私房菜
95 1
异步任务编排神器CompletableFuture
运营研究坊
|
3月前
|
数据采集 JavaScript Java
CompletableFuture异步编排，你还不会？
本文介绍了同步与异步编程的概念，探讨了在复杂业务场景中使用异步编排的重要性。通过对比 `Future` 与 `CompletableFuture`，详细讲解了 `CompletableFuture` 的多种方法，如 `runAsync`、`supplyAsync`、`whenComplete`、`exceptionally` 等，并展示了如何通过 `CompletableFuture` 实现异步任务的组合与异常处理。最后，通过实战案例演示了如何利用线程池与 `CompletableFuture` 优化商品详情页的查询效率，显著减少响应时间。
运营研究坊
90 3
CompletableFuture异步编排，你还不会？
热爱技术的小郑
|
4月前
|
Java 数据库
异步&线程池 CompletableFuture 异步编排 实战应用 【终结篇】
这篇文章通过一个电商商品详情页的实战案例，展示了如何使用`CompletableFuture`进行异步编排，以解决在不同数据库表中查询商品信息的问题，并提供了详细的代码实现和遇到问题（如图片未显示）的解决方案。
热爱技术的小郑
103 1
异步&线程池 CompletableFuture 异步编排 实战应用 【终结篇】
热爱技术的小郑
|
4月前
|
Java
异步&线程池 CompletableFuture 异步编排 【下篇】
这篇文章深入探讨了Java中的`CompletableFuture`类，解释了如何创建异步操作、使用计算完成时的回调方法、异常处理、串行化方法、任务组合以及多任务组合的使用方式，并通过代码示例展示了各种场景下的应用。
热爱技术的小郑
109 0
异步&线程池 CompletableFuture 异步编排 【下篇】
程序员古德
|
7月前
|
前端开发 Java API
Java并发基础：CompletableFuture全面解析
CompletableFuture类使得并发任务的处理变得简单而高效，通过简洁的API，开发者能轻松创建、组合和链式调用异步操作，无需关心底层线程管理，这不仅提升了程序的响应速度，还优化了资源利用率，让复杂的并发逻辑变得易于掌控。
程序员古德
198 1
Java并发基础：CompletableFuture全面解析
小趴菜111.
|
7月前
|
Java
JAVA线程&线程池&异步编排
JAVA线程&线程池&异步编排
小趴菜111.
57 0
2G冲浪词条
|
安全 Java
任务编排：CompletableFuture从入门到精通
最近遇到了一个业务场景，涉及到多数据源之间的请求的流程编排，正好看到了一篇某团介绍CompletableFuture原理和使用的技术文章，主要还是涉及使用层面。网上很多文章涉及原理的部分讲的不是特别详细且比较抽象。因为涉及到多线程的工具必须要理解原理，不然一旦遇到问题排查起来就只能凭玄学，正好借此梳理一下CompletableFuture的工作原理
2G冲浪词条
357 0
宁在春
|
7月前
|
Java
CompletableFuture 异步编排、案例及应用小案例1
CompletableFuture 异步编排、案例及应用小案例
宁在春
174 0
宁在春
|
7月前
|
Java
CompletableFuture 异步编排、案例及应用小案例2
CompletableFuture 异步编排、案例及应用小案例
宁在春
85 0
Shepherd
|
存储 SpringCloudAlibaba Java
Java新特性：异步编排CompletableFuture
CompletableFuture由Java 8提供，是实现异步化的工具类，上手难度较低，且功能强大，支持通过函数式编程的方式对各类操作进行组合编排。 CompletableFuture实现了CompletionStage接口和Future接口，前者是对后者的一个扩展，增加了异步[回调](https://so.csdn.net/so/search?q=回调&spm=1001.2101.3001.7020)、流式处理、多个Future组合处理的能力，使Java在处理多任务的协同工作时更加顺畅便利。
Shepherd
384 1
Java新特性：异步编排CompletableFuture