Java 8:CompletableFuture终极指南
suiw9 2024-10-23 18:48 27 浏览 0 评论
本文由 ImportNew-Grey翻译自javacodegeeks。欢迎加入Java小组。转载请参见文章末尾的要求。
Java 8来了,是时候学一下新的东西了。Java 7和Java 6只不过是稍作修改的版本,而Java 8将会发生重大的改进。或许是Java 8太大了吧?今天我会给你彻底地解释JDK 8中的新的抽象 – CompletableFuture。众所周知,Java 8不到一年就会发布,因此这篇文章是基于JDK 8 build 88 with lambda support的。CompletableFutureextendsFuture提供了方法,一元操作符和促进异步性以及事件驱动编程模型,它并不止步于旧版本的Java中。如果你打开JavaDoc of CompletableFuture你一定会感到震惊。大约有五十种方法(!),而且它们中的一些非常有意思而且不好理解,例如:
public <U,V> CompletableFuture<V> thenCombineAsync( CompletableFuture<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)
不必担心,继续读下去。CompletableFuture收集了所有ListenableFuture in Guava和SettableFuture的特征。此外,内置的lambda表达式使它更接近于Scala/Akka futures。这听起来好得令人难以置信,但是请继续读下去。CompletableFuture有两个主要的方面优于ol中的Future – 异步回调/转换,这能使得从任何时刻的任何线程都可以设置CompletableFuture的值。
提取、修改包装的值
通常futures代表其它线程中运行的代码,但事实并非总是如此。有时你想要创造一个Future来表示你知道将会发生什么,例如JMS message arrival。所以你有Future但是未来并没有潜在的异步工作。你只是想在未来JMS消息到达时简单地完成(解决),这是由一个事件驱动的。在这种情况下,你可以简单地创建CompletableFuture来返还给你的客户端,只要你认为你的结果是可用的,仅仅通过complete就能解锁所有等待Future的客户端。
首先你可以简单地创建新的CompletableFuture并且给你的客户端:
public CompletableFuture<String> ask { final CompletableFuture<String> future = new CompletableFuture<>; //... return future; }
注意这个future和Callable没有任何联系,没有线程池也不是异步工作。如果现在客户端代码调用ask.get它将永远阻塞。如果寄存器完成回调,它们就永远不会生效了。所以关键是什么?现在你可以说:
future.complete("42")
…此时此刻所有客户端Future.get将得到字符串的结果,同时完成回调以后将会立即生效。当你想代表Future的任务时是非常方便的,而且没有必要去计算一些执行线程的任务上。CompletableFuture.complete只能调用一次,后续调用将被忽略。但也有一个后门叫做CompletableFuture.obtrudeValue(…)覆盖一个新Future之前的价值,请小心使用。
有时你想要看到信号发生故障的情况,如你所知Future对象可以处理它所包含的结果或异常。如果你想进一步传递一些异常,可以用CompletableFuture.completeExceptionally(ex) (或者用obtrudeException(ex)这样更强大的方法覆盖前面的异常)。 completeExceptionally也能解锁所有等待的客户端,但这一次从get抛出异常。说到get,也有CompletableFuture.join方法在错误处理方面有着细微的变动。但总体上,它们都是一样的。最后也有CompletableFuture.getNow(valueIfAbsent)方法没有阻塞但是如果Future还没完成将返回默认值,这使得当构建那种我们不想等太久的健壮系统时非常有用。 最后static的方法是用completedFuture(value)来返回已经完成Future的对象,当测试或者写一些适配器层时可能非常有用。
创造和获取CompletableFuture
好了,那么手动地创建CompletableFuture是我们唯一的选择吗?不一定。就像一般的Futures,我们可以关联存在的任务,同时CompletableFuture使用工厂方法:
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier); static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor); static CompletableFuture<Void> runAsync(Runnable runnable); static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);
无参方法Executor
是以…Async
结尾同时将会使用ForkJoinPool.commonPool
(全局的,在JDK8中介绍的通用池),这适用于CompletableFuture
类中的大多数的方法。runAsync
易于理解,注意它需要Runnable
,因此它返回CompletableFuture<Void>
作为Runnable
不返回任何值。如果你需要处理异步操作并返回结果,使用Supplier<U>
:
final CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String> { @Override public String get { //...long running... return "42"; } }, executor);
但是别忘了,Java 8里面还有lambdas表达式呢!
finalCompletableFuture<String> future = CompletableFuture.supplyAsync( -> { //...long running... return "42"; }, executor);
或者:
final CompletableFuture<String> future = CompletableFuture.supplyAsync( -> longRunningTask(params), executor);
虽然这篇文章不是关于Lambda的,但是我会相当频繁地使用lambda表达式。
转换和作用于CompletableFuture(thenApply)
我说过CompletableFuture优于Future但是你还不知道为什么吗?简单说,因为CompletableFuture是一个原子也是一个因子。我说的这句话没什么帮助吗?Scala和JavaScript都允许future完成时允许注册异步回调,直到它准备好我们才要等待和阻止它。我们可以简单地说:运行这个函数时就出现了结果。此外,我们可以叠加这些功能,把多个future组合在一起等。例如如果我们从String转为Integer,我们可以转为在不关联的前提下从CompletableFuture到 CompletableFuture<Integer。这是通过thenApply的方法:
<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn); <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn); <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor);如前所述...Async版本提供对CompletableFuture的大多数操作,因此我将在后面的部分中跳过它们。记住,第一个方法将在future完成的相同线程中调用该方法,而剩下的两个将在不同的线程池中异步地调用它。 让我们来看看thenApply的工作流程:
<pre class="brush: java; gutter: true; first-line: 1; highlight: ; html-script: false"> CompletableFuture<String> f1 = //... CompletableFuture<Integer> f2 = f1.thenApply(Integer::parseInt); CompletableFuture<Double> f3 = f2.thenApply(r -> r * r * Math.PI);
或在一个声明中:
CompletableFuture<Double> f3 = f1.thenApply(Integer::parseInt).thenApply(r -> r * r * Math.PI);
这里,你会看到一个序列的转换,从String到Integer再到Double。但最重要的是,这些转换既不立即执行也不停止。这些转换既不立即执行也不停止。他们只是记得,当原始f1完成他们所执行的程序。如果某些转换非常耗时,你可以提供你自己的Executor来异步地运行他们。注意,此操作相当于Scala中的一元map。
运行完成的代码(thenAccept/thenRun)
CompletableFuture<Void> thenAccept(Consumer<? super T> block); CompletableFuture<Void> thenRun(Runnable action);
在future的管道里有两种典型的“最终”阶段方法。他们在你使用future的值的时候做好准备,当 thenAccept提供最终的值时,thenRun执行 Runnable,这甚至没有方法去计算值。例如:
future.thenAcceptAsync(dbl -> log.debug("Result: {}", dbl), executor); log.debug("Continuing");
…Async变量也可用两种方法,隐式和显式执行器,我不会过多强调这个方法。 thenAccept/thenRun方法并没有发生阻塞(即使没有明确的executor)。它们像一个事件侦听器/处理程序,你连接到一个future时,这将执行一段时间。”Continuing”消息将立即出现,尽管future甚至没有完成。
单个CompletableFuture的错误处理
到目前为止,我们只讨论计算的结果。那么异常呢?我们可以异步地处理它们吗?当然!
CompletableFuture<String> safe = future.exceptionally(ex -> "We have a problem: " + ex.getMessage);
exceptionally接受一个函数时,将调用原始future来抛出一个异常。我们会有机会将此异常转换为和Future类型的兼容的一些值来进行恢复。safe进一步的转换将不再产生一个异常而是从提供功能的函数返回一个String值。 一个更加灵活的方法是handle接受一个函数,它接收正确的结果或异常:
CompletableFuture<Integer> safe = future.handle((ok, ex) -> { if (ok != null) { return Integer.parseInt(ok); } else { log.warn("Problem", ex); return -1; } });
handle总是被调用,结果和异常都非空,这是个一站式全方位的策略。
一起结合两个CompletableFuture
异步处理过程之一的CompletableFuture非常不错但是当多个这样的futures以各种方式组合在一起时确实显示了它的强大。
结合(链接)这两个futures(thenCompose())
有时你想运行一些future的值(当它准备好了),但这个函数也返回了future。CompletableFuture足够灵活地明白我们的函数结果现在应该作为顶级的future,对比CompletableFuture<CompletableFuture>。方法 thenCompose相当于Scala的flatMap:
<U> CompletableFuture<U> thenCompose(Function<? super T,CompletableFuture<U>> fn);
…Async变化也是可用的,在下面的事例中,仔细观察thenApply(map)和thenCompose(flatMap)的类型和差异,当应用calculateRelevance方法返回CompletableFuture:
CompletableFuture<Document> docFuture = //... CompletableFuture<CompletableFuture<Double>> f = docFuture.thenApply(this::calculateRelevance); CompletableFuture<Double> relevanceFuture = docFuture.thenCompose(this::calculateRelevance); //... private CompletableFuture<Double> calculateRelevance(Document doc) //...
thenCompose是一个重要的方法允许构建健壮的和异步的管道,没有阻塞和等待的中间步骤。
两个futures的转换值(thenCombine)
当thenCompose用于链接一个future时依赖另一个thenCombine,当他们都完成之后就结合两个独立的futures:
<U,V> CompletableFuture<V> thenCombine(CompletableFuture<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
…Async变量也是可用的,假设你有两个CompletableFuture,一个加载Customer另一个加载最近的Shop。他们彼此完全独立,但是当他们完成时,您想要使用它们的值来计算Route。这是一个可剥夺的例子:
CompletableFuture<Customer> customerFuture = loadCustomerDetails(123); CompletableFuture<Shop> shopFuture = closestShop; CompletableFuture<Route> routeFuture = customerFuture.thenCombine(shopFuture, (cust, shop) -> findRoute(cust, shop)); //... private Route findRoute(Customer customer, Shop shop) //...
请注意,在Java 8中可以用(cust, shop) -> findRoute(cust, shop)
简单地代替this::findRoute方法的引用:
customerFuture.thenCombine(shopFuture, this::findRoute);
你也知道,我们有customerFuture 和 shopFuture。那么routeFuture包装它们然后“等待”它们完成。当他们准备好了,它会运行我们提供的函数来结合所有的结果(findRoute)。当两个基本的futures完成并且 findRoute也完成时,这样routeFuture将会完成。
等待所有的 CompletableFutures 完成
如果不是产生新的CompletableFuture连接这两个结果,我们只是希望当完成时得到通知,我们可以使用thenAcceptBoth/runAfterBoth系列的方法,(…Async 变量也是可用的)。它们的工作方式与thenAccept 和 thenRun类似,但是是等待两个futures而不是一个:
<U> CompletableFuture<Void> thenAcceptBoth(CompletableFuture<? extends U> other, BiConsumer<? super T,? super U> block) CompletableFuture<Void> runAfterBoth(CompletableFuture<?> other, Runnable action)
想象一下上面的例子,这不是产生新的 CompletableFuture,你只是想要立刻发送一些事件或刷新GUI。这可以很容易地实现:thenAcceptBoth:
customerFuture.thenAcceptBoth(shopFuture, (cust, shop) -> { final Route route = findRoute(cust, shop); //refresh GUI with route });
我希望我是错的,但也许有些人会问自己一个问题:为什么我不能简单地阻塞这两个futures呢? 就像:
Future<Customer> customerFuture = loadCustomerDetails(123); Future<Shop> shopFuture = closestShop; findRoute(customerFuture.get, shopFuture.get);
好了,你当然可以这么做。但是最关键的一点是CompletableFuture是允许异步的,它是事件驱动的编程模型而不是阻塞并急切地等待着结果。所以在功能上,上面两部分代码是等价的,但后者没有必要占用一个线程来执行。
等待第一个 CompletableFuture 来完成任务
另一个有趣的事是CompletableFutureAPI可以等待第一个(与所有相反)完成的future。当你有两个相同类型任务的结果时就显得非常方便,你只要关心响应时间就行了,没有哪个任务是优先的。API方法(…Async变量也是可用的):
CompletableFuture<Void> acceptEither(CompletableFuture<? extends T> other, Consumer<? super T> block) CompletableFuture<Void> runAfterEither(CompletableFuture<?> other, Runnable action)
作为一个例子,你有两个系统可以集成。一个具有较小的平均响应时间但是拥有高的标准差,另一个一般情况下较慢,但是更加容易预测。为了两全其美(性能和可预测性)你可以在同一时间调用两个系统并等着谁先完成。通常这会是第一个系统,但是在进度变得缓慢时,第二个系统就可以在可接受的时间内完成:
CompletableFuture<String> fast = fetchFast; CompletableFuture<String> predictable = fetchPredictably; fast.acceptEither(predictable, s -> { System.out.println("Result: " + s); });
s代表了从fetchFast或是fetchPredictably得到的String。我们不必知道也无需关心。
完整地转换第一个系统
applyToEither算是 acceptEither的前辈了。当两个futures快要完成时,后者只是简单地调用一些代码片段,applyToEither将会返回一个新的future。当这两个最初的futures完成时,新的future也会完成。API有点类似于(…Async 变量也是可用的):
<U> CompletableFuture<U> applyToEither(CompletableFuture<? extends T> other, Function<? super T,U> fn)
这个额外的fn功能在第一个future被调用时能完成。我不确定这个专业化方法的目的是什么,毕竟一个人可以简单地使用:fast.applyToEither(predictable).thenApply(fn)。因为我们坚持用这个API,但我们的确不需要额外功能的应用程序,我会简单地使用Function.identity占位符:
CompletableFuture<String> fast = fetchFast; CompletableFuture<String> predictable = fetchPredictably; CompletableFuture<String> firstDone = fast.applyToEither(predictable, Function.<String>identity);
第一个完成的future可以通过运行。请注意,从客户的角度来看,两个futures实际上是在firstDone的后面而隐藏的。客户端只是等待着future来完成并且通过applyToEither使得当最先的两个任务完成时通知客户端。
多种结合的CompletableFuture
我们现在知道如何等待两个future来完成(使用thenCombine())并第一个完成(applyToEither)。但它可以扩展到任意数量的futures吗?的确,使用static辅助方法:
static CompletableFuture<Void< allOf(CompletableFuture<?<... cfs) static CompletableFuture<Object< anyOf(CompletableFuture<?<... cfs)
allOf当所有的潜在futures完成时,使用了一个futures数组并且返回一个future(等待所有的障碍)。另一方面anyOf将会等待最快的潜在futures,请看一下返回futures的一般类型,这不是你所期望的吗?我们会在接下来的文章中关注一下这个问题。
总结
我们探索了整个CompletableFuture API。我确信这样就能战无不胜了,所以在下一篇文章中我们将研究另一个简单的web爬虫程序的实现,使用CompletableFuture方法和Java 8 lambda表达式,我们也会看看CompletableFuture的缺点和不足。
[ 转载请保留原文出处、译者和译文链接。]
相关文章
相关推荐
- 你要如何学习写一个数据库内核(如何实现一个最简单的数据库)
-
数据库这个方向上还有许多细分方向,每个细分方向上都有许多知识。...
- 每个大数据架构师都需要的6个基本技能
-
为了成为一名出色的大数据架构师,首先必须成为一名数据架构师,但这两种角色的职责各有不同。数据分为结构化和非结构化两种。尽管大数据为各种规模的组织提供了许多洞察和分析的机会,但处理起来非常困难,并且需...
- 警惕!Spring Data MongoDB SpEL表达式注入漏洞风险通告
-
漏洞描述近日,亚信安全CERT监控到SpringDataMongoDB存在表达式注入漏洞(CVE-2022-22980),该漏洞源于SpringDataMongoDB应用程序在使用带有SpEL...
- 既然有MySQL了,为什么还要有MongoDB?
-
大家好,我是哪吒,最近项目在使用MongoDB作为图片和文档的存储数据库,为啥不直接存MySQL里,还要搭个MongoDB集群,麻不麻烦?...
- 分布式系统核心概念及实现(分布式核心原理解析)
-
一、分布式系统核心概念1.分布式系统的定义分布式系统是由多个独立的计算机(节点)通过网络连接,协同完成任务的系统。这些节点可以是物理机、虚拟机或容器。...
- nosql之mongodb(nosql数据库是国产的吗)
-
什么是MongoDB?MongoDB是由C++语言编写的,是一个基于分布式文件存储的开源数据库系统。...
- 如何治理非结构化数据?(非结构化化数据)
-
据调查,当前企业80%的数据为非结构化数据或半结构化数据,而结构化数据是他们管理的重点,非结构化数据却被忽视。然而,非结构化数据也有着它的价值。那么,如何治理非结构化数据?IDC调研显示,目前企业中8...
- Cloudera收购大数据加密初创企业Gazzang
-
Hadoop供应商Cloudera刚刚收购了专门研究下一代数据存储环境加密技术技术的初创企业Gazzang,但交易细节并未透露。这是Cloudera的第一笔重大收购。Gazzang成立于20...
- 全网最全95道MongoDB面试题1万字详细解析
-
1、mongodb是什么?MongoDB是由C++语言编写的,是一个基于分布式文件存储的开源数据库系统。在高负载的情况下,添加更多的节点,可以保证服务器性能。MongoDB旨在给WEB应...
- mongodb——视图(mongodb object)
-
MongoDB视图是一个可查询的对象,其内容由其他集合或视图上的聚合管道定义。MongoDB不会将视图内容持久化到磁盘。当客户端查询视图时,MongoDB可以要求客户端拥有查询视图的权限。MongoD...
- mongodb的优缺点及应用场景(mongodb 优点 应用场景)
-
一、MongoDB是什么1、维基百科MongoDB是一个基于分布式文件存储的数据库。由C++语言编写。旨在为WEB应用提供可扩展的高性能数据存储解决方案。...
- 5款Syslog集中系统日志常用工具对比推荐
-
一、为何要集中管理Syslog?Syslog由Linux/Unix系统及其他网络设备生成,广泛分布于整个网络。因其包含关键信息,可用于识别网络中的恶意活动,所以必须对其进行持续监控。将Sys...
- 跨平台、多数据库支持的开源数据库管理工具——DBeaver
-
简介今天给大家推荐一个开源的数据库管理工具——DBeaver。它支持多种数据库系统,包括Mysql、Oracle、PostgreSQL、SLQLite、SQLServer等。DBeaver的界面友好...
- 强烈推荐!数据库管理工具:Navicat Premium 16.3.2 (64位)
-
NavicatPremium,一款集数据迁移、数据库管理、SQL/查询编辑、智能设计、高效协作于一体的全能数据库开发工具。无论你是MySQL、MariaDB、MongoDB、SQLServer、O...
- 3 年 Java 程序员还玩不转 MongoDB,网友:失望
-
一、什么场景使用MongoDB?...
你 发表评论:
欢迎- 一周热门
-
-
Linux:Ubuntu22.04上安装python3.11,简单易上手
-
宝马阿布达比分公司推出独特M4升级套件,整套升级约在20万
-
MATLAB中图片保存的五种方法(一)(matlab中保存图片命令)
-
别再傻傻搞不清楚Workstation Player和Workstation Pro的区别了
-
Linux上使用tinyproxy快速搭建HTTP/HTTPS代理器
-
如何提取、修改、强刷A卡bios a卡刷bios工具
-
Element Plus 的 Dialog 组件实现点击遮罩层不关闭对话框
-
日本组合“岚”将于2020年12月31日停止团体活动
-
SpringCloud OpenFeign 使用 okhttp 发送 HTTP 请求与 HTTP/2 探索
-
tinymce 号称富文本编辑器世界第一,大家同意么?
-
- 最近发表
- 标签列表
-
- dialog.js (57)
- importnew (44)
- windows93网页版 (44)
- yii2框架的优缺点 (45)
- tinyeditor (45)
- qt5.5 (60)
- windowsserver2016镜像下载 (52)
- okhttputils (51)
- android-gif-drawable (53)
- 时间轴插件 (56)
- docker systemd (65)
- slider.js (47)
- android webview缓存 (46)
- pagination.js (59)
- loadjs (62)
- openssl1.0.2 (48)
- velocity模板引擎 (48)
- pcre library (47)
- zabbix微信报警脚本 (63)
- jnetpcap (49)
- pdfrenderer (43)
- fastutil (48)
- uinavigationcontroller (53)
- bitbucket.org (44)
- python websocket-client (47)