CompletableFuture
Future虽然可以实现获取异步执行结果的需求，但是它没有提供通知的机制，我们无法得知Future什么时候完成。要么使用阻塞，在future.get()的地方等待future返回的结果，这时又变成同步操作。要么使用isDone()轮询地判断Future是否完成，这样会耗费CPU的资源。在Java 8中, 新增加了一个CompletableFuture，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。
说明示例
@Test
public void test1() throws Exception {
// 创建一个CompletableFuture对象
CompletableFuture<String> future1 = new CompletableFuture<>();
// 对CompletableFuture设置一个结束通知
future1.whenComplete(new BiConsumer<String, Throwable>() {
@Override
public void accept(String s, Throwable throwable) {
log.info("s={},throwable={}", s, throwable);
}
});
// 复杂的事情开一个线程
Thread thread = new Thread(() -> {
String result1 = Computer.longTimeDoSomething();
// 完成异步执行，并返回future的结果
future1.complete(result1);
});
thread.start();
String result1 = future1.get();
log.info("结果={}", result1);
}
CompletableFuture类实现了CompletionStage和Future接口，我们还是可以像以前一样通过阻塞（如上get()方法）或者轮询（isDone()方法）的方式获得结果，那就没有什么新特性了。
实例创建
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
runAsync表示执行一个不需要任务结果的异步任务，supplyAsync表示执行一个需要结果的一个异步任务。请注意，它们都不需要我们手动创建一个thread，内部已经帮我们处理了。没有指定Executor的方法会使用ForkJoinPool.commonPool()作为它的线程池执行异步代码，指定Executor的参数就用自己的线程池。
示例：
@Test
public void test2() throws Exception {
// 做一个复杂的任务
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
});
// 设置任务结果通知
Future future = completableFuture.whenComplete((result, throwable) -> {
log.info("结果={},异常信息={}", result, throwable);
});
log.info("do something ...");
// 获取任务结果（会阻塞主线程）
log.info("result={}", future.get());
}
以上代码函数式编程示例是：
@Test
public void test3() throws Exception {
Object[] objects = new Object[2];
CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
}).whenComplete((result, throwable) -> {
log.info("结果={},异常信息={}", result, throwable);
objects[0] = result;
});
System.in.read();
}
对异步任务结果的处理
CompletableFuture继承了CompletionStage接口，CompletionStage接口有很多对结果的处理方法，每个方法都三个，形式是：xxx，xxxAsync，xxxAsync
whenComplete (BiConsumer<? super T, ? super Throwable> action);
whenCompleteAsync (BiConsumer<? super T, ? super Throwable> action);
whenCompleteAsync (BiConsumer<? super T, ? super Throwable> action, Executor executor);
从命名上也好理解，以Async结尾的方法都是可以异步执行的，如果指定了线程池，会在指定的线程池中执行，如果没有指定，默认会在ForkJoinPool.commonPool()线程池中执行。
当运行完成时: whenComplete
whenCompletes可以对结果的记录
@Test
public void testWhenComplete() throws Exception {
CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
});
Future future1 = completableFuture1.whenComplete((result, throwable) -> {
log.info("结果={},异常信息={}", result, throwable);
});
log.info("do something ...");
log.info("future1 result={}", future1.get());
System.in.read();
}
[ main]do something ...
[worker-1]结果=DoSomething-Result,异常信息=null
[ main]future1 result=DoSomething-Result
进行变换：thenApply
可以对结果进行修改
@Test
public void testThenApply() throws Exception {
CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
});
Future future = completableFuture1.thenApply((result) -> {
return result + "_" + System.currentTimeMillis();
});
log.info("do something ...");
log.info("future2 result={}", future.get());
System.in.read();
}
[ main]do something ...
[ main]future2 result=DoSomething-Result_1597467670713
进行消费：thenAccept
收到对结果后可以做别的事情处理
@Test
public void testThenAccept() throws Exception {
CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
});
Future future = completableFuture1.thenAccept((result) -> {
log.info("对结果进行消费处理,result={}", result);
});
log.info("do something ...");
log.info("future3 result={}", future.get());
System.in.read();
}
[ main]do something ...
[worker-1]对结果进行消费处理,result=DoSomething-Result
[ main]future3 result=null
对结果不关心，执行下一个操作：thenRun
对结果不关系，做自己的处理
@Test
public void testThenRun() throws Exception {
CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
});
Future future = completableFuture1.thenRun(() -> {
log.info("对结果不关系，做自己的处理");
});
log.info("do something ...");
log.info("future4 result={}", future.get());
System.in.read();
}
[ main]do something ...
[worker-1]对结果不关系，做自己的处理
[ main]future4 result=null
将两个任务结果结合：thenCombine
将两个任务的结果一起返回
@Test
public void testThenCombine() throws Exception {
CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
});
CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
return Computer.complexCompute();
});
Object[] results = (Object[]) future1.thenCombine(future2, (result1, result2) -> {
Object[] resultObj = new Object[2];
resultObj[0] = result1;
resultObj[1] = result2;
return resultObj;
}).join();
log.info("result0={},result1={}", results[0], results[1]);
}
[ main]result0=DoSomething-Result,result1=100
在两个任务都运行完执行：thenAcceptBoth
可以对收到两个任务的结果后做处理
@Test
public void testThenAcceptBoth() throws Exception {
CompletableFuture.supplyAsync(() -> {
return Computer.longTimeDoSomething();
}).thenAcceptBoth(CompletableFuture.supplyAsync(() -> {
return Computer.complexCompute();
}),(s1,s2)->{
System.out.println(s1);
System.out.println(s2);
});
System.in.read();
}
DoSomething-Result
100
网友评论