Java8新的异步编程方式 CompletableFuture(三)

前面两篇文章已经整理了CompletableFuture大部分的特性，本文会整理完CompletableFuture余下的特性，以及将它跟RxJava进行比较。

3.6 Either

Either 表示的是两个CompletableFuture，当其中任意一个CompletableFuture计算完成的时候就会执行。

方法名 描述
acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) 当任意一个CompletableFuture完成的时候，action这个消费者就会被执行。
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) 当任意一个CompletableFuture完成的时候，action这个消费者就会被执行。使用ForkJoinPool
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) 当任意一个CompletableFuture完成的时候，action这个消费者就会被执行。使用指定的线程池 
        Random random = new Random();

        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{

            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "from future1";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{

            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "from future2";
        });

        CompletableFuture<Void> future =  future1.acceptEither(future2,str->System.out.println("The future is "+str));

        try {
            future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

执行结果：The future is from future1 或者 The future is from future2。
因为future1和future2，执行的顺序是随机的。

applyToEither 跟 acceptEither 类似。

方法名 描述
applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn) 当任意一个CompletableFuture完成的时候，fn会被执行，它的返回值会当作新的CompletableFuture<U>的计算结果。
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn) 当任意一个CompletableFuture完成的时候，fn会被执行，它的返回值会当作新的CompletableFuture<U>的计算结果。使用ForkJoinPool
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor) 当任意一个CompletableFuture完成的时候，fn会被执行，它的返回值会当作新的CompletableFuture<U>的计算结果。使用指定的线程池 
        Random random = new Random();

        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{

            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "from future1";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{

            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "from future2";
        });

        CompletableFuture<String> future =  future1.applyToEither(future2,str->"The future is "+str);

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

执行结果也跟上面的程序类似。

3.7 其他方法

allOf、anyOf是CompletableFuture的静态方法。

3.7.1 allOf

方法名 描述
allOf(CompletableFuture<?>... cfs) 在所有Future对象完成后结束，并返回一个future。

allOf()方法所返回的CompletableFuture，并不能组合前面多个CompletableFuture的计算结果。于是我们借助Java 8的Stream来组合多个future的结果。

        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "tony");

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "cafei");

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "aaron");

        CompletableFuture.allOf(future1, future2, future3)
                .thenApply(v ->
                Stream.of(future1, future2, future3)
                        .map(CompletableFuture::join)
                        .collect(Collectors.joining(" ")))
                .thenAccept(System.out::print);

执行结果：

tony cafei aaron

3.7.2 anyOf

方法名 描述
anyOf(CompletableFuture<?>... cfs) 在任何一个Future对象结束后结束，并返回一个future。 
        Random rand = new Random();
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(rand.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "from future1";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(rand.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "from future2";
        });
        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(rand.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "from future3";
        });

        CompletableFuture<Object> future =  CompletableFuture.anyOf(future1,future2,future3);

        try {
            System.out.println(future.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

使用anyOf()时，只要某一个future完成，就结束了。所以执行结果可能是"from future1"、"from future2"、"from future3"中的任意一个。

anyOf 和 acceptEither、applyToEither的区别在于，后两者只能使用在两个future中，而anyOf可以使用在多个future中。

3.8 CompletableFuture异常处理

CompletableFuture在运行时如果遇到异常，可以使用get()并抛出异常进行处理，但这并不是一个最好的方法。CompletableFuture本身也提供了几种方式来处理异常。

3.8.1 exceptionally

方法名 描述
exceptionally(Function<Throwable,? extends T> fn) 只有当CompletableFuture抛出异常的时候，才会触发这个exceptionally的计算，调用function计算值。 
        CompletableFuture.supplyAsync(() -> "hello world")
                .thenApply(s -> {
                    s = null;
                    int length = s.length();
                    return length;
                }).thenAccept(i -> System.out.println(i))
                .exceptionally(t -> {
                    System.out.println("Unexpected error:" + t);
                    return null;
                });

执行结果：

Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException

对上面的代码稍微做了一下修改，修复了空指针的异常。

        CompletableFuture.supplyAsync(() -> "hello world")
                .thenApply(s -> {
//                    s = null;
                    int length = s.length();
                    return length;
                }).thenAccept(i -> System.out.println(i))
                .exceptionally(t -> {
                    System.out.println("Unexpected error:" + t);
                    return null;
                });

执行结果：

11

3.8.2 whenComplete

whenComplete 在上一篇文章其实已经介绍过了，在这里跟exceptionally的作用差不多，可以捕获任意阶段的异常。如果没有异常的话，就执行action。

        CompletableFuture.supplyAsync(() -> "hello world")
                .thenApply(s -> {
                    s = null;
                    int length = s.length();
                    return length;
                }).thenAccept(i -> System.out.println(i))
                .whenComplete((result, throwable) -> {

                    if (throwable != null) {
                       System.out.println("Unexpected error:"+throwable);
                    } else {
                        System.out.println(result);
                    }

                });

执行结果：

Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException

跟whenComplete相似的方法是handle，handle的用法在上一篇文章中也已经介绍过。

四. CompletableFuture VS Java8 Stream VS RxJava1 & RxJava2

CompletableFuture 有很多特性跟RxJava很像，所以将CompletableFuture、Java 8 Stream和RxJava做一个相互的比较。

composable lazy resuable async cached push back pressure
CompletableFuture 支持 不支持 支持 支持 支持 支持 不支持
Stream 支持 支持 不支持 不支持 不支持 不支持 不支持
Observable(RxJava1) 支持 支持 支持 支持 支持 支持 支持
Observable(RxJava2) 支持 支持 支持 支持 支持 支持 不支持
Flowable(RxJava2) 支持 支持 支持 支持 支持 支持 支持

五. 总结

Java 8提供了一种函数风格的异步和事件驱动编程模型CompletableFuture，它不会造成堵塞。CompletableFuture背后依靠的是fork/join框架来启动新的线程实现异步与并发。当然，我们也能通过指定线程池来做这些事情。

CompletableFuture特别是对微服务架构而言，会有很大的作为。举一个具体的场景，电商的商品页面可能会涉及到商品详情服务、商品评论服务、相关商品推荐服务等等。获取商品的信息时（/productdetails?productid=xxx），需要调用多个服务来处理这一个请求并返回结果。这里可能会涉及到并发编程，我们完全可以使用Java 8的CompletableFuture或者RxJava来实现。

先前的文章：
Java8新的异步编程方式 CompletableFuture(一)
Java8新的异步编程方式 CompletableFuture(二)

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 134,845评论 18赞 139
  • Java8新的异步编程方式 CompletableFuture(一)
    一. Future JDK 5引入了Future模式。Future接口是Java多线程Future模式的实现，在j...
    fengzhizi715阅读 86,227评论 3赞 25
  • Android - 收藏集
    Android 自定义View的各种姿势1 Activity的显示之ViewRootImpl详解 Activity...
    passiontim阅读 172,914评论 25赞 708
  • 人的一生
    人总是能被故事吸引，每个人10年，20年…甚至一辈子，都在上演一部未完结的电影或电视剧，直到生命终结。故事有跌宕...
    艺悬明阅读 169评论 0赞 1
  • 我的第一页简书
    简书是一个很好的阅读、写作社区。关注简书有一段日子了，也从中读到了许多的好文章，学到了许多有用的知识，感谢！ 在下...
    猿球崛起阅读 336评论 0赞 0