一、创建线程的几种方式

1.1 继承Thread类

  1. /**
  2.  * 继承Thread类方式
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/19 10:19
  6.  **/
  7. public class extendsThreadMethod {
  8.     /**
  9.      * 继承Thread类，重写run()方法，填充线程任务
  10.      */
  11.     static class Thread01 extends Thread {
  12.         @Override
  13.         public void run() {
  14.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  15.         }
  16.     }
  19.     public static void main(String[] args) {
  20.         System.out.println("main。。。start。。。");
  22.         Thread01 thread01 = new Thread01();
  23.         thread01.start();   // 启动线程
  25.         System.out.println("main。。。end。。。");
  26.     }
  27. }

运行结果：

  1. main。。。start。。。
  2. main。。。end。。。
  3. 线程运行了。。。22

1.2 实现Runnable接口方式

  1. /**
  2.  * 实现Runnable接口方式
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/19 10:38
  6.  **/
  7. public class implementsRunnableMethod {
  8.     /**
  9.      * 实现Runnable接口，重写其run()方法，填充线程要执行的任务
  10.      */
  11.     static class Runnable01 implements Runnable {
  13.         @Override
  14.         public void run() {
  15.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  16.         }
  17.     }
  19.     public static void main(String[] args) {
  20.         System.out.println("main。。。start。。。");
  22.         /**
  23.          * 1、需要先创建Runnable接口实现类对象
  24.          * 2、将上述对象作为新建Thread对象的入参，传入任务
  25.          * 3、启动线程
  26.          */
  27.         Runnable01 runnable01 = new Runnable01();
  28.         new Thread(runnable01).start();   // 启动线程
  30.         System.out.println("main。。。end。。。");
  31.     }
  32. }

运行结果：

  1. main。。。start。。。
  2. main。。。end。。。
  3. 线程运行了。。。22

1.3 实现Callable接口方式

  1. /**
  2.  * 实现Runnable接口方式
  3.  *      可获取返回值及抛出异常
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/19 10:38
  7.  **/
  8. public class implementsCallableMethod {
  9.     /**
  10.      * 实现Callable接口，指定返回值类型，重写其call()方法，填充线程要执行的任务
  11.      */
  12.     static class Callable01 implements Callable<Integer> {
  14.         @Override
  15.         public Integer call() throws Exception {
  16.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  18.             return 666;
  19.         }
  21.     }
  23.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  24.         System.out.println("main。。。start。。。");
  26.         /**
  27.          * 1、需要先创建Callable接口实现类对象
  28.          * 2、将上述对象作为新建FutureTask对象的入参，传入任务,通过此包装后续可以获取到返回值
  29.          * 3、将上述FutureTask对象作为新建Thread类对象的入参
  30.          * 3、启动线程
  31.          * 4、可以根据需要通过FutureTask去获取返回值
  32.          */
  33.         FutureTask<Integer> futureTask = new FutureTask<>(new Callable01());
  34.         new Thread(futureTask).start();   // 启动线程
  35.         // 使用get()方法，阻塞式等待结果返回
  36.         Integer integer = futureTask.get();
  37.         System.out.println("integer = " + integer);
  39.         System.out.println("main。。。end。。。");
  40.     }
  41. }

运行结果：

  1. main。。。start。。。
  2. 线程运行了。。。22
  3. integer = 666
  4. main。。。end。。。

1.4. 使用线程池创建线程

1.4.1 原生线程池

  1. public ThreadPoolExecutor(int corePoolSize,
  2.                               int maximumPoolSize,
  3.                               long keepAliveTime,
  4.                               TimeUnit unit,
  5.                               BlockingQueue<Runnable> workQueue,
  6.                               ThreadFactory threadFactory,
  7.                               RejectedExecutionHandler handler) {
  8.     ...
  9. }

1. 参数分析

a. int corePoolSize

核心线程容量。

  • 线程池创建好之后就准备就绪的线程数量，核心线程会一直存活，即使没有任务需要执行

  • 除非allowCoreThreadTimeOut参数设置为true，这样的话即使是核心线程也会被超时销毁

    b. int maximumPoolSize

    最大线程容量。

  • 线程池中允许的最大线程数，用于控制资源

    c. long keepAliveTime

    存活时间。

  • 如果当前的线程数量大于corePoolSize数量，则会在线程空闲大于指定的存活时间时进行释放

  • 此释放操作只会释放（maximumPoolSize-corePoolSize）范围的线程

    d. TimeUnit unit

    存活时间单位。

    e. BlockingQueue<Runnable> workQueue

    任务队列、阻塞队列。

  • 用来保存等待被执行任务的阻塞队列，如果当前的任务很多，就会将目前多出来的任务放在队列中，当有线程空闲的时候，就会去队列中取出新任务继续执行。

  • 常用的有：

    • ArrayBlockingQueue
    • LinkedBlockingQueue
      • 默认队列容量为Integer的最大值，可能会导致内存耗尽，所以需要指定队列大小
    • SynchronousQueue
    • PriorityBlockingQueue
      f. ThreadFactory threadFactory
      线程工厂。

  • 线程创建的工厂

    g. RejectedExecutionHandler handler

    拒绝策略。

  • 如果阻塞队列满了，则会按照我们指定的拒绝策略拒绝执行任务

  • 常用的拒绝策略有：
    • AbortPolicy：默认实现，会直接抛出RejectedExecutionException异常；
    • CallerRunsPolicy：用调用者所在的线程来执行任务；
    • DiscardPolicy：直接抛弃，任务不执行；
    • DiscardOldestPolicy：丢弃阻塞队列中最靠前的任务，并执行当前任务。

