CompletableFuture API详解

  这篇文章介绍了CompletableFuture 类的功能和一些使用实例。在我们介绍开始之前,先来了解一下这个类的背景。在JAVA中,一个异步任务的调用可以使用Threads。然而,为了获得最佳性能,需要仔细规划业务流程中的各个步骤的编排,这对于不了解JAVA整个并发体系的人来说,非常容易出错。如果JAVA提供了一个即用的容器来连接一系列任务,并且能为任务的运行提供并发性但是却不用编写复杂的多线程代码呢?CompletableFuture就是这样一个别致的小东西。

创建CompletableFuture对象。

  我们可以直接new一个对象出来,也可以使用CompletableFuture为我们提供的静态方法。
  注意:这种方法直接new出来的CompletableFuture对象是无法运行的,因为他并没有处于一个“完成”状态,也就是说你调用get()方法是会被阻塞的。

CompletableFuture futrue = new CompletableFuture();

  推荐下面这种方法,使用CompletableFuture的静态方法completedFuture(U value)。直接会拿到一个“完成”状态的对象,当用get()方法拿值时,你拿到的值也就是value。

String expectedValue = "the expected value";
CompletableFuture<String> alreadyCompleted = CompletableFuture.completedFuture(expectedValue);
assertThat(alreadyCompleted.get(), is(expectedValue));

开始运行我们的第一个Task

  • runAsync(Runnable)
  • runAsync(Runnable,Executor)
  • supplyAsync(Supplier<U>)
  • supplyAsync(Supplier<U>,Executor)

这里有2种方法来初始化我们的异步任务——使用runAsync()或者supplyAsnync()。先上一段代码:

CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
         System.out.printf("[%s] I am Cool\n", Thread.currentThread().getName());
     });

CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
         System.out.printf("[%s] Am Awesome\n", Thread.currentThread().getName());
         return null;
      });

打印结果:

[ForkJoinPool.commonPool-worker-3] I am Cool
[ForkJoinPool.commonPool-worker-3] Am Awesome

  从上面的例子中可以看出,有2种初始化CompletableFuture对象并运行我们的异步任务的方法。使用runAsync()supplyAsync()。可以很容易的看出来两者之间的差别,supplyAsync()有返回值,这个返回值可以用于被下一个任务链结点所消费,后面我们会讲到。除此之外,上述方法还提供了重载方法,当我们传入Executor时,该Task会使用传入的Executor去执行,否则默认去执行任务的线程池就是fork-join thread pool,关于该线程池,暂不赘述。

PS:我个人觉得supplyAsync方法中传入Callable比传入Supplier更合适。它们俩都是函数式接口,但是Callable和异步任务的联系更紧,并且可以抛出非运行时异常。

构造任务链

  上述方法中我们只是异步的去执行了一个任务,如果我们想拿到这个任务的执行结果,并执行后面的任务呢?或者当该任务运行抛出异常时我们想来处理这些异常时呢(后面讲)?
  CompletableFuture为我们提供了几十种方法来构造任务链,这些任务链的构造过程类似于Stream.map()方法,下面将结合实例详细讲解。

  • thenAccept(Consumer<T>)
    该方法是非静态方法,是用来消费上一个任务运行之后的结点的返回值的。可以将该方法中的Consumer参数视为上一个任务在完成时要调用的回调函数。
      

Tip:该任务有两种方式来执行。第一种是如果当前线程调用thenAccept方法时,上一个任务还没执行完成时(这并不影响任务链执行的顺序性,因为这些任务都被存放在一个容器中),这时候调用此任务的线程就是上一个执行上任务的线程(ForkJoinPool-Thread);第二种是如果当前线程调用该方法时,上一个任务执行完成,这时候调用到了thenAccept方法,那么此任务会被调用thenAccept方法的线程(Main)所调用

请看下面的例子:

