Java8里面CompletableFuture异步编程详解

﹏ヽ暗。殇╰゛Y 2023-02-16 04:22 7阅读 0赞

Java8主要的语言增强的能力有：

（1）lambda表达式

（2）stream式操作

（3）CompletableFuture

其中第三个特性，就是今天我们想要聊的话题，正是因为CompletableFuture的出现，才使得使用Java进行异步编程提供了可能。

### 什么是CompletableFuture？ ###

CompletableFuture在Java里面被用于异步编程，异步通常意味着非阻塞，可以使得我们的任务单独运行在与主线程分离的其他线程中，并且通过 回调可以在主线程中得到异步任务的执行状态，是否完成，和是否异常等信息。CompletableFuture实现了Future, CompletionStage接口，实现了Future接口就可以兼容现在有线程池框架，而CompletionStage接口才是异步编程的接口抽象，里面定义多种异步方法，通过这两者集合，从而打造出了强大的CompletableFuture类。

### Future vs CompletableFuture ###

Futrue在Java里面，通常用来表示一个异步任务的引用，比如我们将任务提交到线程池里面，然后我们会得到一个Futrue，在Future里面有isDone方法来 判断任务是否处理结束，还有get方法可以一直阻塞直到任务结束然后获取结果，但整体来说这种方式，还是同步的，因为需要客户端不断阻塞等待或者不断轮询才能知道任务是否完成。

Future的主要缺点如下：

（1）不支持手动完成

这个意思指的是，我提交了一个任务，但是执行太慢了，我通过其他路径已经获取到了任务结果，现在没法把这个任务结果，通知到正在执行的线程，所以必须主动取消或者一直等待它执行完成。

（2）不支持进一步的非阻塞调用

这个指的是我们通过Future的get方法会一直阻塞到任务完成，但是我还想在获取任务之后，执行额外的任务，因为Future不支持回调函数，所以无法实现这个功能。

（3）不支持链式调用

这个指的是对于Future的执行结果，我们想继续传到下一个Future处理使用，从而形成一个链式的pipline调用，这在Future中是没法实现的。

（4）不支持多个Future合并

比如我们有10个Future并行执行，我们想在所有的Future运行完毕之后，执行某些函数，是没法通过Future实现的。

（5）不支持异常处理

Future的API没有任何的异常处理的api，所以在异步运行时，如果出了问题是不好定位的。

### 简单的使用CompletableFuture ###

1，先看一个最简单的例子

在主线程里面创建一个CompletableFuture，然后主线程调用get方法会阻塞，最后我们在一个子线程中 使其终止。

