CompletableFuture——whenComplete与handle、thenApply与thenCompose的区别

89 阅读 0 评论 59 点赞

我是靠谱客的博主稳重咖啡，最近开发中收集的这篇文章主要介绍CompletableFuture——whenComplete与handle、thenApply与thenCompose的区别，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

引言

举例与说明

whenComplete不消费异常，而handle消费异常

一个关于completeExceptionally的误用：handle没能按预期处理异常？

thenCompose V.S. thenApply

引言

CompletableFuture类提供了使用函数式编程风格来编排异步处理逻辑的方案。

本文将探讨标题所述的几个方法在使用场景方面的区别。

举例与说明

whenComplete不消费异常，而handle消费异常

Two method forms support processing whether the triggering stage completed normally or exceptionally: 

Method {whenComplete} allows injection of an action regardless of outcome, otherwise preserving the outcome in its completion. 

Method {handle} additionally allows the stage to compute a replacement result that may enable further processing by other dependent stages.

着重看“ ... additonally allows ...”部分，该部分说明两者的重要区别在于handle可进行结果的“替换”。如果handle可以把“异常”替换为一个正常结果，那么可以视作handle“消费”了异常。

//Case4 handle
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(()->{
            return new ArrayList<String>().get(1);
        }).handle((res, exp)->{
            if(res!=null) return res;
            else {
                System.out.println(exp.getClass());
                return "Error";
            }
        });

        System.out.println("--Perform cf.get()--");
        System.out.println(cf1.get());

输出结果为：

class java.util.concurrent.CompletionException
--Perform cf.get()--
Error

"Error"被输出，说明handle方法处理掉了被抛出的数组越界异常。

对于whenComplete方法，示例代码如下所示：

        //Case2 whenComplete + thenApply
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(()->{
            return new ArrayList<String>().get(1);
        }).whenComplete((res, exp) -> {
            if(res==null){
                System.out.println("whenComplete Action: "+exp.getClass());
            }else{
                System.out.println("whenComplete Action: "+res);
            }
        }).thenApply((res) -> {
            res+=" dealt by thenApply";
            return res;
        });

        System.out.println("--Perform cf1.get()--");
        System.out.println(cf1.get());

输出结果：

whenComplete Action: class java.util.concurrent.CompletionException
--Perform cf1.get()--
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.IndexOutOfBoundsException: Index: 1, Size: 0
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
	at com.company.ThreadPoolTest.main(ThreadPoolTest.java:36)
Caused by: java.lang.IndexOutOfBoundsException: Index: 1, Size: 0
	at java.util.ArrayList.rangeCheck(ArrayList.java:653)
	at java.util.ArrayList.get(ArrayList.java:429)
	at com.company.ThreadPoolTest.lambda$main$0(ThreadPoolTest.java:23)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
	at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

Process finished with exit code 1

可见whenCompleted不会消费异常，thenApply同样不会。于是代码片段中作为depedentStage的cf1.get() 抛出了异常。

一个关于completeExceptionally的误用：handle没能按预期处理异常？

以下写法中，handle看似不会“消费”被抛出的异常：

//Case completeExceptionally + handle
          CompletableFuture<String> cf1 = new CompletableFuture<String>().handle((res, exp)->{
            if(res!=null) return res;
            else {
                System.out.println(exp.getClass());
                return "Error";
            }
        });

        cf1.completeExceptionally(new Exception("Handle is not triggered"));
        System.out.println("--Perform cf.get()--");
        System.out.println(cf1.get());

而实际上，这是对cf1的错误理解：此处的cf1并没有指向最原初的stage即new CompletableFuture<String>，而是指向了handle对应的dependentStage。本发挥“消费异常”作用的handle自身抛出异常了，自然也就没有”捕获异常“这个说法了。

既然completeExceptionally可以对dependentStage直接操作，那么以下代码段

        //Case completeExceptionally + outBoundException + handle
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(()->{
            try {
                Thread.sleep(99);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new ArrayList<String>().get(1);
        }).handle((res, exp)->{
            if(res!=null) return res;
            else {
                System.out.println(exp.getClass());
                return "Error";
            }
        });
        Thread.sleep(100);
        cf1.completeExceptionally(new Exception("Handle is not triggered"));
        System.out.println("--Perform cf.get()--");
        System.out.println(cf1.get());

存在两种可能的输出结果：

//completeExceptionally先执行完成，handle没有成功消费异常

--Perform cf.get()--
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.Exception: Handle is not triggered
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
	at com.company.ThreadPoolTest.main(ThreadPoolTest.java:33)
Caused by: java.lang.Exception: Handle is not triggered
	at com.company.ThreadPoolTest.main(ThreadPoolTest.java:31)

//越界访问访问先完成，completeExceptionally执行无效。

class java.util.concurrent.CompletionException
--Perform cf.get()--
Error

这一现象对应了文档中的这一句话：

 * <p>When two or more threads attempt to
 * {@link #complete complete},
 * {@link #completeExceptionally completeExceptionally}, or
 * {@link #cancel cancel}
 * a CompletableFuture, only one of them succeeds.

thenCompose V.S. thenApply

参考StackOverFlow

个人认为thenCompose没有非其不可的使用场合，它更像是保证代码简洁度的“语法糖”。

具体代码示例如下所示：

public class CompletableFutureTest {

    public static String syncProcess(String target) {
        return target + " Dealt by syncProcess";
    }

    public static CompletableFuture<String> asyncProcess(String target) {
        return CompletableFuture.supplyAsync(() -> {
            return target + " Dealt by asyncProcess";
        });
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {


        //Use apply & syncProcess
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            return "cf1 test string";
        }).thenApply(CompletableFutureTest::syncProcess);
        // "cf1 test string Dealt by syncProcess"
        System.out.println(cf1.get());

        //Use apply & asyncProcess

        //产生completableFuture nested现象，使代码简洁度降低。
        CompletableFuture<CompletableFuture<String> > cf2 = CompletableFuture.supplyAsync(() -> {
            return "cf2 test string";
        }).thenApply(
                //如果使用 ThreadPoolTest.asyncProcess(res).get(); ，则需要补充捕获get方法异常的逻辑，不是个好选择。
                res -> {
                    return CompletableFutureTest.asyncProcess(res);
                }
        );
        //"cf2 test string Dealt by asyncProcess"
        System.out.println(cf2.get().get());

        //Use compose, now code is clean!
        CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(()->{
            return "cf3 test string";
        }).thenCompose(CompletableFutureTest::asyncProcess);
        //"cf3 test string Dealt by asyncProcess"
        System.out.println(cf3.get());
    }
}

在实际工作中，syncProcess与asyncProcess可以视作三方库提供的、封装好了的函数接口，对于asyncProcess方法，使用thenCompose来令其处理cf3的结果，代码更加整洁。