在Java中异步编程,不肯定非要应用rxJava, Java自身的库中的CompletableFuture能够很好的应答大部分的场景。
这篇文章介绍 Java 8 的 CompletionStage API 和它的规范库的实现 CompletableFuture。API通过例子的形式演示了它的行为,每个例子演示一到两个行为。
既然CompletableFuture
类实现了CompletionStage
接口,首先咱们须要了解这个接口的契约。它代表了一个特定的计算的阶段,能够同步或者异步的被实现。你能够把它看成一个计算流水线上的一个单元,最终会产生一个最终后果,这意味着几个CompletionStage
能够串联起来,一个实现的阶段能够触发下一阶段的执行,接着触发下一次,接着……
除了实现CompletionStage
接口, CompletableFuture
也实现了future
接口, 代表一个未实现的异步事件。CompletableFuture
提供了办法,可能显式地实现这个future,所以它叫CompletableFuture
。
1、 创立一个实现的CompletableFuture
最简略的例子就是应用一个预约义的后果创立一个实现的CompletableFuture,通常咱们会在计算的开始阶段应用它。
static void completedFutureExample() { CompletableFuture cf = CompletableFuture.completedFuture("message"); assertTrue(cf.isDone()); assertEquals("message", cf.getNow(null));}
getNow(null)
办法在future实现的状况下会返回后果,就比方下面这个例子,否则返回null (传入的参数)。
2、运行一个简略的异步阶段
这个例子创立一个一个异步执行的阶段:
static void runAsyncExample() { CompletableFuture cf = CompletableFuture.runAsync(() -> { assertTrue(Thread.currentThread().isDaemon()); randomSleep(); }); assertFalse(cf.isDone()); sleepEnough(); assertTrue(cf.isDone());}
通过这个例子能够学到两件事件:
CompletableFuture的办法如果以Async
结尾,它会异步的执行(没有指定executor的状况下), 异步执行通过ForkJoinPool实现, 它应用守护线程去执行工作。留神这是CompletableFuture的个性, 其它CompletionStage能够override这个默认的行为。
3、在前一个阶段上利用函数
上面这个例子应用后面 #1 的实现的CompletableFuture, #1返回后果为字符串message
,而后利用一个函数把它变成大写字母。
static void thenApplyExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> { assertFalse(Thread.currentThread().isDaemon()); return s.toUpperCase(); }); assertEquals("MESSAGE", cf.getNow(null));}
留神thenApply
办法名称代表的行为。
then
意味着这个阶段的动作产生以后的阶段失常实现之后。本例中,以后节点实现,返回字符串message
。
Apply
意味着返回的阶段将会对后果前一阶段的后果利用一个函数。
函数的执行会被阻塞,这意味着getNow()
只有打斜操作被实现后才返回。
4、在前一个阶段上异步利用函数
通过调用异步办法(办法后边加Async后缀),串联起来的CompletableFuture能够异步地执行(应用ForkJoinPool.commonPool())。
static void thenApplyAsyncExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> { assertTrue(Thread.currentThread().isDaemon()); randomSleep(); return s.toUpperCase(); }); assertNull(cf.getNow(null)); assertEquals("MESSAGE", cf.join());}
5、应用定制的Executor在前一个阶段上异步利用函数
异步办法一个十分有用的个性就是可能提供一个Executor来异步地执行CompletableFuture。这个例子演示了如何应用一个固定大小的线程池来利用大写函数。
static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() { int count = 1; @Override public Thread newThread(Runnable runnable) { return new Thread(runnable, "custom-executor-" + count++); }}); static void thenApplyAsyncWithExecutorExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> { assertTrue(Thread.currentThread().getName().startsWith("custom-executor-")); assertFalse(Thread.currentThread().isDaemon()); randomSleep(); return s.toUpperCase(); }, executor); assertNull(cf.getNow(null)); assertEquals("MESSAGE", cf.join());}
6、生产前一阶段的后果
如果下一阶段接管了以后阶段的后果,然而在计算的时候不须要返回值(它的返回类型是void), 那么它能够不利用一个函数,而是一个消费者, 调用办法也变成了thenAccept
:
static void thenAcceptExample() { StringBuilder result = new StringBuilder(); CompletableFuture.completedFuture("thenAccept message") .thenAccept(s -> result.append(s)); assertTrue("Result was empty", result.length() > 0);}
本例中消费者同步地执行,所以咱们不须要在CompletableFuture调用join
办法。
7、异步地生产迁徙阶段的后果
同样,能够应用thenAcceptAsync
办法, 串联的CompletableFuture能够异步地执行。
static void thenAcceptAsyncExample() { StringBuilder result = new StringBuilder(); CompletableFuture cf = CompletableFuture.completedFuture("thenAcceptAsync message") .thenAcceptAsync(s -> result.append(s)); cf.join(); assertTrue("Result was empty", result.length() > 0);}
8、实现计算异样
当初咱们来看一下异步操作如何显式地返回异样,用来批示计算失败。咱们简化这个例子,操作解决一个字符串,把它转换成答谢,咱们模仿提早一秒。
咱们应用thenApplyAsync(Function, Executor)
办法,第一个参数传入大写函数, executor是一个delayed executor,在执行前会提早一秒。
static void completeExceptionallyExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)); CompletableFuture exceptionHandler = cf.handle((s, th) -> { return (th != null) ? "message upon cancel" : ""; }); cf.completeExceptionally(new RuntimeException("completed exceptionally"));assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally()); try { cf.join(); fail("Should have thrown an exception"); } catch(CompletionException ex) { // just for testing assertEquals("completed exceptionally", ex.getCause().getMessage()); } assertEquals("message upon cancel", exceptionHandler.join());}
让咱们看一下细节。
首先咱们创立了一个CompletableFuture, 实现后返回一个字符串message
,接着咱们调用thenApplyAsync
办法,它返回一个CompletableFuture。这个办法在第一个函数实现后,异步地利用转大写字母函数。
这个例子还演示了如何通过delayedExecutor(timeout, timeUnit)
提早执行一个异步工作。
咱们创立了一个拆散的handler
阶段: exceptionHandler, 它解决异样异样,在异常情况下返回message upon cancel
。
下一步咱们显式地用异样实现第二个阶段。 在阶段上调用join
办法,它会执行大写转换,而后抛出CompletionException(失常的join会期待1秒,而后失去大写的字符串。不过咱们的例子还没等它执行就实现了异样), 而后它触发了handler阶段。
9、勾销计算
和实现异样相似,咱们能够调用cancel(boolean mayInterruptIfRunning)
勾销计算。对于CompletableFuture类,布尔参数并没有被应用,这是因为它并没有应用中断去勾销操作,相同,cancel
等价于completeExceptionally(new CancellationException())
。
static void cancelExample() { CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)); CompletableFuture cf2 = cf.exceptionally(throwable -> "canceled message"); assertTrue("Was not canceled", cf.cancel(true)); assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally()); assertEquals("canceled message", cf2.join());}
10、在两个实现的阶段其中之一上利用函数
上面的例子创立了CompletableFuture
, applyToEither
解决两个阶段, 在其中之一上利用函数(包保障哪一个被执行)。 本例中的两个阶段一个是利用大写转换在原始的字符串上, 另一个阶段是利用小些转换。
static void applyToEitherExample() { String original = "Message"; CompletableFuture cf1 = CompletableFuture.completedFuture(original) .thenApplyAsync(s -> delayedUpperCase(s)); CompletableFuture cf2 = cf1.applyToEither( CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)), s -> s + " from applyToEither"); assertTrue(cf2.join().endsWith(" from applyToEither"));}
11、在两个实现的阶段其中之一上调用生产函数
和前一个例子很相似了,只不过咱们调用的是消费者函数 (Function变成Consumer):
static void acceptEitherExample() { String original = "Message"; StringBuilder result = new StringBuilder(); CompletableFuture cf = CompletableFuture.completedFuture(original) .thenApplyAsync(s -> delayedUpperCase(s)) .acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)), s -> result.append(s).append("acceptEither")); cf.join(); assertTrue("Result was empty", result.toString().endsWith("acceptEither"));}
12、在两个阶段都执行完后运行一个 Runnable
这个例子演示了依赖的CompletableFuture如果期待两个阶段实现后执行了一个Runnable。 留神上面所有的阶段都是同步执行的,第一个阶段执行大写转换,第二个阶段执行小写转换。
static void runAfterBothExample() { String original = "Message"; StringBuilder result = new StringBuilder(); CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth( CompletableFuture.completedFuture(original).thenApply(String::toLowerCase), () -> result.append("done")); assertTrue("Result was empty", result.length() > 0);}
13、 应用BiConsumer解决两个阶段的后果
下面的例子还能够通过BiConsumer来实现:
static void thenAcceptBothExample() { String original = "Message"; StringBuilder result = new StringBuilder(); CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth( CompletableFuture.completedFuture(original).thenApply(String::toLowerCase), (s1, s2) -> result.append(s1 + s2)); assertEquals("MESSAGEmessage", result.toString());}
14、应用BiFunction解决两个阶段的后果
如果CompletableFuture依赖两个后面阶段的后果, 它复合两个阶段的后果再返回一个后果,咱们就能够应用thenCombine()
函数。整个流水线是同步的,所以getNow()
会失去最终的后果,它把大写和小写字符串连接起来。
static void thenCombineExample() { String original = "Message"; CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s)) .thenCombine(CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)), (s1, s2) -> s1 + s2); assertEquals("MESSAGEmessage", cf.getNow(null));}
15、异步应用BiFunction解决两个阶段的后果
相似下面的例子,然而有一点不同: 依赖的前两个阶段异步地执行,所以thenCombine()
也异步地执行,即时它没有Async
后缀。
Javadoc中有正文:
Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method
所以咱们须要join
办法期待后果的实现。
static void thenCombineAsyncExample() { String original = "Message"; CompletableFuture cf = CompletableFuture.completedFuture(original) .thenApplyAsync(s -> delayedUpperCase(s)) .thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)), (s1, s2) -> s1 + s2); assertEquals("MESSAGEmessage", cf.join());}
16、组合 CompletableFuture
咱们能够应用thenCompose()
实现下面两个例子。这个办法期待第一个阶段的实现(大写转换), 它的后果传给一个指定的返回CompletableFuture函数,它的后果就是返回的CompletableFuture的后果。
有点拗口,然而咱们看例子来了解。函数须要一个大写字符串做参数,而后返回一个CompletableFuture, 这个CompletableFuture会转换字符串变成小写而后连贯在大写字符串的前面。
static void thenComposeExample() { String original = "Message"; CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s)) .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)) .thenApply(s -> upper + s)); assertEquals("MESSAGEmessage", cf.join());}
17、当几个阶段中的一个实现,创立一个实现的阶段
上面的例子演示了当任意一个CompletableFuture实现后, 创立一个实现的CompletableFuture.
待处理的阶段首先创立, 每个阶段都是转换一个字符串为大写。因为本例中这些阶段都是同步地执行(thenApply), 从anyOf
中创立的CompletableFuture会立刻实现,这样所有的阶段都已实现,咱们应用whenComplete(BiConsumer<? super Object, ? super Throwable> action)
解决实现的后果。
static void anyOfExample() { StringBuilder result = new StringBuilder(); List messages = Arrays.asList("a", "b", "c"); List<CompletableFuture> futures = messages.stream() .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s))) .collect(Collectors.toList()); CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((res, th) -> { if(th == null) { assertTrue(isUpperCase((String) res)); result.append(res); } }); assertTrue("Result was empty", result.length() > 0);}
18、当所有的阶段都实现后创立一个阶段
上一个例子是当任意一个阶段实现后接着解决,接下来的两个例子演示当所有的阶段实现后才持续解决, 同步地形式和异步地形式两种。
static void allOfExample() { StringBuilder result = new StringBuilder(); List messages = Arrays.asList("a", "b", "c"); List<CompletableFuture> futures = messages.stream() .map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s))) .collect(Collectors.toList()); CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((v, th) -> { futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null)))); result.append("done"); }); assertTrue("Result was empty", result.length() > 0);}
19、当所有的阶段都实现后异步地创立一个阶段
应用thenApplyAsync()
替换那些单个的CompletableFutures的办法,allOf()
会在通用池中的线程中异步地执行。所以咱们须要调用join
办法期待它实现。
static void allOfAsyncExample() { StringBuilder result = new StringBuilder(); List messages = Arrays.asList("a", "b", "c"); List<CompletableFuture> futures = messages.stream() .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> delayedUpperCase(s))) .collect(Collectors.toList()); CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])) .whenComplete((v, th) -> { futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null)))); result.append("done"); }); allOf.join(); assertTrue("Result was empty", result.length() > 0);}
20、实在的例子
当初你曾经理解了CompletionStage 和 CompletableFuture 的一些函数的性能,上面的例子是一个实际场景:
- 首先异步调用
cars
办法取得Car的列表,它返回CompletionStage场景。cars
生产一个近程的REST API。 - 而后咱们复合一个CompletionStage填写每个汽车的评分,通过
rating(manufacturerId)
返回一个CompletionStage, 它会异步地获取汽车的评分(可能又是一个REST API调用) - 当所有的汽车填好评分后,咱们完结这个列表,所以咱们调用
allOf
失去最终的阶段, 它在后面阶段所有阶段实现后才实现。 - 在最终的阶段调用
whenComplete()
,咱们打印出每个汽车和它的评分。
cars().thenCompose(cars -> { List<CompletionStage> updatedCars = cars.stream() .map(car -> rating(car.manufacturerId).thenApply(r -> { car.setRating(r); return car; })).collect(Collectors.toList()); CompletableFuture done = CompletableFuture .allOf(updatedCars.toArray(new CompletableFuture[updatedCars.size()])); return done.thenApply(v -> updatedCars.stream().map(CompletionStage::toCompletableFuture) .map(CompletableFuture::join).collect(Collectors.toList()));}).whenComplete((cars, th) -> { if (th == null) { cars.forEach(System.out::println); } else { throw new RuntimeException(th); }}).toCompletableFuture().join();
因为每个汽车的实例都是独立的,失去每个汽车的评分都能够异步地执行,这会进步零碎的性能(提早),而且,期待所有的汽车评分被解决应用的是allOf
办法,而不是手工的线程期待(Thread#join() 或 a CountDownLatch)。
最初,关注公众号Java技术栈,在后盾回复:面试,能够获取我整顿的 Java 并发多线程系列面试题和答案,十分齐全。
原文:https://mahmoudanouti.wordpre...\
起源:https://colobu.com/2018/03/12...
参考文档:
https://medium.com/@johnmccle...\
http://javaday.org.ua/complet...\
https://blog.takipi.com/back-...\
https://github.com/spotify/co...
近期热文举荐:
1.600+ 道 Java面试题及答案整顿(2021最新版)
2.终于靠开源我的项目弄到 IntelliJ IDEA 激活码了,真香!
3.阿里 Mock 工具正式开源,干掉市面上所有 Mock 工具!
4.Spring Cloud 2020.0.0 正式公布,全新颠覆性版本!
5.《Java开发手册(嵩山版)》最新公布,速速下载!
感觉不错,别忘了顺手点赞+转发哦!