2. 工作流程

多线程与异步 - 图1
多线程与异步 - 图2

  1. 线程池创建，准备好 core 数量的核心线程，准备接受任务
  2. 新的任务进来，用 core 准备好的空闲线程执行。
    1. core 满了，就将再进来的任务放入阻塞队列中。空闲的 core 就会自己去阻塞队 列获取任务执行
    2. 阻塞队列满了，就直接开新线程执行，最大只能开到 max 指定的数量
    3. max 都执行好了。Max-core 数量空闲的线程会在 keepAliveTime 指定的时间后自动销毁。最终保持到 core 大小
    4. 如果线程数开到了 max 的数量，还有新任务进来，就会使用 reject 指定的拒绝策 略进行处理
  3. 所有的线程创建都是由指定的 factory 创建的。

1.4.2 几种常用的线程池

以下的线程池类型，在工作中并不常用，阿里巴巴开发手册中规定，不要直接使用Excutors来创建线程池，而是要使用ThreadPoolExcutor，指定相关的七个参数来实现线程池。

1. newSingleThreadExecutor

创建方式：

  1. Executors.newSingleThreadExecutor();

内部实现：

  1. // 单线程的线程池，后台从队列中获取任务依次执行
  2. public static ExecutorService newSingleThreadExecutor() {
  3.         return new FinalizableDelegatedExecutorService
  4.             (new ThreadPoolExecutor(1, 1,
  5.                                     0L, TimeUnit.MILLISECONDS,
  6.                                     new LinkedBlockingQueue<Runnable>()));
  7.     }

一个单线程的线程池，可以用于需要保证顺序执行的场景，并且只有一个线程在执行。

2. newFixedThreadPool

创建方式：

  1. Executors.newFixedThreadPool(5);

内部实现：

  1. // 固定大小，core=max，都不可回收
  2. public static ExecutorService newFixedThreadPool(int nThreads) {
  3.         return new ThreadPoolExecutor(nThreads, nThreads,
  4.                                       0L, TimeUnit.MILLISECONDS,
  5.                                       new LinkedBlockingQueue<Runnable>());
  6.     }

一个固定大小的线程池，可以用于已知并发压力的情况下，对线程数做限制。

3. newCachedThreadPool

创建方式：

  1. Executors.newCachedThreadPool();

内部实现

  1. // core是0，所有都可回收
  2. public static ExecutorService newCachedThreadPool() {
  3.         return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
  4.                                       60L, TimeUnit.SECONDS,
  5.                                       new SynchronousQueue<Runnable>());
  6.     }
  • 一个最大容量为Integer的最大值的线程池，比较适合处理执行时间比较小的任务。
  • 创建一个可缓存线程池，
    • 如果线程池长度超过处理需要，可灵活回收空闲线程，
    • 若无可回收，则新建线程

4. newScheduledThreadPool

创建方式：

  1. Executors.newScheduledThreadPool(5);

内部实现：

  1. // 定时任务的线程池
  2. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
  3.         return new ScheduledThreadPoolExecutor(corePoolSize);
  4.     }
  5. // 使用延时队列
  6. public ScheduledThreadPoolExecutor(int corePoolSize) {
  7.         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
  8.               new DelayedWorkQueue());
  9.     }
  10. // 使用原生的线程池创建
  11. public ThreadPoolExecutor(int corePoolSize,
  12.                               int maximumPoolSize,
  13.                               long keepAliveTime,
  14.                               TimeUnit unit,
  15.                               BlockingQueue<Runnable> workQueue) {
  16.         this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
  17.              Executors.defaultThreadFactory(), defaultHandler);
  18.     }

可以延时启动，定时启动的线程池，适用于需要多个后台线程执行周期任务的场景。

5. newWorkStealingPool

创建方式：

  1. Executors.newWorkStealingPool();

一个拥有多个任务队列的线程池，可以减少连接数，创建当前可用cpu数量的线程来并行执行。

1.4.3 线程池用法

  1. // 可以通过execute()方法提交任务
  2. void execute(Runnable command);
  1. // 也可以通过submit()方法提交任务
  2. Future<?> submit(Runnable task);
  4. // 可以将Runnable接口实现类中的任务的返回值获取到
  5. <T> Future<T> submit(Runnable task, T result);
  7. // 可以获取Callable接口实现类的返回值
  8. <T> Future<T> submit(Callable<T> task);

1.4.4 开发中使用线程池优势

  • 降低资源的消耗
    • 通过重复利用已经创建好的线程降低线程的创建和销毁带来的损耗
  • 提高响应速度
    • 因为线程池中的线程数没有超过线程池的最大上限时，有的线程处于等待分配任务的状态，当任务来时无需创建新的线程就能执行
  • 提高线程的可管理性
    • 线程池会根据当前系统特点对池内的线程进行优化处理，减少创建和销毁线程带来的系统开销。无限的创建和销毁线程不仅消耗系统资源，还降低系统的稳定性，使用线程池进行统一分配