while (true) {
            {
                CompletableFuture cf = CompletableFuture.supplyAsync(() -> "I am Cool").thenAccept(msg ->
                        System.out.printf("[%s] %s and am also Awesome\n", Thread.currentThread().getName(), msg));
                try {
                    cf.get();
                } catch (Exception ex) {
                    ex.printStackTrace(System.err);
                }
            }
        }

可以看到有多个运行结果:

[ForkJoinPool.commonPool-worker-2] I am Cool and am also Awesome
[main] I am Cool and am also Awesome
  • thenApply(Function<T, U>)
    该方法接收一个Function参数,T类型是上一个结点传入,需要被消费的值,U类型是生成的,会被输出的值。此Function会在上一个任务执行结束时被调用。调用该Function的线程使用规则请参考上述的Tip。以下是调用该线程的实例:
CompletableFuture cf = CompletableFuture.supplyAsync(() -> "I'm Awesome")
/*这里的msg就是上一个异步任务的返回结果:I'm Awesome*/
        .thenApply(msg -> String.format("%s and am Super COOL !!!", msg))
        .thenAccept(msg -> System.out.printf("%s\n", msg));

输出结果:

I'm Awesome and am Super COOL !!!

对比于thenApply,thenAccept更适用于作为任务链的结尾。

  • thenCombine(CompletionStage<U>, BiFunction<T, U, R>)
    该方法能组合2个彼此独立的异步任务的输出。该方法接收2个参数,一个CompletionStage引用和一个BiFunction,类型T和类型U就是2个异步任务的输入,类型R为输出值。BiFunction会在上一个任务和传入第一个参数(CompletionStage)完成时被调用。下面给出实例代码:
ExecutorService executor = Executors.newFixedThreadPool(4);
CompletableFuture cf = CompletableFuture.supplyAsync(() -> "I'm Stunning", executor)
        .thenCombineAsync(CompletableFuture.supplyAsync(() -> "am New !!!"),
            (s1, s2) -> String.format("%s AND %s", s1, s2), executor)
         .thenAcceptAsync(msg -> 
                System.out.printf("[%s] %s\n", Thread.currentThread().getName(), msg), executor);

输出结果:

[pool-1-thread-3] I'm Stunning AND am New !!!

注意这里用的是自定义线程池。当线程池在任务执行结束之前shutdown则会抛出rejectedExecution。

  • thenCompose(Function<T, CompletionStage<U>>)
    该方法接收一个Function,也是用来消费上一个任务结点的,不过返回的是CompletionStage<U>,当这个CompletionStage<U>执行结束时,它会返回一个类型U的值。请看下面代码:
 CompletableFuture cf = CompletableFuture.supplyAsync(() -> "I'm Smart")
         .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " & am NIMBLE !!!"))
         .thenAccept(msg -> 
         System.out.printf("[%s] %s\n", Thread.currentThread().getName(), msg));

输出结果:

I'm Smart & am NIMBLE !!!

thenCompose VS thenApply
这两个方法都是接收一个参数并且返回的是CompletableFuture对象。它们的不同之处在于Function的返回值,thenCompose返回的是CompletableFuture,是一个你自己已经包装好的对象;而thenApply返回的是值,它底层会将这个值包装成对象返回给你。这就类似于Optianal中faltMap()和map()之间的区别。如果你想链接一个已经存在的返回CompletableFuture的方法,thenCompose是一个更好的选择,如下:

CompletableFuture<Integer> computeAnother(Integer i){
    return CompletableFuture.supplyAsync(() -> 10 + i);
}
CompletableFuture<Integer> finalResult = compute().thenCompose(this::computeAnother);
  • thenAcceptBoth(CompletionStage<U>, BiConsumer<T, U>)
    该方法作用如其名,该方法的功能可以为2个不相关的并行执行的异步任务执行完成时,提供回调。第一个传入的参数CompletionStage<U>就是其中一个需要执行的异步任务,BiConsumer<T, U>这个参数就是当2个异步任务都执行完成时执行的回调函数。类型T是上一个任务的返回值,U就是传入的任务完成时的返回值。下面请看代码:
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> "I am Fast");
CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> "am Nimble !!!");
CompletableFuture cf3 = cf1.thenAcceptBoth(cf2, (s1, s2) -> 
System.out.printf("[%s] %s and %s\n", Thread.currentThread().getName(), s1, s2));

