异步利刃CompletableFuture

2023-05-23 70

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 异步利刃CompletableFuture

什么是CompletableFuture?

CompletableFuture 类实现了 Future 和 CompletionStage 接口并且新增了许多方法，它支持 lambda，通过回调利用非阻塞方法，提升了异步编程模型。简单来说可以帮我们实现任务编排。【文中所有代码已上传码云】

CompletableFuture的创建

先来看一个创建例子，后续再展开说说这个类中的方法：

CompletableFuture<String>completableFuture=newCompletableFuture<>();
completableFuture.complete("Hello CompletableFuture");
System.out.println(completableFuture.get());

需要注意的是当我们对不完整的 CompleteableFuture调用 get 方法的话，会由于 Future 未完成，因此 get 调用会一直阻塞。

创建异步任务

CompletableFuture 提供了四个静态方法来创建异步任务。

publicstatic<U>CompletableFuture<U>supplyAsync(Supplier<U>supplier);
publicstatic<U>CompletableFuture<U>supplyAsync(Supplier<U>supplier, Executorexecutor);
publicstaticCompletableFuture<Void>runAsync(Runnablerunnable);
publicstaticCompletableFuture<Void>runAsync(Runnablerunnable, Executorexecutor);

没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码；如果指定线程池，则使用指定的线程池运行。

staticExecutorscreenExecutor(Executore) {
if (!useCommonPool&&e==ForkJoinPool.commonPool())
returnasyncPool;
if (e==null) thrownewNullPointerException();
returne;
    }

二者的区别很明显，supplyAsync有返回值，runAsync方法无返回值。我们通过静态方法会立刻开启异步线程执行Supplier或者Runnable提交的任务。任务执行完成，就可以打印返回值，不再需要其它线程主动调用complete来表示任务执行完成。

获取任务执行结果

publicTget();
publicTget(longtimeout, TimeUnitunit);
publicTgetNow(TvalueIfAbsent);
publicTjoin();

get()和get(long timeout, TimeUnit unit)是实现了Future接口的功能，两者主要区别就是get()会一直阻塞直到获取到结果，get(long timeout, TimeUnit unit)值可以指定超时时间，当到了指定的时间还未获取到任务，就会抛出TimeoutException异常。

getNow(T valueIfAbsent)：就是获取任务的执行结果，但不会产生阻塞。如果任务还没执行完成，那么就会返回你传入的 valueIfAbsent 参数值，如果执行完成了，就会返回任务执行的结果。

join()：跟get()的主要区别就是，get()会抛出检查时异常，join()不会。

任务完成后的处理

whenComplete

publicCompletableFuture<T>whenComplete(BiConsumer<?superT,?superThrowable>action)
publicCompletableFuture<T>whenCompleteAsync(BiConsumer<?superT,?superThrowable>action)
publicCompletableFuture<T>whenCompleteAsync(BiConsumer<?superT,?superThrowable>action, Executorexecutor)
publicCompletableFuture<T>exceptionally(Function<Throwable,?extendsT>fn)

whenComplete是等任务完成后继续执行whenComplete的action，这里也能看出执行action的是主线程

如果有异常，当主线程在获取任务结果时就会抛出异常。

而whenCompleteAsync是通过异步线程去执行action

exceptionally

任务执行过程中出现异常的时候，会回调exceptionally方法指定的回调，但是如果没有出现异常，是不会回调的。

thenApply

当线程B依赖于线程A的执行结果时，可以使用thenApply方法来把这两个线程串行化

public<U>CompletableFuture<U>thenApply(Function<?superT,?extendsU>fn)
public<U>CompletableFuture<U>thenApplyAsync(Function<?superT,?extendsU>fn)
public<U>CompletableFuture<U>thenApplyAsync(Function<?superT,?extendsU>fn, Executorexecutor)

将supplyAsync方法里的结果（Integer）作为thenApply方法里Function接口的apply方法的入参

如果有异常thenApply就不会执行

handle

public<U>CompletionStage<U>handle(BiFunction<?superT, Throwable, ?extendsU>fn);
public<U>CompletionStage<U>handleAsync(BiFunction<?superT, Throwable, ?extendsU>fn);
public<U>CompletionStage<U>handleAsync(BiFunction<?superT, Throwable, ?extendsU>fn,Executorexecutor);

不同于thenApply的是handle里的方法是在supplyAsync/runAsync执行后一定会执行。

thenAccept

publicCompletionStage<Void>thenAccept(Consumer<?superT>action);
publicCompletionStage<Void>thenAcceptAsync(Consumer<?superT>action);
publicCompletionStage<Void>thenAcceptAsync(Consumer<?superT>action,Executorexecutor);

不同于上述的方法，thenAccept纯消费无返回值。

组合任务

thenCompose

这个方法也是线程B需要用到线程A的结果，不同于thenApply的是：thenCompose()用来连接两个CompletableFuture，返回值是新的CompletableFuture；thenApply()转换的是泛型中的类型，是同一个CompletableFuture。

public<U>CompletableFuture<U>thenCompose(Function<?superT,?extendsCompletionStage<U>>fn)
public<U>CompletableFuture<U>thenComposeAsync(Function<?superT,?extendsCompletionStage<U>>fn)
public<U>CompletableFuture<U>thenComposeAsync(Function<?superT,?extendsCompletionStage<U>>fn, Executorexecutor)

thenCombine

会把两个CompletableFuture的任务都执行完成后把结果一块交给thenCombine来处理，并生成新的CompletableFuture任务。

public<U,V>CompletionStage<V>thenCombine(CompletionStage<?extendsU>other,BiFunction<?superT,?superU,?extendsV>fn);
public<U,V>CompletionStage<V>thenCombineAsync(CompletionStage<?extendsU>other,BiFunction<?superT,?superU,?extendsV>fn);
public<U,V>CompletionStage<V>thenCombineAsync(CompletionStage<?extendsU>other,BiFunction<?superT,?superU,?extendsV>fn,Executorexecutor);

谁调用的thenCombine，则BiFunction的第一个参数就是它的结果。

其他方法

allOf

allOf方法的入参是多个CompletableFuture任务，返回类型是CompletableFuture<Void>，allOf方法是等所有的CompletableFuture都执行完后再执行计算，一般后面会跟链式的thenApply方法或者thenAccept方法对所有的异步任务进行汇总处理。