二、异步编排

使用多线程可以提高性能，但在业务复杂情况下，一个异步调用可能会依赖于另一个异步调用的执行结果。所以我们需要借助于异步编排来指定多个异步调用的执行顺序，保证业务的正确完成。

比如：查询商品详情页的逻辑比较复杂，有些数据还需要远程调用，必然需要花费更多的时间。
image.png
假如商品详情页的每个查询，需要如下标注的时间才能完成。那么，用户需要 5.5s 后才能看到商品详情页的内容。很显然是不能接受的。 如果有多个线程同时完成这 6 步操作，也许只需要 1.5s 即可完成响应。

Future是 Java 5 添加的类，用来描述一个异步计算的结果。你可以使用isDone方法检查计算是否完成，或者使用get阻塞住调用线程，直到计算完成返回结果，你也可以使用cancel方法停止任务的执行。 虽然Future以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的 CPU 资源，而且也不能及时地得到计算结果，为什么不能用观察者设计模式当计算结果完成及时通知监听者呢？
在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture，提供了非常强大的 Future 的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以 通过回调的方式处理计算结果，并且提供了转换和组合 CompletableFuture 的方法。 CompletableFuture 类实现了 Future 接口，所以你还是可以像以前一样通过get方法阻塞或 者轮询的方式获得结果，但是这种方式不推荐使用。

CompletableFuture 和 FutureTask 同属于 Future 接口的实现类，都可以获取线程的执行结果。
image.png

2.1 创建异步对象

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

2.1.1 runXxx()方式

runXxxx 都是没有返回结果的。

  1. // 使用默认的线程池
  2. public static CompletableFuture<Void> runAsync(Runnable runnable);
  4. // 使用指定的线程池
  5. public static CompletableFuture<Void> runAsync(Runnable runnable,
  6.                                                    Executor executor)

测试：

  1. /**
  2.  *  异步编排
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/20 11:25
  6.  **/
  7. public class CompletableFutureDemo {
  9.     // 创建线程池
  10.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  12.     public static void main(String[] args) {
  13.         System.out.println("main...start...");
  14.         // 使用runAsync()执行任务
  15.         CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(()->{
  16.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  17.         }, executor);
  18.         System.out.println("main...end...");
  19.     }
  20. }

运行结果：

  1. main...start...
  2. main...end...
  3. 线程运行了。。。22

2.1.2 supplyXxx()方式

supplyXxx 都是可以获取返回结果的。

  1. // 使用默认的线程池
  2. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
  4. // 使用指定的线程池
  5. public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
  6.                                                        Executor executor);

测试：

  1. /**
  2.  *  异步编排
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/20 11:25
  6.  **/
  7. public class CompletableFutureDemo {
  9.     // 创建线程池
  10.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         System.out.println("main...start...");
  15.         // 使用supplyAsync()执行任务
  16.         CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
  17.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  18.             int i = 10 / 2;
  19.             System.out.println("运行结果：" + i);
  21.             return i;   // 返回结果
  22.         }, executor);
  24.         Integer integer = integerCompletableFuture.get();
  26.         System.out.println("main...end..." + integer);
  27.     }
  28. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 运行结果：5
  4. main...end...5

2.2 计算完成时回调方法

有如下几种不同的回调方式：

  1. // 可以用来接收上步骤中得到的结果或抛出的异常，在 同一线程 中进行后续操作
  2. public CompletableFuture<T> whenComplete(
  3.         BiConsumer<? super T, ? super Throwable> action)
  5. // 可以用来接收上步骤中得到的结果或抛出的异常，使用 默认线程池 创建新线程进行后续操作
  6. public CompletableFuture<T> whenCompleteAsync(
  7.     BiConsumer<? super T, ? super Throwable> action)
  9. // 可以用来接收上步骤中得到的结果或抛出的异常，使用 指定线程池 创建新线程进行后续操作
  10. public CompletableFuture<T> whenCompleteAsync(
  11.         BiConsumer<? super T, ? super Throwable> action, Executor executor)

2.2.1 测试示例：

1. 使用whenComplete()方式，做计算完成时的回调：

此种方式们可以获取到上一步骤中的返回值以及抛出的异常，然后使用上一步骤中的线程继续执行传入新的任务。当有异常抛出的时候无法进一步做处理。

无异常的情况
  1. /**
  2.  *  异步编排
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/20 11:25
  6.  **/
  7. public class CompletableFutureDemo {
  9.     // 创建线程池
  10.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         System.out.println("main...start...");
  15.         // 使用whenComplete()进行回调执行
  16.         CompletableFuture.supplyAsync(() -> {
  17.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  18.             int i = 10 / 2;
  19.             System.out.println("运行结果：" + i);
  21.             return i;   // 返回结果
  22.         }, executor).whenComplete((result, exception) -> {
  23.             System.out.println("异步任务成功完成...结果是：" + result + "；异常是：" + exception);
  24.         });
  26.         System.out.println("main...end...");
  27.     }
  28. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 运行结果：5
  4. 异步任务成功完成...结果是：5；异常是：null
  5. main...end...

