20个使用 Java CompletableFuture的示例,不服不行
共 9902字,需浏览 20分钟
·
2020-08-29 22:20
来源:colobu.com
在Java中异步编程,不一定非要使用rxJava, Java本身的库中的CompletableFuture可以很好的应对大部分的场景。
CompletableFuture
类实现了CompletionStage
接口,首先我们需要理解这个接口的契约。它代表了一个特定的计算的阶段,可以同步或者异步的被完成。你可以把它看成一个计算流水线上的一个单元,最终会产生一个最终结果,这意味着几个CompletionStage
可以串联起来,一个完成的阶段可以触发下一阶段的执行,接着触发下一次,接着……CompletionStage
接口,CompletableFuture
也实现了future
接口, 代表一个未完成的异步事件。CompletableFuture
提供了方法,能够显式地完成这个future,所以它叫CompletableFuture
。1、 创建一个完成的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());
}
Async
结尾,它会异步的执行(没有指定executor的情况下), 异步执行通过ForkJoinPool实现, 它使用守护线程去执行任务。注意这是CompletableFuture的特性, 其它CompletionStage可以override这个默认的行为。3、在前一个阶段上应用函数
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、在前一个阶段上异步应用函数
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在前一个阶段上异步应用函数
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、消费前一阶段的结果
thenAccept
:static void thenAcceptExample() {
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture("thenAccept message")
.thenAccept(s -> result.append(s));
assertTrue("Result was empty", result.length() > 0);
}
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());
}
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、在两个完成的阶段其中之一上调用消费函数
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
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处理两个阶段的结果
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处理两个阶段的结果
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
后缀。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的结果。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、当几个阶段中的一个完成,创建一个完成的阶段
anyOf
中创建的CompletableFuture会立即完成,这样所有的阶段都已完成,我们使用whenComplete(BiConsumer super Object, ? super Throwable> action)
处理完成的结果。static void anyOfExample() {
StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
Listfutures = 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");
Listfutures = 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");
Listfutures = 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、真实的例子
首先异步调用 cars
方法获得Car的列表,它返回CompletionStage场景。cars
消费一个远程的REST API。然后我们复合一个CompletionStage填写每个汽车的评分,通过 rating(manufacturerId)
返回一个CompletionStage, 它会异步地获取汽车的评分(可能又是一个REST API调用)当所有的汽车填好评分后,我们结束这个列表,所以我们调用 allOf
得到最终的阶段, 它在前面阶段所有阶段完成后才完成。在最终的阶段调用 whenComplete()
,我们打印出每个汽车和它的评分。
cars().thenCompose(cars -> {
ListupdatedCars = 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)。后台回复 学习资料 领取学习视频
如有收获,点个在看,诚挚感谢