Java8并发新特性CompletableFuture

网友投稿 314 2022-07-26


目录1.CompletableFuture是什么?2.CompletableFuture的方法使用说明2.1 CompletableFuture类提供几个静态方法来进行异步操作2.2 获取异步任务执行结果的方法 get()/join()3.CompletionStage的方法使用说明3.1 纯消费类型3.2 有返回值类型3.3 不消费也不返回类型3.4 组合类型3.5 任务事件类型4.CompletionStage异常处理方法5.方法类型总结

1.CompletableFuture是什么?

各位小伙伴是否有一种感觉,整天大量时间沉迷于业务开发的同时,缺少对于一些技术更新的关注,忽略掉了很多实用又简单的方法,以往我们做异步任务的时候都习惯于使用Callable或者Runnable接口去实现,今天我们就来聊聊与之不同的CompletableFuture类。

CompletableFuture针对Future接口做了改进,相比Callable/Runnable接口它支持多任务进行链式调用、组合、多任务并发处理。很多时候我们在设计过程中会想在一个异步任务执行完成后,直接获取它的结果传递给下一个任务继续执行后续的流程,这时候CompletableFuture的作用就来了。

CompletableFuture类关系图:

从以下类图可以看到,CompletableFuture实现了Future和CompletionStage两个接口,Future提供了获取任务执行结果和任务执行状态的功能。 CompletionStage表示一个任务的执行阶段,提供了诸多方法支持了多任务的聚合功能。

2.CompletableFuture的方法使用说明

2.1 CompletableFuture类提供几个静态方法来进行异步操作

supplyAsync与runAsync主要用于构建异步事件。

supplyAsync:带有返回值的异步任务,支持在默认线程池ForkJoinPool.commonPool()中完成异步任务,也可以使用自定义线程池执行异步任务,结果返回一个新的CompletableFuture,返回结果类型U。最终的任务执行结果可通过返回CompletableFuture对象的 get()/join() 方法获取返回值。

// 使用默认线程池

public static CompletableFuture supplyAsync(Supplier supplier) {...}

// 使用自定义线程池Executor

public static gbKxnPF CompletableFuture supplyAsync(Supplier supplier, Executor executor) {...}

// ====================================demo华丽分割线============================================

CompletableFuture supplyAsyncFuture = CompletableFuture.supplyAsync(() -> {

log.info("executing supplyAsync task ...");

return "this is supplyAsync";

});

// 进入阻塞获取异步任务结果

log.info(supplyAsyncFuture.get()); // 输出结果:this is supplyAsync

runAsync:不带返回值的异步任务,支持在默认线程池ForkJoinPool.commonPool()中完成异步任务,也可以使用自定义线程池执行异步任务,结果返回一个新的CompletableFuture,返回结果类型为Void,也就是无返回值。

public static CompletableFuture runAsync(Runnable runnable) {...}

public static CompletableFuture runAsync(Runnable runnable, Executor executor) {...}

// ====================================demo华丽分割线============================================

CompletableFuture runAsyncFuture = CompletableFuture.runAsync(() -> {

log.info("executing runAsync task ...");

});

runAsyncFuture.get();

allOf:多个CompletableFuture任务并发执行,所有CompletableFuture任务完成时,返回一个新的CompletableFuture对象,其返回值为Void,也就是无返回值。

public static CompletableFuture allOf(CompletableFuture>... cfs) {...}

// ====================================demo华丽分割线============================================

// allOf,可传递返回值不同类型的future,最终结果按自己设计预期处理即可

CompletableFuture cf11 = CompletableFuture.supplyAsync(() -> {

log.info("executing supplyAsync task cf11 ...");

return "this is supplyAsync";

});

CompletableFuture cf12 = CompletableFuture.supplyAsync(() -> {

log.info("executing supplyAsync task cf12 ...");

return "this is supplyAsync";

});

CompletableFuture allOfFuture = CompletableFuture.allOf(cf11, cf12);

