Java多线程工具CompletableFuture的使用教程
前言
Future的问题
写多线程程序的时候,可以使用Future从一个异步线程中拿到结果,但是如果使用过程中会发现一些问题:
- 如果想要对Future的结果做进一步的操作,需要阻塞当前线程
- 多个Future不能被链式的执行,每个Future的结果都是独立的,期望对一个Future的结果做另外一件异步的事情;
- 没有异常处理策略,如果Future执行失败了,需要手动捕捉
CompletableFuture应运而生
为了解决Future问题,JDK在1.8的时候给我们提供了一个好用的工具类CompletableFuture;
它实现了Future和CompletionStage接口,针对Future的不足之处给出了相应的处理方式。
- 在异步线程执行结束后可以自动回调我们新的处理逻辑,无需阻塞
- 可以对多个异步任务进行编排,组合或者排序
- 异常处理
CompletableFuture的核心思想是将每个异步任务都可以看做一个步骤(CompletionStage),然后其他的异步任务可以根据这个步骤做一些想做的事情。
CompletionStage定义了许多步骤处理的方法,功能非常强大,这里就只列一下日常中常用到的一些方法供大家参考。
使用方式
基本使用-提交异步任务
简单的使用方式
异步执行,无需结果:
// 可以执行Executors异步执行,如果不指定,默认使用ForkJoinPool CompletableFuture.runAsync(() -> System.out.println("Hello CompletableFuture!"));
异步执行,同时返回结果:
// 同样可以指定线程池 CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> "Hello CompletableFuture!"); System.out.println(stringCompletableFuture.get());
处理上个异步任务结果
thenRun: 不需要上一步的结果,直接直接新的操作
thenAccept:获取上一步异步处理的内容,进行新的操作
thenApply: 获取上一步的内容,然后产生新的内容
所有加上Async后缀的,代表新的处理操作仍然是异步的。Async的操作都可以指定Executors进行处理
// Demo CompletableFuture .supplyAsync(() -> "Hello CompletableFuture!") // 针对上一步的结果做处理,产生新的结果 .thenApplyAsync(s -> s.toUpperCase()) // 针对上一步的结果做处理,不返回结果 .thenAcceptAsync(s -> System.out.println(s)) // 不需要上一步返回的结果,直接进行操作 .thenRunAsync(() -> System.out.println("end")); ;
对两个结果进行选用-acceptEither
当我们有两个回调在处理的时候,任何完成都可以使用,两者结果没有关系,那么使用acceptEither。
两个异步线程谁先执行完成,用谁的结果,其余类型的方法也是如此。
// 返回abc CompletableFuture .supplyAsync(() -> { SleepUtils.sleep(100); return "Hello CompletableFuture!"; }) .acceptEither(CompletableFuture.supplyAsync(() -> "abc"), new Consumer<String>() { @Override public void accept(String s) { System.out.println(s); } }); // 返回Hello CompletableFuture! CompletableFuture .supplyAsync(() -> "Hello CompletableFuture!") .acceptEither(CompletableFuture.supplyAsync(() -> { SleepUtils.sleep(100); return "abc"; }), new Consumer<String>() { @Override public void accept(String s) { System.out.println(s); } });
对两个结果进行合并-thenCombine, thenAcceptBoth
thenCombine
当我们有两个CompletionStage时,需要对两个的结果进行整合处理,然后计算得出一个新的结果。
- thenCompose是对上一个CompletionStage的结果进行处理,返回结果,并且返回类型必须是CompletionStage。
- thenCombine是得到第一个CompletionStage的结果,然后拿到当前的CompletionStage,两者的结果进行处理。
CompletableFuture<Integer> heightAsync = CompletableFuture.supplyAsync(() -> 172); CompletableFuture<Double> weightAsync = CompletableFuture.supplyAsync(() -> 65) .thenCombine(heightAsync, new BiFunction<Integer, Integer, Double>() { @Override public Double apply(Integer wight, Integer height) { return wight * 10000.0 / (height * height); } }) ;
thenAcceptBoth
需要两个异步CompletableFuture的结果,两者都完成的时候,才进入thenAcceptBoth回调。
// thenAcceptBoth案例: CompletableFuture .supplyAsync(() -> "Hello CompletableFuture!") .thenAcceptBoth(CompletableFuture.supplyAsync(() -> "abc"), new BiConsumer<String, String>() { // 参数一为我们刚开始运行时的CompletableStage,新传入的作为第二个参数 @Override public void accept(String s, String s2) { System.out.println("param1=" + s + ", param2=" + s2); } }); // 结果:param1=Hello CompletableFuture!, param2=abc
异常处理
当我们使用CompleteFuture进行链式调用的时候,多个异步回调中,如果有一个执行出现问题,那么接下来的回调都会停止,所以需要一种异常处理策略。
exceptionally
exceptionally是当出现错误时,给我们机会进行恢复,自定义返回内容。
CompletableFuture.supplyAsync(() -> { throw new RuntimeException("发生错误"); }).exceptionally(throwable -> { log.error("调用错误 {}", throwable.getMessage(), throwable); return "异常处理内容"; });
handle
exceptionally是只有发生异常时才会执行,而handle则是不管是否发生错误都会执行。
CompletableFuture.supplyAsync(() -> { return "abc"; }) .handle((r,err) -> { log.error("调用错误 {}", err.getMessage(), err); // 对结果做额外的处理 return r; }) ;
案例
大量用户发送短信|消息
需求为对某个表中特定条件的用户进行短信通知,但是短信用户有成百上千万,如果使用单线程读取效率会很慢。这个时候可以考虑使用多线程的方式进行读取;
1、将读取任务拆分为多个不同的子任务,指定读取的偏移量和个数
// 假设有500万条记录 long recordCount = 500 * 10000; int subTaskRecordCount = 10000; // 对记录进行分片 List<Map> subTaskList = new LinkedList<>(); for (int i = 0; i < recordCount / 500; i++) { // 如果子任务结构复杂,建议使用对象 HashMap<String, Integer> subTask = new HashMap<>(); subTask.put("index", i); subTask.put("offset", i * subTaskRecordCount); subTask.put("count", subTaskRecordCount); subTaskList.add(subTask); }
2、使用多线程进行批量读取
// 进行subTask批量处理,拆分为不同的任务 subTaskList.stream() .map(subTask -> CompletableFuture.runAsync(()->{ // 读取数据,然后处理 // dataTunel.read(subTask); },excuturs)) // 使用应用的通用任务线程池 .map(c -> ((CompletableFuture<?>) c).join());
3、进行业务逻辑处理,或者直接在读取完进行业务逻辑处理也是可以;
并发获取商品不同信息
在系统拆分比较细的时候,价格,优惠券,库存,商品详情等信息分散在不同的系统中,有时候需要同时获取商品的所有信息, 有时候可能只需要获取商品的部分信息。
当然问题点在于要调用多个不同的系统,需要将RT降低下来,那么需要进行并发调用;
List<Task> taskList = new ArrayList<>(); List<Object> result = taskList.stream() .map(task -> CompletableFuture.supplyAsync(()->{ // handlerMap.get(task).query(); return ""; }, executorService)) .map(c -> c.join()) .collect(Collectors.toList());
问题
thenRun和thenRunAsync有什么区别
- 如果不使用传入的线程池,大家用默认的线程池ForkJoinPool
- thenRun用的默认和上一个任务使用相同的线程池
- thenRunAsync在执行新的任务的时候可以接受传入一个新的线程池,使用新的线程池执行任务;
handle和exceptional有什么区别
exceptionally是只有发生异常时才会执行,而handle则是不管是否发生错误都会执行。
最后
一般情况下上述简单的API已经满足绝大部分的场景了,如果有更复杂的诉求,可继续深入研究。
到此这篇关于Java多线程工具CompletableFuture的使用教程的文章就介绍到这了,更多相关Java CompletableFuture内容请搜索猪先飞以前的文章或继续浏览下面的相关文章希望大家以后多多支持猪先飞!
原文出处:https://juejin.cn/post/7133394175059787790
相关文章
- 这篇文章主要介绍了如何利用java语言实现经典《复杂迷宫》游戏,文中采用了swing技术进行了界面化处理,感兴趣的小伙伴可以动手试一试...2022-02-01
java 运行报错has been compiled by a more recent version of the Java Runtime
java 运行报错has been compiled by a more recent version of the Java Runtime (class file version 54.0)...2021-04-01- 这篇文章主要介绍了在java中获取List集合中最大的日期时间操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-15
- 这篇文章主要介绍了教你怎么用Java获取国家法定节假日,文中有非常详细的代码示例,对正在学习java的小伙伴们有非常好的帮助,需要的朋友可以参考下...2021-04-23
- 这篇文章主要介绍了Java如何发起http请求的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-03-31
- 说起C#和Java这两门语言(语法,数据类型 等),个人以为,大概有90%以上的相似,甚至可以认为几乎一样。但是在工作中,我也发现了一些细微的差别...2020-06-25
- 这篇文章主要介绍了解决Java处理HTTP请求超时的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-03-29
- 这篇文章主要介绍了java 判断两个时间段是否重叠的案例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-15
- 这篇文章主要介绍了超简洁java实现双色球若干注随机号码生成(实例代码),本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-04-02
- 这篇文章主要介绍了Java生成随机姓名、性别和年龄的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-10-01
java 画pdf用itext调整表格宽度、自定义各个列宽的方法
这篇文章主要介绍了java 画pdf用itext调整表格宽度、自定义各个列宽的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-01-31- 这篇文章主要介绍了java正则表达式判断前端参数修改表中另一个字段的值,需要的朋友可以参考下...2021-05-07
Java使用ScriptEngine动态执行代码(附Java几种动态执行代码比较)
这篇文章主要介绍了Java使用ScriptEngine动态执行代码,并且分享Java几种动态执行代码比较,需要的朋友可以参考下...2021-04-15- 这篇文章主要介绍了Java开发实现人机猜拳游戏,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2020-08-03
- 这篇文章主要介绍了Java List集合返回值去掉中括号('[ ]')的操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-29
Java中lombok的@Builder注解的解析与简单使用详解
这篇文章主要介绍了Java中lombok的@Builder注解的解析与简单使用,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-01-06- 下面小编就为大家带来一篇java中String类型变量的赋值问题介绍。小编觉得挺不错的。现在分享给大家,给大家一个参考。...2016-03-28
Java 8 Stream 的终极技巧——Collectors 功能与操作方法详解
这篇文章主要介绍了Java 8 Stream Collectors 功能与操作方法,结合实例形式详细分析了Java 8 Stream Collectors 功能、操作方法及相关注意事项,需要的朋友可以参考下...2020-05-20- 这篇文章主要介绍了Java线程池中的各个参数如何合理设置操作,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-06-19
- 在Java中,我们可以利用多线程来最大化地压榨CPU多核计算的能力,下面这篇文章主要给大家介绍了关于java中多线程与线程池基本使用的相关资料,需要的朋友可以参考下...2021-09-13