CompletableFuture<String> completableFuture=new CompletableFuture<String>();
 Runnable runnable=new Runnable() {
 @Override
 public void run() {
 try {
 TimeUnit.SECONDS.sleep(3);
 System.out.println(getThreadName()+"执行.....");
 completableFuture.complete("success");//在子线程中完成主线程completableFuture的完成
 
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 };
 Thread t1=new Thread(runnable);
 t1.start();//启动子线程
 
 String result=completableFuture.get();//主线程阻塞，等待完成
 System.out.println(getThreadName()+"1x: "+result);

输出结果：

Thread-0线程=> 执行.....
    main线程=> 1x:  success

2，运行一个简单的没有返回值的异步任务

CompletableFuture<Void> future=CompletableFuture.runAsync(new Runnable() {
 @Override
 public void run() {
 try {
 System.out.println(getThreadName()+"正在执行一个没有返回值的异步任务。");
 TimeUnit.SECONDS.sleep(2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 
 }
 });
 future.get();
 System.out.println(getThreadName()+" 结束。");

输出如下：

ForkJoinPool.commonPool-worker-1线程=> 正在执行一个没有返回值的异步任务。
    main线程=>  结束。

从上面我们可以看到CompletableFuture默认运行使用的是ForkJoin的的线程池。当然，你也可以用lambda表达式使得代码更精简。

3，运行一个有返回值的异步任务

CompletableFuture<String> future=CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 
 try {
 System.out.println(getThreadName()+"正在执行一个有返回值的异步任务。");
 TimeUnit.SECONDS.sleep(2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 
 return "OK";
 }
 });
 
 String result=future.get();
 
 System.out.println(getThreadName()+" 结果："+result);

输出结果：

ForkJoinPool.commonPool-worker-1线程=> 正在执行一个有返回值的异步任务。
    main线程=>   结果：OK

当然，上面默认的都是ForkJoinPool我们也可以换成Executor相关的Pool，其api都有支持如下：

static CompletableFuture<Void> runAsync(Runnable runnable)
 static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
 static CompletableFuture supplyAsync(Supplier supplier)
 static CompletableFuture supplyAsync(Supplier supplier, Executor executor)

### 高级的使用CompletableFuture ###

前面提到的几种使用方法是使用异步编程最简单的步骤，CompletableFuture.get()的方法会阻塞直到任务完成，这其实还是同步的概念，这对于一个异步系统是不够的，因为真正的异步是需要支持回调函数，这样以来，我们就可以直接在某个任务干完之后，接着执行回调里面的函数，从而做到真正的异步概念。

在CompletableFuture里面，我们通过thenApply(), thenAccept()，thenRun()方法，来运行一个回调函数。

（1）thenApply()

这个方法，其实用过函数式编程的人非常容易理解，类似于scala和spark的map算子，通过这个方法可以进行多次链式转化并返回最终的加工结果。

看下面一个例子：

public static void asyncCallback() throws ExecutionException, InterruptedException {
 
 CompletableFuture<String> task=CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 System.out.println(getThreadName()+"supplyAsync");
 return "123";
 }
 });
 
 CompletableFuture<Integer> result1 = task.thenApply(number->{
 System.out.println(getThreadName()+"thenApply1");
 return Integer.parseInt(number);
 });
 
 CompletableFuture<Integer> result2 = result1.thenApply(number->{
 System.out.println(getThreadName()+"thenApply2");
 return number*2;
 });
 
 System.out.println(getThreadName()+" => "+result2.get());
 
 }

输出结果：

ForkJoinPool.commonPool-worker-1线程=> supplyAsync
    main线程=> thenApply1
    main线程=> thenApply2
    main线程=>  => 246

（2）thenAccept()

这个方法，可以接受Futrue的一个返回值，但是本身不在返回任何值，适合用于多个callback函数的最后一步操作使用。

例子如下：

public static void asyncCallback2() throws ExecutionException, InterruptedException {
 CompletableFuture<String> task=CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 System.out.println(getThreadName()+"supplyAsync");
 return "123";
 }
 });
 
 CompletableFuture<Integer> chain1 = task.thenApply(number->{
 System.out.println(getThreadName()+"thenApply1");
 return Integer.parseInt(number);
 });
 
 CompletableFuture<Integer> chain2 = chain1.thenApply(number->{
 System.out.println(getThreadName()+"thenApply2");
 return number*2;
 });
 
 CompletableFuture<Void> result=chain2.thenAccept(product->{
 System.out.println(getThreadName()+"thenAccept="+product);
 });
 
 result.get();
 System.out.println(getThreadName()+"end");
 }

结果如下：

ForkJoinPool.commonPool-worker-1线程=> supplyAsync
    main线程=> thenApply1
    main线程=> thenApply2
    main线程=> thenAccept=246
    main线程=> end

（3） thenRun()

这个方法与上一个方法类似，一般也用于回调函数最后的执行，但这个方法不接受回调函数的返回值，纯粹就代表执行任务的最后一个步骤：

public  static void asyncCallback3() throws ExecutionException, InterruptedException {
            CompletableFuture.supplyAsync(()->{
                System.out.println(getThreadName()+"supplyAsync: 一阶段任务");
                return null;
            }).thenRun(()->{
                System.out.println(getThreadName()+"thenRun: 收尾任务");
            }).get();
        }

结果：

ForkJoinPool.commonPool-worker-1线程=> supplyAsync: 一阶段任务
    main线程=> thenRun: 收尾任务