输出结果:

[main] I am Fast and am Nimble !!!
  • acceptEither(CompletionStage<T>, Consumer<T>)
    该方法接收2个参数:一个CompletionStage的引用和一个Consumer函数式接口。该回调函数会在上一个任务执行完成第一个传入的参数执行完成时被调用。返回值为CompletableFuture<Void>。下面请看代码:
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
        randomDelay();
        return "I am Awesome";
            });
CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
        randomDelay();
        return "I am Cool";
            });
CompletableFuture cf3 = cf1.acceptEither(cf2, msg ->
System.out.printf("[%s] %s and am NIMBLE !!!\n", Thread.currentThread().getName(), msg));

可能的输出结果:

[ForkJoinPool.commonPool-worker-9] I am Awesome and am NIMBLE !!!
或者
[ForkJoinPool.commonPool-worker-2] I am Cool and am NIMBLE !!!
  • applyToEither(CompletionStage<T>, Function<T, U>)该方法和上述方法类型,两个异步任务的其中一个执行完成时会调用Function,不过返回值为CompletableFuture<U>。
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
      randomDelay();
      return "I am Awesome";
      });
CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
      randomDelay();
      return "I am Bold";
      });
CompletableFuture cf3 = cf1.applyToEither(cf2, msg -> String.format("%s and am Cool !!!", msg))
            .thenAccept(msg -> System.out.printf("[%s] %s\n", Thread.currentThread().getName(), msg));

可能出现以下结果

[ForkJoinPool.commonPool-worker-9] I am Awesome and am Cool !!!
或者
[ForkJoinPool.commonPool-worker-2] I am Bold and am Cool !!!

值得注意的是:以上两种"Either"方法在执行完成时,另一个还没执行完的任务会继续执行。

还有2个静态方法和Either,Both类方法大同小异
  • CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) 会执行cfs中所有的异步任务。
  • CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) 返回cfs中第一个执行完成的任务,其他任务都会继续执行。此处便不再赘述。

下面来考一考大家,请看下面一道代码题,输出该任务链的最终值:

        Function<String,CompletableFuture<String>> upperCaseFunction = s -> CompletableFuture.completedFuture(s.toUpperCase());
        CompletableFuture<String> stage1 = CompletableFuture.completedFuture("the quick ");
        CompletableFuture<String> stage2 = CompletableFuture.completedFuture("brown fox ");
        CompletableFuture<String> stage3 = stage1.thenCombine(stage2,(s1,s2) -> s1+s2);
        CompletableFuture<String> stage4 = stage3.thenCompose(upperCaseFunction);
        //simulatedTask第一个参数为执行时间,第二个参数为返回值。
        CompletableFuture<String> stage5 = CompletableFuture.supplyAsync(simulatedTask(2,"jumped over"));
        CompletableFuture<String> stage6 = stage4.thenCombineAsync(stage5,(s1,s2)-> s1+s2,service);
        CompletableFuture<String> stage6_sub_1_slow = CompletableFuture.supplyAsync(simulatedTask(4,"fell into"));
        CompletableFuture<String> stage7 = stage6.applyToEitherAsync(stage6_sub_1_slow,String::toUpperCase,service);
        CompletableFuture<String> stage8 = CompletableFuture.supplyAsync(simulatedTask(3," the lazy dog"),service);
        CompletableFuture<String> finalStage = stage7.thenCombineAsync(stage8,(s1,s2)-> s1+s2,service);

答案会放在文末。

至此,我们讲述了绝大大部分构造任务链的方法,这些方法能让我们不断地向后传递不同的返回值,并且保证了任务链的顺序性。

