理解Java8里面CompletableFuture异步编程

我是攻城师

发布于 2018-11-23 15:44:38

16.4K0

发布于 2018-11-23 15:44:38

文章被收录于专栏：我是攻城师我是攻城师

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 <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> 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
<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> 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我们可以实现异步编程的能力，从而使得我们开发的任务可以拥有更强大的能力。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2018-10-24，如有侵权请联系 cloudcommunity@tencent.com 删除

java

本文分享自我是攻城师微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

java

登录后参与评论

0 条评论

热度