存在异常的情况
  1. /**
  2.  *  异步编排
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/20 11:25
  6.  **/
  7. public class CompletableFutureDemo {
  9.     // 创建线程池
  10.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         System.out.println("main...start...");
  15.         // 使用whenComplete()进行回调执行 -- 存在异常的情况
  16.         CompletableFuture.supplyAsync(() -> {
  17.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  18.             int i = 10 / 0;
  19.             System.out.println("运行结果：" + i);
  21.             return i;   // 返回结果
  22.         }, executor).whenComplete((result, exception) -> {
  23.             System.out.println("异步任务成功完成...结果是：" + result + "；异常是：" + exception);
  24.         });
  26.         System.out.println("main...end...");
  27.     }
  28. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 异步任务成功完成...结果是：null；异常是：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
  4. main...end...

2. 使用exceptionally()进行异常的默认处理

此种方式可以让我们在上一步执行失败，抛出异常的时候，仍可以指定默认的处理逻辑，并返回结果。

  1. import java.util.concurrent.CompletableFuture;
  2. import java.util.concurrent.ExecutionException;
  3. import java.util.concurrent.ExecutorService;
  4. import java.util.concurrent.Executors;
  6. /**
  7.  *  异步编排
  8.  *
  9.  * @author maomaochong
  10.  * @date 2022/04/20 11:25
  11.  **/
  12. public class CompletableFutureDemo {
  14.     // 创建线程池
  15.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  18.         System.out.println("main...start...");
  20.         // 使用exceptionally()进行异常情况下的默认处理
  21.         CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
  22.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  23.             int i = 10 / 0;
  24.             System.out.println("运行结果：" + i);
  26.             return i;   // 返回结果
  27.         }, executor).whenComplete((result, exception) -> {
  28.             System.out.println("异步任务成功完成...结果是：" + result + "；异常是：" + exception);
  29.         }).exceptionally(exception -> {
  30.             return 10;
  31.         });
  32.         Integer integer = integerCompletableFuture.get();
  33.         System.out.println("main...end..." + integer);
  34.     }
  35. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 异步任务成功完成...结果是：null；异常是：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
  4. main...end...10

3. 对比

whenComplete 可以处理正常和异常的计算结果，exceptionally 处理异常情况。

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

方法不以 Async 结尾，意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）。

2.3 handle方法

和 complete 一样，可对结果做最后的处理（可处理异常），可改变返回值。handle()方法可以同时处理正常和异常的情况，根据不同的情况做不同的处理逻辑。

  1. public <U> CompletableFuture<U> handle(
  2.         BiFunction<? super T, Throwable, ? extends U> fn)
  4. public <U> CompletableFuture<U> handleAsync(
  5.         BiFunction<? super T, Throwable, ? extends U> fn)
  7. public <U> CompletableFuture<U> handleAsync(
  8.         BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

2.3.1 测试示例

有异常的情况

  1. /**
  2.  *  异步编排
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/20 11:25
  6.  **/
  7. public class CompletableFutureDemo {
  9.     // 创建线程池
  10.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         System.out.println("main...start...");
  15.         // 使用handle()，方法执行完成后的处理
  16.         CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
  17.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  18.             int i = 10 / 0;
  19.             System.out.println("运行结果：" + i);
  21.             return i;   // 返回结果
  22.         }, executor).handle((result, throwable) -> {
  23.             if (result != null) {
  24.                 return result*2;
  25.             }
  26.             if (throwable != null) {
  27.                 System.out.println("throwable = " + throwable);
  28.                 return 0;
  29.             }
  31.             return 0;
  32.         });
  33.         Integer integer = integerCompletableFuture.get();
  34.         System.out.println("main...end..." + integer);
  35.     }
  36. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. throwable = java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
  4. main...end...0

无异常的情况

  1. /**
  2.  *  异步编排
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/20 11:25
  6.  **/
  7. public class CompletableFutureDemo {
  9.     // 创建线程池
  10.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         System.out.println("main...start...");
  16.         // 使用handle()，方法执行完成后的处理
  17.         CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
  18.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  19.             int i = 10 / 4;
  20.             System.out.println("运行结果：" + i);
  22.             return i;   // 返回结果
  23.         }, executor).handle((result, throwable) -> {
  24.             if (result != null) {
  25.                 return result*2;
  26.             }
  27.             if (throwable != null) {
  28.                 System.out.println("throwable = " + throwable);
  29.                 return 0;
  30.             }
  32.             return 0;
  33.         });
  34.         Integer integer = integerCompletableFuture.get();
  35.         System.out.println("main...end..." + integer);
  36.     }
  37. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 运行结果：2
  4. main...end...4

2.4 线程串行化方法

带有 Async 默认是异步执行的。同之前。 下面的方法都要前置任务成功完成。

2.4.1 thenRun() 方法

