CompletableFuture使用场景和实现原理，java异步回调

ゝ一纸荒年。 2023-10-10 14:34 7阅读 0赞

1.概述  
CompletableFuture是jdk1.8引入的实现类。扩展了Future和CompletionStage，是一个可以在任务完成阶段触发一些操作Future。简单的来讲就是可以实现异步回调。

2.为什么引入CompletableFuture  
对于jdk1.5的Future，虽然提供了异步处理任务的能力，但是获取结果的方式很不优雅，还是需要通过阻塞（或者轮训）的方式。如何避免阻塞呢？其实就是注册回调。

CompletableFuture是java8新增的并发工具类，继承了FutureTask的同步任务的特点，同时新增了[异步调用][Link 1]的特点（其中异步的方法名称都带有Async），换而言之同步获取方法的返回值的方式可以用CompletableFuture完成，与此同时，想要异步获取方法的返回值也可以使用CompletableFuture来完成。

--------------------

异步带[Async][]，并且底层执行的线程由ForkJoinPool支持。于此同时还多了异常处理（执行任务的时候可能会发生异常，以前使用FutureTask的同步的方式是需要在执行的方法内处理异常的，而使用CompletableFuture后则可以对异常捕捉，并且修改返回值，受java8的函数式编程的特点）

重点知识：  
CompletableFuture在执行异步任务的时候默认采用的是ForkJoinPoin线程池（前提是cpu是多核的，也就是Runtime.getRuntime().availableProcessors() - 1，例如cpu是12核心的，那么就是返回11）  
**CompletableFuture-异步执行，异步回调**  
使用jdk8新特性：CompletableFuture

*  a.加入适量的水
 *  b.打开煤气
 *  c.playGame // 打一局小游戏
 *  d.主动提醒，水已烧开 ---- //主动通知-回调
 *  e.关闭煤气

水烧好后就关煤气，烧水就回调，游戏打完后，游戏再回调。

### 三、CompletableFuture的整体结构和流程 ###

CompletableFuture不需要new得到，起点是由CompletableFuture的public static方法，当我们调用这些方法时底层就会自动创建一个CompletableFuture实例，我们只需要把任务以及自定义的线程传入即可。所以一般的流程是以下。

### 四、实际的多任务并发组合场景 ###

如果想要写好高并发程序，利用CompletableFuture可以快速解决一些开发中常见的并发场景

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池，一般放在静态成员变量中
    CompletableFuture<Integer> exceptionally = CompletableFuture.supplyAsync(() -> {
        //返回一个数据
        return 1;
    },executorService).whenComplete((res, throwable) -> {
        //完成异步时获取返回值
        System.out.println("返回值是："+res+",异常是："+throwable);
    }).exceptionally((throwable) -> {
        //接收异常，修改返回值
        System.out.println("获得到的异常是:"+throwable);
        return 10;//返回一个值替代原来的返回值
    });

**handle方法感知返回值，处理异常，并修改返回值**  
上面的方式通过whenComplete方法感知结果的产生，并且能接收异常，但是不能修改返回值，如果要修改返回值，就得和上面的案例一样结合exceptionally方法才能达到效果。

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池，一般放在静态成员变量中
    CompletableFuture<Integer> exceptionally = CompletableFuture.supplyAsync(() -> {
        //返回一个数据
        return 1;
    },executorService).handle((res,throwable)->{
        if (res==1){
            System.out.println("返回的结果是"+res);
            return 1;
        }
        if (throwable!=null){
            System.out.println("异常是"+throwable);
            return 0;
        }
        return 100;
    });

#### 多任务组合场景 ####

**情景一、异步组合两任务都完成，执行任务三**  
runAfterBoth方法无法感知任务一和任务二是否完成

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池，一般放在静态成员变量中
     
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务一开始");
        return 1;
    },executorService);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务二开始");
        return 2;
    },executorService);
    future1.runAfterBothAsync(future2,()->{
        System.out.println("联合任务1和2");
    },executorService);

**使用thenAcceptBothAsync会感知任务一和任务二都完成。**

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池，一般放在静态成员变量中
     
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务一开始");
        return 1;
    },executorService);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务二开始");
        return 2;
    },executorService);
    future1.thenAcceptBothAsync(future2,(result1,result2)->{
        System.out.println("感知任务一和任务二完成，执行任务三");
    },executorService);

**合并多个任务使用thenCombineAsync在上面的结果，基础上，如果想要有返回值则使用它**

####    情景二、异步两任务只要有一个完成，就执行任务三 ####

