前言:jdk1.8 lamda记录
异步执行动作
static void thenApplyAsyncExample() {
CompletableFuture<String>cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
assertTrue(Thread.currentThread().isDaemon());
randomSleep();
returns.toUpperCase();
});
assertNull(cf.getNow(null));
assertEquals("MESSAGE", cf.join());
}
使用固定的线程池完成异步执行动作
可以通过使用线程池方式来管理异步动作申请,以下代码基于固定的线程池,也是做一个大写字母转换动作
staticExecutorService 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<String>cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));
assertFalse(Thread.currentThread().isDaemon());
randomSleep();
returns.toUpperCase();
}, executor);
assertNull(cf.getNow(null));
assertEquals("MESSAGE", cf.join());
}
作为消费者消费计算结果
假设我们本次计算只需要前一次的计算结果,而不需要返回本次计算结果,那就有点类似于生产者(前一次计算)-消费者(本次计算)模式了
static void thenAcceptExample() {
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture("thenAccept message")
.thenAccept(s ->result.append(s));
assertTrue("Result was empty", result.length() > 0);
}
消费者是同步执行的,所以不需要在 CompletableFuture 里对结果进行合并。
异步消费
相较于前一个示例的同步方式,我们也对应有异步方式,代码如清单 11 所示。
static void thenAcceptAsyncExample() {
StringBuilder result = new StringBuilder();
CompletableFuture<Void>cf = CompletableFuture.completedFuture("thenAcceptAsync message")
.thenAcceptAsync(s ->result.append(s));
cf.join();
assertTrue("Result was empty", result.length() > 0);
}
计算过程中的异常
接下来介绍异步操作过程中的异常情况处理。下面这个示例中我们会在字符转换异步请求中刻意延迟 1 秒钟,然后才会提交到 ForkJoinPool 里面去执行。
static void completeExceptionallyExample() {
CompletableFuture<String>cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
CompletableFuture<String>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(计算完毕),然后调用 thenApplyAsync 返回一个新的 CompletableFuture,接着通过使用 delayedExecutor(timeout, timeUnit)方法延迟 1 秒钟执行。然后我们创建一个 handler(exceptionHandler),它会处理异常,返回另一个字符串"message upon cancel"。接下来进入 join()方法,执行大写转换操作,并且抛出 CompletionException 异常。
取消计算任务
与前面一个异常处理的示例类似,我们可以通过调用 cancel(boolean mayInterruptIfRunning)方法取消计算任务。此外,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());
}
一个 CompletableFuture VS 两个异步计算
我们可以创建一个 CompletableFuture 接收两个异步计算的结果,下面代码首先创建了一个 String 对象,接下来分别创建了两个 CompletableFuture 对象 cf1 和 cf2,cf2 通过调用 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"));
}
使用消费者替换用于处理异步计算结果
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"));
}
运行两个阶段后执行
下面这个示例程序两个阶段执行完毕后返回结果,首先将字符转为大写,然后将字符转为小写,在两个计算阶段都结束之后触发 CompletableFuture
。
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);
}
也可以通过以下方式处理异步计算结果,
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());
}
整合两个计算结果
我们可以通过 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));
}
按照同步方式执行两个方法后再合成字符串,以下代码采用异步方式同步执行两个方法,由于异步方式情况下不能够确定哪一个方法最终执行完毕,所以我们需要调用 join()方法等待后一个方法结束后再合成字符串,这一点和线程的 join()方法是一致的,主线程生成并起动了子线程,如果子线程里要进行大量的耗时的运算,主线程往往将于子线程之前结束,但是如果主线程处理完其他的事务后,需要用到子线程的处理结果,也就是主线程需要等待子线程执行完成之后再结束,这个时候就要用到 join()方法了,即 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());
}
除了 thenCombine()方法以外,还有另外一种方法-thenCompose(),这个方法也会实现两个方法执行后的返回结果的连接。
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());
}
anyOf()方法
以下代码模拟了如何在几个计算过程中任意一个完成后创建 CompletableFuture,在这个例子中,我们创建了几个计算过程,然后转换字符串到大写字符。由于这些 CompletableFuture 是同步执行的(下面这个例子使用的是 thenApply()方法,而不是 thenApplyAsync()方法),使用 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);
}
当所有的 CompletableFuture 完成后创建 CompletableFuture
清单 22 所示我们会以同步方式执行多个异步计算过程,在所有计算过程都完成后,创建一个 CompletableFuture。
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);
}
相较于前一个同步示例,我们也可以异步执行
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);
}
首先异步地通过调用 cars()方法获取 Car 对象,返回一个 CompletionStage<List>实例。Cars()方法可以在内部使用调用远端服务器上的 REST 服务等类似场景。
然后和其他的 CompletionStage<List>组合,通过调用 rating(manufacturerId)方法异步地返回 CompletionStage 实例。
当所有的 Car 对象都被填充了 rating 后,调用 allOf()方法获取最终值。
调用 whenComplete()方法打印最终的评分(rating)。
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();
自己改造的lamda+CompletableFuture调用模板
public class FurtureTest {
public static void allOfExample() {
StringBuilder result = new StringBuilder();
List<String> messages = Arrays.asList("a", "b", "c");
List<CompletableFuture> futures = messages.stream()
.map(msg -> CompletableFuture.completedFuture(msg).thenApply(s ->result.append(s)))
.collect(Collectors.toList());
CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((v, th) -> {
futures.forEach(cf -> { System.out.println("a");});
result.append("done");
});
}
public static String delayedLowerCase(String s){
return s.toString();
}
public static void main(String[] argas){
allOfExample();
}
}