• Java并发-CompletableFuture的详解

    目录

    1 前言

     2 常用方法

    3 测试

    3.1 runAsync：无返回值 和 SupplyAsync：有返回值

     3.2 串行执行

    3.3 任务3等待等任务1和任务2都执行完毕后执行

    3. 4 任务3等待等任务1或者任务2执行完毕后执行

    3.5 handleAsync

    3.6 多任务执行

    1 前言

    CompletableFuture 是对 Future 的扩展和增强。CompletableFuture 实现了Future和CompletionStage接口，其中最重要的是CompletionStage，它定义了CompletableFuture 实现任务编排的一些基本方法。只能说非常好用。

    其继承结构如下图所示，在IDEA中使用ctrl + alt + U 实现

     CompletionStage接口定义了任务编排的方法，执行某一阶段，可以向下执行后续阶段。异步执行的，默认线程池是ForkJoinPool.commonPool()，但为了业务之间互不影响，且便于定位问题，强烈推荐使用自定义线程池。

     2 常用方法

    主要有同步和异步两大类，这里只讲异步方法。同步方法可以类比。

    创建类型的方法

    • runAsync：无返回值
    • SupplyAsync：有返回值

    中间处理类型的方法

    • 任务与任务之间串行执行
      • thenRunAsync：不能获取上一次的返回值，本次没有返回值。
      • thenAccpetAsync：能获取上一次的返回值，但是本次没有返回值。
      • thenApplyAsync：能获取上一次的返回值，本次有返回值。
    • 任务3等待等任务1和任务2都执行完毕后执行
      • runAfterBothAsync：不能获取任务1和任务2的返回值，本次没有返回值。
      • thenAcceptBothAsync：能获取任务1和任务2的返回值，但是本次没有返回值。
      • thenCombineAsync：能获取任务1和任务2的返回值，本次也有返回值。
    • 任务3等待任务1或者任务2其中一个执行完毕、
      • runAfterEitherAsync：不能获取任务1或任务2的返回值，本次没有返回值
      • acceptEitherAsync：能获取任务1或任务2的返回值，本次没有返回值
      • applyToEither：能获取任务1或任务2的返回值，本次有返回值
    • 多任务
      • allof：当所有给定的 CompletableFuture 完成时，返回一个新的 CompletableFuture
      • anyof：当任何一个给定的CompletableFuture完成时，返回一个新的CompletableFuture

    结果处理类型的方法

    • whenCompleteAsync：获取结果和异常，但是没有返回值
    • exceptionally：在whenCompleteAsync后时候，获取异常并且有返回值
    • handleAsync：相当于whenCompleteAsync和exceptionally的结合体。

    3 测试

    3.1 runAsync：无返回值 和 SupplyAsync：有返回值

    • runAsync 相当于开启一个异步线程执行任务，主线程不等待结果返回。
    • supplyAsync，开启一个异步线程执行任务，主线程要等待结果返回。
    1. package cn.itcast.n6.c3;
    2.  
    3. import java.util.concurrent.*;
    4.  
    5. /**
    6.  * @author : msf
    7.  * @date : 2022/12/4
    8.  * completable 讲解。
    9.  */
    10. public class CompletableFutureD1 {
    11.     static ThreadPoolExecutor executor = new ThreadPoolExecutor(
    12.             5,50,10, TimeUnit.SECONDS,
    13.             new LinkedBlockingQueue<>(100),Executors.defaultThreadFactory()
    14.             ,new ThreadPoolExecutor.AbortPolicy()
    15.     );
    16.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    17.         System.out.println("主线程 start ...");
    18.        CompletableFuture future = CompletableFuture.runAsync(()->{
    19.            System.out.println("任务1子线程执行..." + Thread.currentThread().getName());
    20.        },executor);
    21.  
    22.         CompletableFuture futureInt = CompletableFuture.supplyAsync(() -> {
    23.             System.out.println("任务2子线程执行..." + Thread.currentThread().getName());
    24.             return 10;
    25.         }, executor);
    26.  
    27.         System.out.println(futureInt.get());
    28.         System.out.println("主线程 end ...");
    29.     }
    30. }

    上述代码执行结果：

     3.2 串行执行

    • thenRunAsync：不能获取上一次的返回值，本次没有返回值。
    • thenAccpetAsync：能获取上一次的返回值，但是本次没有返回值。
    • thenApplyAsync：能获取上一次的返回值，本次有返回值

    1. package cn.itcast.n6.c3;
    2.  
    3. import java.util.concurrent.*;
    4.  
    5. /**
    6.  * @author : msf
    7.  * @date : 2022/12/4
    8.  * completable 讲解--串行编排
    9.  */
    10. public class CompletableFutureD4 {
    11.     static ThreadPoolExecutor executor = new ThreadPoolExecutor(
    12.             5, 50, 10, TimeUnit.SECONDS,
    13.             new LinkedBlockingQueue<>(100), Executors.defaultThreadFactory()
    14.             , new ThreadPoolExecutor.AbortPolicy()
    15.     );
    16.  
    17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    18.         System.out.println("主线程 start ...");
    19.         CompletableFuture future = CompletableFuture.runAsync(() -> {
    20.             System.out.println("任务1子线程执行..." + Thread.currentThread().getName());
    21.             try {
    22.                 Thread.sleep(1000);
    23.             } catch (InterruptedException e) {
    24.                 e.printStackTrace();
    25.             }
    26.         }, executor).thenRunAsync(()->{
    27.             // 等任务1 执行完毕后，任务2执行
    28.             System.out.println("任务2执行了" +Thread.currentThread().getName());
    29.         },executor);
    30.  
    31.  
    32.         CompletableFuture.supplyAsync(() -> {
    33.             System.out.println("任务3子线程执行..." + Thread.currentThread().getName());
    34.             try {
    35.                 Thread.sleep(1000);
    36.             } catch (InterruptedException e) {
    37.                 e.printStackTrace();
    38.             }
    39.             return 100;
    40.         }, executor).thenAcceptAsync((t)->{
    41.             System.out.println("任务4子线程执行..." + Thread.currentThread().getName());
    42.             System.out.println("t = " + t);
    43.         },executor);
    44.  
    45.         CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
    46.             System.out.println("任务5子线程执行..." + Thread.currentThread().getName());
    47.             try {
    48.                 Thread.sleep(1000);
    49.             } catch (InterruptedException e) {
    50.                 e.printStackTrace();
    51.             }
    52.             return 100;
    53.         }, executor).thenApplyAsync((t) -> {
    54.             System.out.println("任务6子线程执行..." + Thread.currentThread().getName());
    55.             System.out.println("t = " + t);
    56.             return 120;
    57.         }, executor);
    58.         System.out.println("主线程 end ..." + future1.get());
    59.  
    60.     }
    61. }

    上述代码执行结果：

    3.3 任务3等待等任务1和任务2都执行完毕后执行

    开启一个任务1，然后开启任务2，我们需要的是，当任务1和2都执行结束后任务3再启动。

    场景：例如两个人从不同地方去吃饭，都有两个人都到了然后开始点餐。

    1. package cn.itcast.n6.c3;
    2.  
    3. import java.util.concurrent.*;
    4.  
    5. /**
    6.  * @author : msf
    7.  * @date : 2022/12/4
    8.  * completable 讲解-- 两个一起完成，或者两者其中之一完成
    9.  */
    10. public class CompletableFutureD5 {
    11.     static ThreadPoolExecutor executor = new ThreadPoolExecutor(
    12.             5, 50, 10, TimeUnit.SECONDS,
    13.             new LinkedBlockingQueue<>(100), Executors.defaultThreadFactory()
    14.             , new ThreadPoolExecutor.AbortPolicy()
    15.     );
    16.  
    17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    18.         System.out.println("主线程 start ...");
    19.         CompletableFuture future = CompletableFuture.runAsync(() -> {
    20.             System.out.println("任务1子线程执行..." + Thread.currentThread().getName());
    21.             try {
    22.                 Thread.sleep(500);
    23.             } catch (InterruptedException e) {
    24.                 e.printStackTrace();
    25.             }
    26.         }, executor);
    27.  
    28.  
    29.         CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
    30.             System.out.println("任务2子线程执行..." + Thread.currentThread().getName());
    31.             try {
    32.                 Thread.sleep(600);
    33.             } catch (InterruptedException e) {
    34.                 e.printStackTrace();
    35.             }
    36.             return 100;
    37.         }, executor);
    38.  
    39.         future.runAfterBothAsync(future1, () -> {
    40.             System.out.println("任务3..." + Thread.currentThread().getName());
    41.             try {
    42.                 Thread.sleep(1000);
    43.             } catch (InterruptedException e) {
    44.                 e.printStackTrace();
    45.             }
    46.             System.out.println("任务3.... end " + Thread.currentThread().getName());
    47.         }, executor);
    48.  
    49.         System.out.println("主线程 end ..." + future1.get());
    50.  
    51.     }
    52. }

    上述代码执行结果： 

    3. 4 任务3等待等任务1或者任务2执行完毕后执行

    场景：当两个人都出去吃饭，其中一个人到了可以先点餐。

    1. package cn.itcast.n6.c3;
    2.  
    3. import java.util.concurrent.*;
    4.  
    5. /**
    6.  * @author : msf
    7.  * @date : 2022/12/4
    8.  * completable 讲解-- 两个一起完成，或者两者其中之一完成
    9.  */
    10. public class CompletableFutureD6 {
    11.     static ThreadPoolExecutor executor = new ThreadPoolExecutor(
    12.             5, 50, 10, TimeUnit.SECONDS,
    13.             new LinkedBlockingQueue<>(100), Executors.defaultThreadFactory()
    14.             , new ThreadPoolExecutor.AbortPolicy()
    15.     );
    16.  
    17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    18.         System.out.println("主线程 start ...");
    19.         CompletableFuture future = CompletableFuture.runAsync(() -> {
    20.             System.out.println("任务1子线程执行..." + Thread.currentThread().getName());
    21.             try {
    22.                 Thread.sleep(500);
    23.             } catch (InterruptedException e) {
    24.                 e.printStackTrace();
    25.             }
    26.             System.out.println("任务1子线程执行...end " + Thread.currentThread().getName());
    27.         }, executor);
    28.  
    29.  
    30.         CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
    31.             System.out.println("任务2子线程执行..." + Thread.currentThread().getName());
    32.             try {
    33.                 Thread.sleep(600);
    34.             } catch (InterruptedException e) {
    35.                 e.printStackTrace();
    36.             }
    37.             System.out.println("任务2子线程执行...end " + Thread.currentThread().getName());
    38.             return 100;
    39.         }, executor);
    40.  
    41.         future.runAfterEitherAsync(future1, () -> {
    42.             System.out.println("任务3..." + Thread.currentThread().getName());
    43.         }, executor);
    44.  
    45.         System.out.println("主线程 end ..." + future1.get());
    46.  
    47.     }
    48. }

    上述代码执行如下： 

    3.5 handleAsync

    handleAsync：相当于whenCompleteAsync和exceptionally的结合体，可以收集结果和异常信息然后进行下一步处理。

    1. package cn.itcast.n6.c3;
    2.  
    3. import java.util.concurrent.*;
    4.  
    5. /**
    6.  * @author : msf
    7.  * @date : 2022/12/4
    8.  * completable 讲解。
    9.  */
    10. public class CompletableFutureD3 {
    11.     static ThreadPoolExecutor executor = new ThreadPoolExecutor(
    12.             5, 50, 10, TimeUnit.SECONDS,
    13.             new LinkedBlockingQueue<>(100), Executors.defaultThreadFactory()
    14.             , new ThreadPoolExecutor.AbortPolicy()
    15.     );
    16.  
    17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    18.         System.out.println("主线程 start ...");
    19.         CompletableFuture future = CompletableFuture.runAsync(() -> {
    20.             System.out.println("任务1子线程执行..." + Thread.currentThread().getName());
    21.         }, executor).handleAsync((res, exec) -> {
    22.             System.out.println("res1 = " + res);
    23.             System.out.println("exec1 = " + exec);
    24.             return null;
    25.         });
    26.         CompletableFuture futureInt = CompletableFuture.supplyAsync(() -> {
    27.             System.out.println("任务2子线程执行..." + Thread.currentThread().getName());
    28.             int i = 10 / 0;
    29.             return 10;
    30.         }, executor).handleAsync((res, exec) -> {
    31.             System.out.println("res2 = " + res);
    32.             System.out.println("exec2 = " + exec);
    33.             return 55;
    34.         });
    35.  
    36.         System.out.println(futureInt.get());
    37.         System.out.println("主线程 end ...");
    38.     }
    39. }

    上述代码执行结果：

    3.6 多任务执行

    模拟：三人吃饭，只要一个人到了就可以点餐，但是只有所有人到了，并且厨子把菜做好了才能上菜；

    1. package cn.itcast.n6.c3;
    2.  
    3. import java.util.concurrent.CompletableFuture;
    4. import java.util.concurrent.ExecutionException;
    5.  
    6. /**
    7.  * @author : msf
    8.  * @date : 2022/12/5
    9.  * 模拟三人点餐，只要有一个人到了就可以点餐。然后交给厨子进行做饭
    10.  */
    11. public class Test {
    12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    13.         long begin = System.currentTimeMillis();
    14.         CompletableFuture client1 = CompletableFuture.runAsync(() -> {
    15.             System.out.println("客户1开始出发了.....1小时后到");
    16.             try {
    17.                 Thread.sleep(1000);
    18.             } catch (InterruptedException e) {
    19.                 e.printStackTrace();
    20.             }
    21.             System.out.println("客户1到了");
    22.         });
    23.         CompletableFuture client2 = CompletableFuture.runAsync(()->{
    24.             System.out.println("客户2开始出发了.....半小时后到");
    25.             try {
    26.                 Thread.sleep(500);
    27.             } catch (InterruptedException e) {
    28.                 e.printStackTrace();
    29.             }
    30.             System.out.println("客户2到了");
    31.         });
    32.         CompletableFuture client3 = CompletableFuture.runAsync(()->{
    33.             System.out.println("客户3开始出发了.....1.5小时后到");
    34.             try {
    35.                 Thread.sleep(1500);
    36.             } catch (InterruptedException e) {
    37.                 e.printStackTrace();
    38.             }
    39.             System.out.println("客户3到了");
    40.         });
    41.         CompletableFuture result = CompletableFuture.anyOf(client1, client2, client3);
    42.         System.out.println("已经有客户到了开始点餐" +result.get());
    43.  
    44.         CompletableFuture cook = CompletableFuture.runAsync(() -> {
    45.             System.out.println("厨师开始做饭");
    46.             try {
    47.                 Thread.sleep(2000);
    48.             } catch (InterruptedException e) {
    49.                 e.printStackTrace();
    50.             }
    51.             System.out.println("厨师做完饭了");
    52.         });
    53.         CompletableFuture reuslt1 = CompletableFuture.allOf(client1, client2, client3, cook);
    54.         reuslt1.join();
    55.         System.out.println("人全部来齐了，厨师也做好饭了,上菜！");
    56.         long end = System.currentTimeMillis();
    57.         System.out.println("总共时间盲猜2.5秒, 真实时间" + (end - begin) );
    58.     }
    59. }
      60.  

       

       

      
                
                    
                

                
    60. 

                    相关阅读:
      
                    
经济师十大专业通过人数分析！选专业有谱了！                            
      
2023高教社杯 国赛数学建模D题思路 - 圈养湖羊的空间利用率                            
      
c# wpf template ItemTemplate 简单试验                            
      
VLAN和VLAN间路由                            
      
【LeetCode刷题】1两数之和                            
      
C专家编程 第11章 你懂得C，所以C++不再话下 11.7 如何调用成员函数                            
      
C#每天复习一个重要小知识day2：有参与无参构造函数                            
      
【JavaScript】掌握BOM浏览器对象模型                            
      
Python Flask框架 入门详解与进阶                            
      
基于TCP传输的网络编程异常处理                            
      
                          
                
      61. 
                
    61. 
                    原文地址：https://blog.csdn.net/abc123mma/article/details/128183712
                
      62.