任务链完成时的回调和异常处理

在讲本节内容之前我们先来看2个方法。

  • T get() throws InterruptedException, ExecutionException
  • T join()
    这两个方法都是从CompletableFuture里面取值的,调用时会阻塞当前线程。可以很容易看出join没有抛出非运行时异常,不过它会抛出2个运行时异常:当这个任务被取消时会抛出CancellationException;当任务执行时抛出异常时,会抛出CompletionException。当我们需要从任务中拿值时,推荐join()方法。

我们知道上面两种方法在调用时都会阻塞当前线程,如果想继续向下运行,则必须中断或取消任务,但是如果我们想正常的结束任务,或者在拿值时如果值不存在,则抛出自定义的异常或返回默认值呢?CompletableFuture中为我们提供以下解决方法。

  • T getNow(T valueIfAbsent)
    该方法不阻塞,如果任务尚未完成,则返回默认值。
  • boolean complete(T)
    如果任务未完成,则在拿取返回值时返回T。如果此次调用将CompletableFuture转化为“完成”状态,则返回true,否则返回false。意思是如果通过get()或join()拿到的是类型T的值,则返回true,否则返回false。
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
                /*循环*/
                infiniteLoop();
                return "I am Awesome";
            });
System.out.println("complete:"+cf1.complete("Default"));
System.out.println("isDone:"+cf1.isDone());
System.out.println("result:"+cf1.join());

输出结果:

complete:true
isDone:true
result:Default
  • boolean completeExceptionally(Throwable)
    如果任务未完成,则在拿值时抛出Throwable异常,该异常可以自定义。如果此次调用将CompletableFuture转化为“完成”状态,则返回true,否则返回false。如果你想进一步传递某个异常,可以使用该方法。
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
       /*循环*/
       infiniteLoop();
       return "I am Awesome";
       });
System.out.println("complete:"+cf1.completeExceptionally(new RuntimeException("completeExceptionally")));
System.out.println("isDone:"+cf1.isDone());
System.out.println("isCompletedExceptionally:"+cf1.isCompletedExceptionally());
System.out.println("result:"+cf1.join());

输出结果:

complete:true
isDone:true
Exception in thread "main" 
isCompletedExceptionally:true
java.util.concurrent.CompletionException: java.lang.RuntimeException: completeExceptionally
    at java.util.concurrent.CompletableFuture.reportJoin(CompletableFuture.java:375)
    at java.util.concurrent.CompletableFuture.join(CompletableFuture.java:1934)
    at CompletableFutureSample.main(CompletableFutureSample.java:27)
Caused by: java.lang.RuntimeException: completeExceptionally
    at CompletableFutureSample.main(CompletableFutureSample.java:24)

上述方法在构建我们不想等待太多时间的健壮系统时很有用。

  • obtrudeValue(T value)

  • obtrudeException(Throwable ex)
    使用这两个方法可以强制地将值设置或者将异常抛出,无论该之前任务是否完成。这两个方法类似于complete和completeExceptionally,但是complete的功能是如果任务未完成才返回设定的值。在某些场景下,你有可能想放弃该任务的执行,发出一个失败的信号。请谨慎使用这两个方法,因为他们会覆盖前一个Future的值(或异常)。

  • CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> )
    该方法的作用是:当前面的任务出了异常时,就会返回T值;否则还是返回原先前面任务应该返回的值。请看下面的代码:

CompletableFuture<String> cf1 = CompletableFuture.complete(user).thenApply((user) -> {
      return user.getName();
      });
cf1.exceptionally(ex -> ex.getMessage())
    /*如果抛出了异常,这里传递给下一个结点的值是ex.getMessage()*/
    .thenAccept(System.out::println);
    /*如果使用get或join拿里面的值的话,如果任务有异常,会抛出CompletionException异常的*/
System.out.println("isDone:" + cf1.isDone());
System.out.println("isCompletedExceptionally:" + cf1.isCompletedExceptionally());