这里注意，截止到目前，前面的例子代码只会涉及两个线程，一个是主线程一个是ForkJoinPool池的线程，但其实上面的每一步都是支持异步运行的，其api如下：

// thenApply() variants
 CompletableFuture thenApply(Function<? super T,? extends U> fn)
 CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn)
 CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

我们看下改造后的一个例子：

public static void asyncCallback4() throws ExecutionException, InterruptedException {
 
 CompletableFuture<String> ref1= CompletableFuture.supplyAsync(()->{
 try {
 System.out.println(getThreadName()+"supplyAsync开始执行任务1.... ");
 // TimeUnit.SECONDS.sleep(1);
 } catch (Exception e) {
 e.printStackTrace();
 }
 System.out.println(getThreadName()+"supplyAsync: 任务1");
 return null;
 });
 
 CompletableFuture<String> ref2= CompletableFuture.supplyAsync(()->{
 try {
 
 } catch (Exception e) {
 e.printStackTrace();
 }
 System.out.println(getThreadName()+"thenApplyAsync: 任务2");
 return null;
 });
 
 CompletableFuture<String> ref3=ref2.thenApplyAsync(value->{
 System.out.println(getThreadName()+"thenApplyAsync: 任务2的子任务");
 return null;
 });
 Thread.sleep(4000);
 System.out.println(getThreadName()+ref3.get());
 }

输出结果如下：

ForkJoinPool.commonPool-worker-1线程=> supplyAsync开始执行任务1.... 
    ForkJoinPool.commonPool-worker-1线程=> supplyAsync: 任务1
    ForkJoinPool.commonPool-worker-1线程=> supplyAsync: 任务2
    ForkJoinPool.commonPool-worker-2线程=> thenApplyAsync: 任务2的子任务
    main线程=> null

我们可以看到，ForkJoin池的线程1，执行了前面的三个任务，但是第二个任务的子任务，因为我们了使用也异步提交所以它用的线程是ForkJoin池的线程2，最终由于main线程处执行了get是最后结束的。

还有一点需要注意：

ForkJoinPool所有的工作线程都是守护模式的，也就是说如果主线程退出，那么整个处理任务都会结束，而不管你当前的任务是否执行完。如果需要主线程等待结束，可采用ExecutorsThreadPool，如下：

ExecutorService pool = Executors.newFixedThreadPool(5);
 final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
 ... }, pool);

（4）thenCompose合并两个有依赖关系的CompletableFutures的执行结果

CompletableFutures在执行两个依赖的任务合并时，会返回一个嵌套的结果列表，为了避免这种情况我们可以使用thenCompose来返回，直接获取最顶层的结果数据即可：

public static void asyncCompose() throws ExecutionException, InterruptedException {
 
 CompletableFuture<String> future1=CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 return "1";
 }
 });
 
 CompletableFuture<String>nestedResult = future1.thenCompose(value->
 CompletableFuture.supplyAsync(()->{
 return value+"2";
 }));
 
 System.out.println(nestedResult.get());
 }

（5）thenCombine合并两个没有依赖关系的CompletableFutures任务

CompletableFuture<Double> d1= CompletableFuture.supplyAsync(new Supplier<Double>() {
 @Override
 public Double get() {
 return 1d;
 }
 });
 CompletableFuture<Double> d2= CompletableFuture.supplyAsync(new Supplier<Double>() {
 @Override
 public Double get() {
 return 2d;
 }
 });
 CompletableFuture<Double> result= d1.thenCombine(d2,(number1,number2)->{
 return number1+number2;
 });
 
 System.out.println(result.get());

（6）合并多个任务的结果allOf与anyOf

上面说的是两个任务的合并，那么多个任务需要使用allOf或者anyOf方法。

allOf适用于，你有一系列独立的future任务，你想等其所有的任务执行完后做一些事情。举个例子，比如我想下载100个网页，传统的串行，性能肯定不行，这里我们采用异步模式，同时对100个网页进行下载，当所有的任务下载完成之后，我们想判断每个网页是否包含某个关键词。

下面我们通过随机数来模拟上面的这个场景如下：