只要上面的任务执行完成，就开始执行 thenRun，只是处理完任务后，执行 thenRun 的后续操作，不能接收上一步的执行结果，自己也没有返回值。

  1. // 指定接下来执行的任务，不能接收上一步返回值，无返回结果
  3. // 无需新开线程处理
  4. public CompletableFuture<Void> thenRun(Runnable action)
  6. // 新开线程处理，使用默认的线程池    
  7. public CompletableFuture<Void> thenRunAsync(Runnable action)
  9. // 新开线程处理，使用指定的线程池    
  10. public CompletableFuture<Void> thenRunAsync(Runnable action,
  11.                                                 Executor executor)

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *
  4.  * @author maomaochong
  5.  * @date 2022/04/20 11:25
  6.  **/
  7. public class CompletableFutureDemo {
  9.     // 创建线程池
  10.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         System.out.println("main...start...");
  16.         // 使用thenRunAsync()，方法执行完成后的处理
  17.         CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
  18.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  19.             int i = 10 / 4;
  20.             System.out.println("运行结果：" + i);
  22.             return i;   // 返回结果
  23.         }, executor).thenRunAsync(() -> {
  24.             System.out.println("任务2启动了...");
  25.         }, executor);
  26.         System.out.println("main...end...");
  27.     }
  28. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 运行结果：2
  4. main...end...
  5. 任务2启动了...

2.4.2 thenAccept() 方法

消费处理结果。可以接收上一步任务的执行结果，但自己无返回值。

  1. // 指定接下来执行的任务，可以接收上一步的返回结果，无返回结果
  3. // 无需新开线程处理 
  4. public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
  6. // 新开线程处理，使用默认的线程池    
  7. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
  9. // 新开线程处理，使用指定的线程池
  10. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
  11.                                                    Executor executor)

1. 测试示例

  1. import java.util.concurrent.CompletableFuture;
  2. import java.util.concurrent.ExecutionException;
  3. import java.util.concurrent.ExecutorService;
  4. import java.util.concurrent.Executors;
  6. /**
  7.  *  异步编排
  8.  *
  9.  * @author maomaochong
  10.  * @date 2022/04/20 11:25
  11.  **/
  12. public class CompletableFutureDemo {
  14.     // 创建线程池
  15.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  18.         System.out.println("main...start...");
  20.         // 使用thenRunAsync()，方法执行完成后的处理
  21.         CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
  22.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  23.             int i = 10 / 4;
  24.             System.out.println("运行结果：" + i);
  26.             return i;   // 返回结果
  27.         }, executor).thenAcceptAsync((result) -> {
  28.             System.out.println("任务2启动了..." + result);
  29.         }, executor);
  30.         System.out.println("main...end...");
  31.     }
  32. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 运行结果：2
  4. main...end...
  5. 任务2启动了...2

2.4.3 thenApply() 方法

当一个线程依赖另一个线程时，可以获取上一个任务的执行结果，并返回当前任务的返回值。

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

  1. // 指定接下来执行的任务，可以接收上一步的返回结果，可以返回结果
  3. // 无需新开线程处理
  4. public <U> CompletableFuture<U> thenApply(
  5.         Function<? super T,? extends U> fn)
  7. // 新开线程处理，使用默认的线程池
  8. public <U> CompletableFuture<U> thenApplyAsync(
  9.         Function<? super T,? extends U> fn)
  11. // 新开线程处理，使用指定的线程池
  12. public <U> CompletableFuture<U> thenApplyAsync(
  13.         Function<? super T,? extends U> fn, Executor executor)

1. 测试示例

  1. import java.util.concurrent.CompletableFuture;
  2. import java.util.concurrent.ExecutionException;
  3. import java.util.concurrent.ExecutorService;
  4. import java.util.concurrent.Executors;
  6. /**
  7.  *  异步编排
  8.  *
  9.  * @author maomaochong
  10.  * @date 2022/04/20 11:25
  11.  **/
  12. public class CompletableFutureDemo {
  14.     // 创建线程池
  15.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  18.         System.out.println("main...start...");
  21.         // 使用thenRunAsync()，方法执行完成后的处理
  22.         CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {
  23.             System.out.println("线程运行了。。。" + Thread.currentThread().getId());
  24.             int i = 10 / 4;
  25.             System.out.println("运行结果：" + i);
  27.             return i;   // 返回结果
  28.         }, executor).thenApplyAsync((result) -> {
  29.             System.out.println("任务2启动了..." + result);
  31.             return "Hello " + result;
  32.         }, executor);
  34.         String s = stringCompletableFuture.get();
  36.         System.out.println("main...end..." + s);
  37.     }
  38. }

运行结果：

  1. main...start...
  2. 线程运行了。。。22
  3. 运行结果：2
  4. 任务2启动了...2
  5. main...end...Hello 2

2.5 两任务组合-都要完成

两个任务必须都完成，触发该任务。

2.5.1 runAfterBoth()方法