allOfFuture.get();

anyOf:多个CompletableFuture任务并发执行,只要有一个CompletableFuture任务完成时,就会返回一个新的CompletableFuture对象,并返回该CompletableFuture执行完成任务的返回值。

public static CompletableFuture anyOf(CompletableFuture>... cfs) {...}

// ====================================demo华丽分割线============================================

CompletableFuture cf21 = CompletableFuture.supplyAsync(() -> {

log.info("executing supplyAsync task cf21 ...");

return "this is supplyAsync cf21";

});

CompletableFuture cf22 = CompletableFuture.supplyAsync(() -> {

log.info("executing supplyAsync task cf22 ...");

return "this is supplyAsync cf22";

});

CompletableFuture anyOfFuture = CompletableFuture.anyOf(cf21, cf22);

log.info("{}", anyOfFuture.get()); // 输出结果:this is supplyAsync cf21或cf22

2.2 获取异步任务执行结果的方法 get()/join()

join()和get()方法都是CompletableFuture对象基于阻塞的方式来获取异步任务执行结果。

get方法会抛出显示异常必须捕获处理,任务允许被中断抛出InterruptedException异常,通过带有超时时间的阻塞方式获取异步任务执行结果,超时等待无结果则中断任务抛出TimeoutException异常。join方法会抛出未检查异常,与get()方法不同的是join()方法不允许被中断。

// 可中断,可设置超时时间

public T get() throws InterruptedException, ExecutionException {...}

public T get(long timeout, TimeUnit unit) throws InterruptedException,

ExecutionException, TimeoutException {...}

/**

* 不可中断

*/

public T join() {...}

3.CompletionStage的方法使用说明

CompletionStage表示一个任务的执行阶段,每个任务都会返回一个CompletionStage对象,可以对多个CompletionStage对象进行串行、并行或者聚合的方式来进行下阶段的操作,也就是说实现异步任务的回调功能。CompletionStage总共提供了38个方法来实现多个CompletionStage任务的各种操作, 接下来我们就针对这些方法分类来了解一下。

以下类型均有三种使用方式:

thenAccept:方法名不带Async的使用主线程同步执行回调函数,不做异步处理thenAcceptAsync:方法名带Async,但是无executor参数的,使用默认线程池ForkJoinPool.commonPool异步执行任务thenAcceptAsync:方法名带Async,有executor参数的,使用自定义线程池异步执行任务

3.1 纯消费类型

依赖单个任务完成(thenAccept):由上一个CompletionStage任务执行完成的结果传递到action进行回调处理,即仅仅消费了上一个CompletionStage任务的返回值,回调处理结果无返回值。

// 不使用线程池,仅依赖当前线程执行,不做异步

public CompletionStage thenAccept(Consumer super T> action);

// 使用默认线程池ForkJoinPool.commonPool执行任务

public CompletionStage thenAcceptAsync(Consumer super T> action);

// 使用自定义线程池执行任务

