技术干货每日送达!
你知道的越多,不知道的就越多,业余的像一棵小草!
你来,我们一起精进!你不来,我和你的竞争对手一起精进!
编辑:业余草
来源:urlify.cn/ayaMBb
在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()); }