组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后， 再处理该任务。

  1. // 先执行上一个任务和other指定的任务，都完成后再执行action中的任务，此种是在同一线程进行
  2. public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
  3.                                                 Runnable action)
  5. // 先执行上一个任务和other指定的任务，都完成后再执行action中的任务，此种是使用新线程进行，使用默认的线程池
  6. public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
  7.                                                      Runnable action)
  9. // 先执行上一个任务和other指定的任务，都完成后再执行action中的任务，此种是使用新线程进行，使用指定的线程池
  10. public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
  11.                                                      Runnable action,
  12.                                                      Executor executor)

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *      两任务组合 - 都要完成
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo05 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         // 使用runAsync()执行任务
  17.         // 任务一
  18.         CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
  19.             System.out.println("任务1 线程运行了。。。" + Thread.currentThread().getId());
  20.             int i = 10 / 4;
  21.             System.out.println("任务1 运行结果：" + i);
  23.             return i;   // 返回结果
  24.         }, executor);
  26.         // 任务二
  27.         CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
  28.             System.out.println("任务2 线程：" + Thread.currentThread().getId());
  30.             System.out.println("任务2 结束：");
  32.             return "Hello";
  33.         }, executor);
  35.         // 编排，当future01和future02都完成后进行任务3，不能获取到前两个任务的执行结果
  36.         future01.runAfterBothAsync(future02, ()->{
  37.             System.out.println("任务3 开始...");
  38.         }, executor);
  40.         System.out.println("main...end...");
  41.     }
  42. }

运行结果：

  1. main...start...
  2. 任务1 线程运行了。。。22
  3. 任务1 运行结果：2
  4. 任务2 线程：23
  5. 任务2 结束：
  6. main...end...
  7. 任务3 开始...

2.5.2 thenAccept() 方法

组合两个 future，获取两个future 任务的返回结果，然后处理任务，没有返回值。

  1. public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
  3. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
  5. public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
  6.                                                    Executor executor)

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *      两任务组合 - 都要完成
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo05 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         // 使用runAsync()执行任务
  17.         // 任务一
  18.         CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
  19.             System.out.println("任务1 线程运行了。。。" + Thread.currentThread().getId());
  20.             int i = 10 / 4;
  21.             System.out.println("任务1 运行结果：" + i);
  23.             return i;   // 返回结果
  24.         }, executor);
  26.         // 任务二
  27.         CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
  28.             System.out.println("任务2 线程：" + Thread.currentThread().getId());
  30.             System.out.println("任务2 结束：");
  32.             return "Hello";
  33.         }, executor);
  36.         // 编排，当future01和future02都完成后进行任务3，可以获取到前两个任务的执行结果
  37.         future01.thenAcceptBothAsync(future02, (result01, result02)->{
  38.             System.out.println("任务3 开始...之前的结果：" + result01 + "，--》" + result02);
  39.         }, executor);
  41.         System.out.println("main...end...");
  42.     }
  43. }

运行结果：

  1. main...start...
  2. 任务1 线程运行了。。。22
  3. 任务1 运行结果：2
  4. 任务2 线程：23
  5. 任务2 结束：
  6. main...end...
  7. 任务3 开始...之前的结果：2，--》Hello

2.5.3 thenCombine() 方法

  1. public <U,V> CompletableFuture<V> thenCombine(
  2.         CompletionStage<? extends U> other,
  3.         BiFunction<? super T,? super U,? extends V> fn)
  5. public <U,V> CompletableFuture<V> thenCombineAsync(
  6.         CompletionStage<? extends U> other,
  7.         BiFunction<? super T,? super U,? extends V> fn)    
  9. public <U,V> CompletableFuture<V> thenCombineAsync(
  10.         CompletionStage<? extends U> other,
  11.         BiFunction<? super T,? super U,? extends V> fn, Executor executor)

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *      两任务组合 - 都要完成
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo05 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         // 使用runAsync()执行任务
  17.         // 任务一
  18.         CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
  19.             System.out.println("任务1 线程运行了。。。" + Thread.currentThread().getId());
  20.             int i = 10 / 4;
  21.             System.out.println("任务1 运行结果：" + i);
  23.             return i;   // 返回结果
  24.         }, executor);
  26.         // 任务二
  27.         CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
  28.             System.out.println("任务2 线程：" + Thread.currentThread().getId());
  30.             System.out.println("任务2 结束：");
  32.             return "Hello";
  33.         }, executor);
  35.         // 编排，当future01和future02都完成后进行任务3，可以获取到前两个任务的执行结果，还可以处理并返回值
  36.         CompletableFuture<String> future = future01.thenCombineAsync(future02, (result01, result02) -> {
  37.             return result01 + " : " + result02 + "-->" + "HAHA";
  38.         }, executor);
  40.         String s = future.get();
  42.         System.out.println("main...end..." + s);
  43.     }
  44. }

运行结果：

  1. main...start...
  2. 任务1 线程运行了。。。22
  3. 任务1 运行结果：2
  4. 任务2 线程：23
  5. 任务2 结束：
  6. main...end...2 : Hello-->HAHA

2.6 两任务组合 - 一个完成

当两个任务中，任意一个 future 任务完成的时候，执行任务。

2.6.1 runAfter() 方法