public static void mutilTaskTest() throws ExecutionException, InterruptedException {
 //添加n个任务
 CompletableFuture<Double> array[]=new CompletableFuture[3];
 for ( int i = 0; i < 3; i++) {
 array[i]=CompletableFuture.supplyAsync(new Supplier<Double>() {
 @Override
 public Double get() {
 return Math.random();
 }
 });
 }
 //获取结果的方式一
 // CompletableFuture.allOf(array).get();
 // for(CompletableFuture<Double> cf:array){
 // if(cf.get()>0.6){
 // System.out.println(cf.get());
 // }
 // }
 //获取结果的方式二，过滤大于指定数字，在收集输出
 List<Double> rs= Stream.of(array).map(CompletableFuture::join).filter(number->number>0.6).collect(Collectors.toList());
 System.out.println(rs);
 }

结果如下：

[0.8228784717152199]

注意其中的join方法和get方法类似，仅仅在于在Future不能正常完成的时候抛出一个unchecked的exception，这可以确保它用在Stream的map方法中，直接使用get是没法在map里面运行的。

anyOf方法，也比较简单，意思就是只要在多个future里面有一个返回，整个任务就可以结束，而不需要等到每一个future结束。

CompletableFuture<String> f1=CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 try {
 TimeUnit.SECONDS.sleep(4);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "wait 4 seconds";
 }
 });
 CompletableFuture<String> f2=CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 try {
 TimeUnit.SECONDS.sleep(2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "wait 2 seconds";
 }
 });
 
 CompletableFuture<String> f3=CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 try {
 TimeUnit.SECONDS.sleep(4);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "wait 10 seconds";
 }
 });
 
 CompletableFuture<Object> result= CompletableFuture.anyOf(f1,f2,f3);
 
 System.out.println(result.get());

输出结果：

wait 2 seconds

注意由于Anyof返回的是其中任意一个Future所以这里没有明确的返回类型，统一使用Object接受，留给使用端处理。

（7）exceptionally异常处理

异常处理是异步计算的一个重要环节，下面看看如何在CompletableFuture中使用：

int age=-1;
 CompletableFuture<String> task= CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 
 if(age<0){
 throw new IllegalArgumentException("性别必须大于0");
 }
 
 if(age<18){
 return "未成年人";
 }
 
 return "成年人";
 }
 }).exceptionally(ex->{
 System.out.println(ex.getMessage());
 return "发生 异常"+ex.getMessage();
 });
 
 System.out.println(task.get());

结果如下：

java.lang.IllegalArgumentException: 性别必须大于0
    发生 异常java.lang.IllegalArgumentException: 性别必须大于0

此外还有另外一种异常捕捉方法handle，无论发生异常都会执行，示例如下：

int age=10;
 CompletableFuture<String> task= CompletableFuture.supplyAsync(new Supplier<String>() {
 @Override
 public String get() {
 
 if(age<0){
 throw new IllegalArgumentException("性别必须大于0");
 }
 
 if(age<18){
 return "未成年人";
 }
 
 return "成年人";
 }
 }).handle((res,ex)->{
 System.out.println("执行handle");
 if(ex!=null){
 System.out.println("发生异常");
 return "发生 异常"+ex.getMessage();
 }
 
 return res;
 });
 
 System.out.println(task.get());

输出结果：

执行handle
    发生异常
    发生 异常java.lang.IllegalArgumentException: 性别必须大于0

注意上面的方法如果正常执行，也会执行handle方法。

### JDK9 CompletableFuture 类增强的主要内容 ###

（1）支持对异步方法的超时调用

orTimeout()
    completeOnTimeout()

（2）支持延迟调用

Executor delayedExecutor(long delay, TimeUnit unit, Executor executor)
    Executor delayedExecutor(long delay, TimeUnit unit)

详细内容，可以参考Oracle官网文档，这里不再过多介绍。

### 总结： ###

本文主要介绍了CompletableFuture的定义，概念及在Java中使用的例子，通过CompletableFuture我们可以实现异步编程的能力，从而使得我们开发的任务可以拥有更强大的能力。