public CompletionStage thenAcceptAsync(Consumer super T> action, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture.supplyAsync(() -> "this is supplyAsync")

.thenAcceptAsync((result) -> {

log.info("{} thenAcceptAsync", result);

}).join();

// 输出结果:this is supplyAsync thenAcceptAsync

依赖两个任务都完成(thenAcceptBoth):两个CompletionStage任务并发执行,必须都完成了才执行action回调处理,即仅仅消费了两个CompletionStage任务的返回值,回调处理结果无返回值。

/**

* 额外多了CompletionStage参数表示CompletionStage任务依赖的另一个CompletionStage任务

* action接收两个参数,分别表示两个CompletionStage任务的返回值

*/

public CompletionStage thenAcceptBoth(CompletionStage extends U> other,

BiConsumer super T, ? super U> action);

// 原理同上,使用默认线程池执行异步任务

public CompletionStage thenAcceptBothAsync(CompletionStage extends U> other,

BiConsumer super T, ? super U> action);

// 原理同上,使用自定义线程池执行异步任务

public CompletionStage thenAcceptBothAsync(CompletionStage extends U> other,

BiConsumer super T, ? super U> action, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture cf311 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf311");

CompletableFuture cf312 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf312");

cf311.thenAcceptBothAsync(cf312, (r1, r2) -> {

log.info("{} and {}", r1, r2);

}).join();

// 输出结果:this is supplyAsync cf311 and this is supplygbKxnPFAsync cf312

依赖两个任务中的任何一个完成(acceptEither):两个CompletionStage任务并发执行,只要其中一个先完成了就携带返回值执行action回调处理,即仅仅消费了优先完成的CompletionStage任务的返回值,回调处理结果无返回值。

/**

* 类似thenAcceptBothAsync,只不过acceptEither只需两个任务中的其中一个完成即可回调action

* action中的值为两个任务中先执行完任务的返回值

*/

public CompletionStage acceptEither(CompletionStage extends T> other,

Consumer super T> action);

public CompletionStage acceptEitherAsync(CompletionStage extends T> other,

Consumer super T> action);

public CompletionStage acceptEitherAsync(CompletionStage extends T> other,

Consumer super T> action, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture cf311 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf311");

CompletableFuture cf312 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf312");

cf311.acceptEitherAsync(cf312, (r) -> {

log.info(r); // 输出结果:this is supplyAsync cf311或cf312

}).join();

3.2 有返回值类型

依赖单个任务完成(thenApply):由上一个CompletionStage任务执行完成的结果传递到action进行回调处理,即不止消费了上一个CompletaionStage任务的返回值,同时回调处理结果也有返回值

public CompletionStage thenApply(Function super T,? extends U> fn);

public CompletionStage thenApplyAsync(Function super T,? extends U> fn);

public CompletionStage thenApplyAsync(Function super T,? extends U> fn, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture cf32 = CompletableFuture.supplyAsync(() -> "this is supplyAsync")

.thenApplyAsync(result -> result + " and thenApplyAsync");

log.info(cf32.join()); // 输出结果:this is supplyAsync and thenApplyAsync

依赖两个任务都完成(thenCombine):两个CompletionStage任务并发执行,必须都完成了才执行action回调处理,即不止消费了两个CompletaionStage任务的返回值,同时回调处理结果也有返回值。

public CompletionStage thenCombine(CompletionStage extends U> other,

BiFunction super T,? super U,? extends V> fn);

public CompletionStage thenCombineAsync(CompletionStage extends U> other,

BiFunction super T,? super U,? extends V> fn);

public CompletionStage thenCombineAsync(CompletionStage extends U> other,

BiFunction super T,? super U,? extends V> fn, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture cf321 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf321");

CompletableFuture cf322 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf322");

CompletableFuture thenCombineFuture = cf321.thenCombineAsync(cf322, (r1, r2) -> {

return r1 + " and " + r2;

});

log.info(thenCombineFuture.join());

// 输出结果:this is supplyAsync cf321 and this is supplyAsync cf322

依赖两个任务中的任何一个完成(applyToEither):两个CompletionStage任务并发执行,只要其中一个任务执行完成就会action回调处理,即不止消费了优先完成的CompletionStage的返回值,同时回调处理结果也有返回值。

// 原理同3.1的acceptEither,只不过applyToEither任务执行完成会返回一个带有返回值的CompletionStage

public CompletionStage applyToEither(CompletionStage extends T> other,

Function super T, U> fn);

public CompletionStage applyToEitherAsync(CompletionStage extends T> other,

Function super T, U> fn);

public CompletionStage applyToEitherAsync(CompletionStage extends T> other,

Function super T, U> fn, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture cf321 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf321");

CompletableFuture cf322 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf322");

CompletableFuture thenCombineFuture = cf321.applyToEitherAsync(cf322, (r) -> {

return r;

});

log.info(thenCombineFuture.join());

// 输出结果:this is supplyAsync cf321或cf322

3.3 不消费也不返回类型

依赖单个任务完成(thenRun):单个CompletionStage任务执行完成回调action处理,即执行action回调方法无参数,回调处理结果也无返回值。

// 上一个CompletionStage任务执行完成后直接回调action处理,无返回值

public CompletionStage thenRun(Runnable action);

// 同上,使用默认线程池执行action处理

public CompletionStage thenRunAsync(Runnable action);

// 同上,使用自定义线程池执行action处理

public CompletionStage thenRunAsync(Runnable action, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture.runAsync(() -> {

// TODO

}).thenRunAsync(() -> {

log.info("this is thenRunAsync"); // 输出结果:this is thenRunAsync

}).join();

依赖两个任务都完成(runAfterBoth):两个CompletionStage任务并发执行,必须两个任务都完成才执行action回调处理,即执行action回调方法无参数,回调处理结果也无返回值。

// 原理同3.1的thenAcceptBoth,只不过runAfterBoth的action回调处理不接收参数且任务执行完成无返回值

public CompletionStage runAfterBoth(CompletionStage> other, Runnable action);

// 同上,使用默认线程池执行action处理

public CompletionStage runAfterBothAsync(CompletionStage> other, Runnable action);

// 同上,使用自定义线程池执行action处理

public CompletionStage runAfterBothAsync(CompletionStage> other, Runnable action, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture cf331 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf331");

CompletableFuture cf332 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf332");

cf331.runAfterBoth(cf332, () -> {

log.info("this is runAfterBoth");

}).join();

// 输出结果:this is runAfterBoth

依赖两个任务中的任何一个完成(runAfterEither):两个CompletionStage任务并发执行,只需其中任何一个任务完成即可回调action处理,即执行action回调方法无参数,回调处理结果也无返回值。

public CompletionStage runAfterEither(CompletionStage> other, Runnable action);

public CompletionStage runAfterEitherAsync(CompletionStage> other, Runnable action);

public CompletionStage runAfterEitherAsync(CompletionStage> other, Runnable action, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture cf331 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf331");

CompletableFuture cf332 = CompletableFuture.supplyAsync(() -> "this is supplyAsync cf332");

cf331.runAfterEitherAsync(cf332, () -> {

log.info("this is runAfterEitherAsync");

}).join();

// 输出结果:this is runAfterEitherAsync

3.4 组合类型

thenCompose:存在先后关系的两个任务进行串行组合,由第一个CompletionStage任务执行结果作为参数传递给第二个CompletionStage任务,最终返回第二个CompletionStage。

public CompletionStage thenCompose(Function super T, ? extends CompletionStage> fn);

public CompletionStage thenComposeAsync(Function super T, ? extends CompletionStage> fn);

public CompletionStage thenComposeAsync(Function super T, ? extends CompletionStage> fn, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture supplyFuture = CompletableFuture.supplyAsync(() -> {

return "this is supplyAsync";

});

CompletableFuture thenComposeFuture = supplyFuture.thenComposeAsync((r) -> {

return CompletableFuture.supplyAsync(() -> {

return r + " and this is thenComposeAsync";

});

});

log.info(thenComposeFuture.join());

// 输出结果:this is supplyAsync and this is thenComposeAsync

3.5 任务事件类型

CompletionStage接口也支持类似我们常用的try-catch-finally中的finally的作用,无论这个任务的执行结果是正常还是出现异常的情况,都必须要去执行的一个代码块。在CompletionStage接口提供了以下两种接口回调的形式(whenComplete、handle),并支持主线程同步执行同时也支持使用默认线程池,或者使用自定义线程池去异步执行最终的回调处理。例如我们一个事务操作,无论这段代码执行是否成功,我们都必须要去关闭事务。

任务完成事件(whenComplete):结果无返回值,若出现异常执行完whenComplete回调处理完成后将中断主线程的运行。

// 1.whenComplete回调函数中Throwable对象不对空代表出现异常,为空则表示无异常

public CompletionStage whenComplete(BiConsumer super T, ? super Throwable> action);

public CompletionStage whenCompleteAsync(BiConsumer super T, ? super Throwable> action);

public CompletionStage whenCompleteAsync(BiConsumer super T, ? super Throwable> action, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture whenCompleteFufute = CompletableFuture.supplyAsync(() -> {

int a = 0;

int b = 100 / a;

return "this is supplyAsync normal";

}).whenCompleteAsync((r, th) -> {

if (th != null) {

log.error("this is whenCompleteAsync error");

}

else {

log.info("this is whenCompleteAsync success");

}

});

log.info(whenCompleteFufute.join()); // 输出结果:this is whenCompleteAsync error

任务完成回调事件(handle):结果有返回值,若出现异常执行完handle回调处理完成后将继续执行主线程的后续操作,不中断主线程运行。

// 2.handle回调函数中Throwable对象不对空代表出现异常,为空则表示无异常

public CompletionStage handle(BiFunction super T, Throwable, ? extends U> fn);

public CompletionStage handleAsync(BiFunction super T, Throwable, ? extends U> fn);

public CompletionStage handleAsync(BiFunction super T, Throwable, ? extends U> fn, Executor executor);

// ====================================demo华丽分割线============================================

CompletableFuture whenCompleteFufute = CompletableFuture.supplyAsync(() -> {

int a = 0;

int b = 100 / a;

return "this is supplyAsync normal";

}).handleAsync((r, th) -> {

if (th != null) {

return "this is handleAsync error";

}

else {

return "this is handleAsync success";

}

});

log.info(whenCompleteFufute.join());

// 输出结果:this is handleAsync error

log.info("main thread is running");

// 输出结果:main thread is running

4.CompletionStage异常处理方法

exceptionally:只要是个程序,就会有异常出现的情况,例如一个CompletionStage任务,如果执行过程中出现异常,我们为了保证异常情况下程序能够正常处理业务逻辑,那么在这里我们就可以使用exceptionally进行异常回调处理。当CompletionStage任务出现异常时就会触发回调exceptionally,否则CompletionStage任务正常执行业务不进行异常回调处理。

public CompletionStage exceptionally(Function fn);

// ====================================demo华丽分割线============================================

CompletableFuture exceptionallyFuture = CompletableFuture.supplyAsync(() -> {

int a = 0;

int b = 10 / a; // 除数为0将抛异常

return "this is supplyAsync normal";

}).exceptionally(th -> {

log.error("exception:{}", th.getMessage());

return "this is exceptionally";

});

log.info(exceptionallyFuture.join()); // 输出结果:this is exceptionally

注:以下这两种情况可能大家在实际开发过程中会比较少见,但还是得在这里做个提醒,以免到最后准备不充分出现设计上的缺陷。

当whenCompleteAsync与exceptionally同时使用时,若出现异常情况,由于exceptionally有返回值,所以优先执行whenCompleteAsync,后执行exceptionally。当handleAsync与exceptionally同时出现时,由于handleAsync已经包含了exceptionally的所有操作,即handleAsync回调有返回值,且有Throwable异常对象能够进行异常处理,所以这两者同时出现时exceptionally将失效。

5.方法类型总结

根据以上的方法我们可以总结出这些任务其实就分为三大类,相当于通过CompletionStage的回调机制来实现多个任务串、多个任务并行、多个任务聚合的操作,因此CompletableFuture对于异步任务的处理提供了更加强大的编程模型。所以说java8提供的CompletableFuture类相对原来的Future接口来说提供了一些链式的编程,省去了我们很多异步任务回调操作复杂的步骤,让我们这些码农们能够有更高的效率输出产品。


版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。

上一篇:Spring中的@Transactional的工作原理(spring actuator原理)
下一篇:SpringBoot 项目使用hutool 工具进行 http 接口调用的处理方法(springboot项目实战)
相关文章

 发表评论

暂时没有评论,来抢沙发吧~