两个任务有一个执行完成，不需要获取future 的结果，处理任务，也没有返 回值。

  1. // 任务1 和 任务2 有一个完成，即执行action中的任务，使用同一线程
  2. public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
  3.                                                   Runnable action) 
  5. // 任务1 和 任务2 有一个完成，即执行action中的任务，使用默认线程池，创建新线程执行
  6. public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
  7.                                                    Runnable action) 
  9. // 任务1 和 任务2 有一个完成，即执行action中的任务，使用指定线程池，创建新线程执行
  10. public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
  11.                                                    Runnable action,
  12.                                                    Executor executor)

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *      两任务组合 - 一个完成
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo06 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         // 使用runAsync()执行任务
  17.         // 任务一
  18.         CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
  19.             System.out.println("任务1 线程运行了。。。" + Thread.currentThread().getId());
  20.             int i = 10 / 4;
  21.             System.out.println("任务1 运行结果：" + i);
  23.             return i;   // 返回结果
  24.         }, executor);
  26.         // 任务二
  27.         CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
  28.             System.out.println("任务2 线程：" + Thread.currentThread().getId());
  29.             try {
  30.                 Thread.sleep(3000);
  31.             } catch (InterruptedException e) {
  32.                 e.printStackTrace();
  33.             }
  34.             System.out.println("任务2 结束：");
  36.             return "Hello";
  37.         }, executor);
  39.         // 编排，当future01和future02有任一完成后进行任务3，不能获取到前两个任务的执行结果
  40.         future01.runAfterEitherAsync(future02, ()->{
  41.             System.out.println("任务3 开始...");
  42.         }, executor);
  43.         System.out.println("main...end...");
  45.     }
  46. }

运行结果：

  1. main...start...
  2. 任务1 线程运行了。。。22
  3. 任务1 运行结果：2
  4. 任务2 线程：23
  5. main...end...
  6. 任务3 开始...
  7. 任务2 结束：

2.6.2 acceptEither() 方法

两个任务有一个执行完成，获取它的返回值，处理任务，没有新的返回值。

  1. // 任务1 和 任务2 有一个完成，即执行action中的任务，可获取前两个任务的执行结果，使用同一线程
  2. public CompletableFuture<Void> acceptEither(
  3.     CompletionStage<? extends T> other, Consumer<? super T> action) 
  5. // 任务1 和 任务2 有一个完成，即执行action中的任务，可获取前两个任务的执行结果，使用默认线程池创建新线程
  6. public CompletableFuture<Void> acceptEitherAsync(
  7.     CompletionStage<? extends T> other, Consumer<? super T> action) 
  9. // 任务1 和 任务2 有一个完成，即执行action中的任务，可获取前两个任务的执行结果，使用指定线程池创建新线程
  10. public CompletableFuture<Void> acceptEitherAsync(
  11.     CompletionStage<? extends T> other, Consumer<? super T> action,
  12.     Executor executor)

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *      两任务组合 - 一个完成
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo06 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         // 使用runAsync()执行任务
  17.         // 任务一
  18.         CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
  19.             System.out.println("任务1 线程运行了。。。" + Thread.currentThread().getId());
  20.             int i = 10 / 4;
  21.             System.out.println("任务1 运行结果：" + i);
  23.             return i;   // 返回结果
  24.         }, executor);
  26.         // 任务二
  27.         CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
  28.             System.out.println("任务2 线程：" + Thread.currentThread().getId());
  29.             try {
  30.                 Thread.sleep(3000);
  31.             } catch (InterruptedException e) {
  32.                 e.printStackTrace();
  33.             }
  34.             System.out.println("任务2 结束：");
  36.             return "Hello";
  37.         }, executor);
  40.         // 编排，当future01和future02有任一完成后进行任务3，可以获取到先完成的任务的执行结果
  41.         future01.acceptEitherAsync(future02, (result)->{
  42.             System.out.println("任务3 开始...之前的结果：" + result);
  43.         }, executor);
  44.         System.out.println("main...end...");
  45.     }
  46. }

运行结果：

  1. main...start...
  2. 任务1 线程运行了。。。22
  3. 任务1 运行结果：2
  4. 任务2 线程：23
  5. main...end...
  6. 任务3 开始...之前的结果：2
  7. 任务2 结束：

2.6.3 applyToEither() 方法

两个任务有一个执行完成，获取它的返回值，处理任务并有新的返回值。

  1. // 任务1 和 任务2 有一个完成，即执行action中的任务，可获取前两个任务的执行结果，可返回值，使用同一线程
  2. public <U> CompletableFuture<U> applyToEither(
  3.     CompletionStage<? extends T> other, Function<? super T, U> fn) 
  5. // 任务1 和 任务2 有一个完成，即执行action中的任务，可获取前两个任务的执行结果，可返回值，使用默认线程池创建新线程
  6. public <U> CompletableFuture<U> applyToEitherAsync(
  7.     CompletionStage<? extends T> other, Function<? super T, U> fn) 
  9. // 任务1 和 任务2 有一个完成，即执行action中的任务，可获取前两个任务的执行结果，可返回值，使用指定线程池创建新线程
  10. public <U> CompletableFuture<U> applyToEitherAsync(
  11.     CompletionStage<? extends T> other, Function<? super T, U> fn,
  12.     Executor executor)

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *      两任务组合 - 一个完成
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo06 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         // 使用runAsync()执行任务
  17.         // 任务一
  18.         CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
  19.             System.out.println("任务1 线程运行了。。。" + Thread.currentThread().getId());
  20.             int i = 10 / 4;
  21.             System.out.println("任务1 运行结果：" + i);
  23.             return i;   // 返回结果
  24.         }, executor);
  26.         // 任务二
  27.         CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
  28.             System.out.println("任务2 线程：" + Thread.currentThread().getId());
  29.             try {
  30.                 Thread.sleep(3000);
  31.             } catch (InterruptedException e) {
  32.                 e.printStackTrace();
  33.             }
  34.             System.out.println("任务2 结束：");
  36.             return "Hello";
  37.         }, executor);
  39.         // 编排，当future01和future02有任一完成后进行任务3，可以获取到先完成任务的执行结果，还可以处理并返回值
  40.         CompletableFuture<String> future = future01.applyToEitherAsync(future02, (result) -> {
  41.             return result + "-->" + "HAHA";
  42.         }, executor);
  44.         String s = future.get();
  46.         System.out.println("main...end..." + s);
  47.     }
  48. }