输出结果:

/*user为空时的打印结果*/
java.lang.NullPointerException
isDone:true
isCompletedExceptionally:true
/*正常的打印结果*/
Jack
isDone:true
isCompletedExceptionally:false
  • handle(BiFunction<T, Throwable, U>)
    这个方法有点类似于thenApply和exceptionally的结合,如果上一个任务出了异常,则传入的T类型为null,Throwable为抛出的异常;如果正常运行,则T类型为运行后的返回值,Throwable为null。
    下面用handle来还原上个例子的代码的功能:
 CompletableFuture<String> cf1 = CompletableFuture.complete(user).thenApply((user) -> {
         return user.getName();
      }).handle((v,ex)->{
          if(v==null)
              return "user is null";
          return v;
          }).thenAccept(System.out::println);
            /*如果使用get或join拿里面的值的话,如果任务有异常,会抛出CompletionException异常的*/
System.out.println("isDone:" + cf1.isDone());
            /*使用handle时,无论出不出异常,该值都为false。*/
System.out.println("isCompletedExceptionally:" + cf1.isCompletedExceptionally());
/*user为空时的打印结果*/
user is null!
isDone:true
isCompletedExceptionally:false
/*正常的打印结果*/
Jack
isDone:true
isCompletedExceptionally:false
  • whenComplete(BiConsumer<T, Throwable>)
    该方法有点像thenAccept(Consumer<T>) ,只不过增加了一个异常处理的功能。传入的参数类似于handle,如果上一个任务出了异常,则传入的T类型为null,Throwable为抛出的异常;如果正常运行,则T类型为之前任务运行后的返回值,Throwable为null。下面上代码:
CompletableFuture<String> cf1 = CompletableFuture.complete(user).thenApply((user) -> {
                return user.getName();
     }).handle((v,ex)->{
            if(v==null)
                 System.out.println("user is null");
            else
                 System.out.println(v);
       }).thenAccept(System.out::println);

输出结果和上例相同,为了简洁就不展示了。

Async Methods

CompletableFuture API为绝大部分的方法提供了2个额外的方法变体,它们后缀名都是“Async”。这些异步方法都会使用线程池来执行,如果传入的线程池为null,则还会使用默认的fork/join pool来执行任务,这可以更有效率地提高你任务的并发性。

总结

  这篇文章我们总结了CompletableFuture API的功能,从创建对象,到构造任务链,最后再到异常的处理。

答案

THE QUICK BROWN FOX JUMPED OVER the lazy dog

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 201,552评论 5 474
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 84,666评论 2 377
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 148,519评论 0 334
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,180评论 1 272
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,205评论 5 363
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,344评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,781评论 3 393
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,449评论 0 256
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,635评论 1 295
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,467评论 2 317
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,515评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,217评论 3 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,775评论 3 303
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,851评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,084评论 1 258
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,637评论 2 348
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,204评论 2 341

推荐阅读更多精彩内容

  • 在现代软件开发中,系统功能越来越复杂,管理复杂度的方法就是分而治之,系统的很多功能可能会被切分为小的服务,对外提供...
    天堂鸟6阅读 7,117评论 0 23
  • Swift1> Swift和OC的区别1.1> Swift没有地址/指针的概念1.2> 泛型1.3> 类型严谨 对...
    cosWriter阅读 11,079评论 1 32
  • Java 8 CompletableFuture Java 8 有大量的新特性和增强如 Lambda 表达式,St...
    单纯小码农阅读 2,068评论 0 8
  • Java 8 有大量的新特性和增强如 Lambda 表达式,Streams,CompletableFuture等。...
    YDDMAX_Y阅读 4,739评论 0 15
  • 为什么有男人和女人呢?他们是那样的不同,不能相互理解,但又相互爱恋、依靠,必然互相伤害。有时候我想,设计男女这样一...
    孤独小说家阅读 395评论 0 0