**两个任务只要有一个完成就执行任务三的方法都是 XXXEither都带有Either**runAfterEitherAsync：不感知结果，并且无返回值

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池
     
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务一开始");
        try {
            TimeUnit.MILLISECONDS.sleep(200);//让任务二晚点执行完，等任务二完成，任务三就会开始
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 1;
    },executorService);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务二开始");
        return 2;
    });
    future1.runAfterEitherAsync(future2,()->{
        System.out.println("任务三开始，任务一任务二只要有一个完成就会开始执行任务三");
    },executorService);

**acceptEitherAsync：感知结果，接收返回值并消费掉，不产生返回值**

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务一开始");
        try {
            TimeUnit.MILLISECONDS.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 1;
    },executorService);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务二开始");
        return 2;
    });
    future1.acceptEitherAsync(future2,(res)->{
        System.out.println("感知结果执行任务三");//需要注意使用lambda表达式需要future1和future2返回值类型相同
    },executorService);

**applyToEitherAsync：感知返回值，转换返回值得到一个新的结果**

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务一开始");
        try {
            TimeUnit.MILLISECONDS.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 1;
    },executorService);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务二开始");
        return 2;
    });
    future1.applyToEitherAsync(future2,(res)->{
        System.out.println("感知结果执行任务三");//需要注意使用lambda表达式需要future1和future2返回值类型相同
        return "新的返回值"+res*2;
    },executorService);

#### 情景三、多任务组合完成 ####

**allOf方法介绍，多任务都完成**  
如果使用下面的代码等待多任务完成。

问题一：是执行future1.get();时主线程是堵塞的，因此future2.get();future3.get();都不会执行需要等待future1先得到返回值，也就是说会加上多余的堵塞时间。虽然任务是异步的(的确任务已经完成了，但是线程没有交还线程池)  
而我们希望的是对于提前完成任务的线程将他交还给线程池，让它可以再次被其它任务执行

例如：future1 耗时4s，future2耗时3秒，future3耗时5s。如果是直接下面代码，则会导致future2在3秒的时候就已经可以完成任务了，但是由于没有进行get，线程一直在等待返回数据，那么future2的线程相当于被占用着。

问题二：就是产生冗余代码

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池
     
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务一开始");
        return 1;
    },executorService);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务二开始");
        return 2;
    });
    CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务三开始");
        return 3;
    }, executorService);
    CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3);
    try {
        allOf.get();//等待所有任务都完成
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ExecutionException e) {
        e.printStackTrace();
    }

如果要获得每一个任务的返回结果还是需要使用future1.get();,future2.get(),future3.get()得到返回结果。

**情景四、多任务组合只要有一个完成**

ExecutorService executorService = Executors.newFixedThreadPool(10);//线程池
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务一开始");
        return 1;
    }, executorService);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务二开始");
        return 2;
    });
    CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务三开始");
        return 3;
    }, executorService);
    CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2, future3);
    try {
        anyOf.get();//获得完成的那个任务结果,其它任务的结果就获取不到，想要获取得调用各自得get
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ExecutionException e) {
        e.printStackTrace();
    }

多线程任务，堵塞执行，获取所有结果

List<CompletableFuture<String>> list = new ArrayList<CompletableFuture<String>>();
            
            list.add(CompletableFuture.supplyAsync(() -> getData(), executorServe));
            list.add(CompletableFuture.supplyAsync(() -> getData(), executorServe));
            list.add(CompletableFuture.supplyAsync(() -> getData(), executorServe));
                    
            CompletableFuture.allOf(list.toArray(new CompletableFuture[list.size()])).join();
    
            log.info(String.format("********全部执行完毕，总耗时:%s ********", (ChronoUnit.SECONDS.between(start, Instant.now()))));
            //打印结果
            list.parallelStream().forEach((cf)->{
            	log.info(cf.join());
            });
    
    
    //另一种
    
    List<CompletableFuture<ItemCodeRuleVO>> futures = new ArrayList<>();
     if(futures.size() != 0){
         CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
         //获取返回结果集
          List<ItemCodeRuleVO> itemCodeRuleVOResult =
             allFutures.thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList())).join();
                
         for(ItemCodeRuleVO itemOne : itemCodeRuleVOResult) {}
    }

[Link 1]: https://so.csdn.net/so/search?q=%E5%BC%82%E6%AD%A5%E8%B0%83%E7%94%A8&spm=1001.2101.3001.7020
[Async]: https://so.csdn.net/so/search?q=Async&spm=1001.2101.3001.7020