运行结果：

  1. main...start...
  2. 任务1 线程运行了。。。22
  3. 任务1 运行结果：2
  4. 任务2 线程：23
  5. main...end...2-->HAHA
  6. 任务2 结束：

2.7 多任务组合

  1. public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
  3. public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

2.7.1 allOf() 方法

1. 测试示例

不使用get()方法：

  1. /**
  2.  *  异步编排
  3.  *      多任务组合
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo07 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         CompletableFuture<String> imgFuture = CompletableFuture.supplyAsync(() -> {
  17.             System.out.println("查询商品的图片信息");
  19.             return "hello.jpg";
  20.         }, executor);
  22.         CompletableFuture<String> attrFuture = CompletableFuture.supplyAsync(() -> {
  23.             System.out.println("查询商品的属性");
  25.             return "黑色 + 256G";
  26.         }, executor);
  28.         CompletableFuture<String> introFuture = CompletableFuture.supplyAsync(() -> {
  29.             try {
  30.                 Thread.sleep(3000);
  31.             } catch (InterruptedException e) {
  32.                 e.printStackTrace();
  33.             }
  34.             System.out.println("查询商品介绍");
  36.             return "华为";
  37.         }, executor);
  39.         CompletableFuture<Void> allOf = CompletableFuture.allOf(imgFuture, attrFuture, introFuture);
  40. //        allOf.get(); // 等待所有线程执行结束
  42.         System.out.println("main...end...");
  43.     }
  44. }

运行结果：

  1. main...start...
  2. 查询商品的图片信息
  3. 查询商品的属性
  4. main...end...
  5. 查询商品介绍

使用get()方法：

  1. /**
  2.  *  异步编排
  3.  *      多任务组合
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo07 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         CompletableFuture<String> imgFuture = CompletableFuture.supplyAsync(() -> {
  17.             System.out.println("查询商品的图片信息");
  19.             return "hello.jpg";
  20.         }, executor);
  22.         CompletableFuture<String> attrFuture = CompletableFuture.supplyAsync(() -> {
  23.             System.out.println("查询商品的属性");
  25.             return "黑色 + 256G";
  26.         }, executor);
  28.         CompletableFuture<String> introFuture = CompletableFuture.supplyAsync(() -> {
  29.             try {
  30.                 Thread.sleep(3000);
  31.             } catch (InterruptedException e) {
  32.                 e.printStackTrace();
  33.             }
  34.             System.out.println("查询商品介绍");
  36.             return "华为";
  37.         }, executor);
  39.         CompletableFuture<Void> allOf = CompletableFuture.allOf(imgFuture, attrFuture, introFuture);
  40.         allOf.get(); // 等待所有线程执行结束
  42.         System.out.println("main...end...");
  43.     }
  44. }

运行结果：

  1. main...start...
  2. 查询商品的图片信息
  3. 查询商品的属性
  4. 查询商品介绍
  5. main...end...

2.7.2 anyOf() 方法

1. 测试示例

  1. /**
  2.  *  异步编排
  3.  *      多任务组合
  4.  *
  5.  * @author maomaochong
  6.  * @date 2022/04/20 11:25
  7.  **/
  8. public class CompletableFutureDemo07 {
  10.     // 创建线程池
  11.     public static ExecutorService executor = Executors.newFixedThreadPool(10);
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         System.out.println("main...start...");
  16.         CompletableFuture<String> imgFuture = CompletableFuture.supplyAsync(() -> {
  17.             System.out.println("查询商品的图片信息");
  19.             return "hello.jpg";
  20.         }, executor);
  22.         CompletableFuture<String> attrFuture = CompletableFuture.supplyAsync(() -> {
  23.             System.out.println("查询商品的属性");
  25.             return "黑色 + 256G";
  26.         }, executor);
  28.         CompletableFuture<String> introFuture = CompletableFuture.supplyAsync(() -> {
  29.             try {
  30.                 Thread.sleep(3000);
  31.             } catch (InterruptedException e) {
  32.                 e.printStackTrace();
  33.             }
  34.             System.out.println("查询商品介绍");
  36.             return "华为";
  37.         }, executor);
  39. //        CompletableFuture<Void> allOf = CompletableFuture.allOf(imgFuture, attrFuture, introFuture);
  40. //        allOf.get(); // 等待所有线程执行结束
  42.         CompletableFuture<Object> anyOf = CompletableFuture.anyOf(imgFuture, attrFuture, introFuture);
  43.         anyOf.get(); // 等待所有线程执行结束
  45.         System.out.println("main...end...");
  46.     }
  47. }

运行结果：

  1. main...start...
  2. 查询商品的图片信息
  3. 查询商品的属性
  4. main...end...
  5. 查询商品介绍