理解Java8里面CompletableFuture异步编程
Posted 我是攻城师
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了理解Java8里面CompletableFuture异步编程相关的知识，希望对你有一定的参考价值。
Java8主要的语言增强的能力有：
（1）lambda表达式
（2）stream式操作
（3）CompletableFuture
其中第三个特性，就是今天我们想要聊的话题，正是因为CompletableFuture的出现，才使得使用Java进行异步编程提供了可能。
什么是CompletableFuture？
CompletableFuture在Java里面被用于异步编程，异步通常意味着非阻塞，可以使得我们的任务单独运行在与主线程分离的其他线程中，并且通过 回调可以在主线程中得到异步任务的执行状态，是否完成，和是否异常等信息。CompletableFuture实现了Future
Future vs CompletableFuture
Futrue在Java里面，通常用来表示一个异步任务的引用，比如我们将任务提交到线程池里面，然后我们会得到一个Futrue，在Future里面有isDone方法来 判断任务是否处理结束，还有get方法可以一直阻塞直到任务结束然后获取结果，但整体来说这种方式，还是同步的，因为需要客户端不断阻塞等待或者不断轮询才能知道任务是否完成。
Future的主要缺点如下：
（1）不支持手动完成
这个意思指的是，我提交了一个任务，但是执行太慢了，我通过其他路径已经获取到了任务结果，现在没法把这个任务结果，通知到正在执行的线程，所以必须主动取消或者一直等待它执行完成。
（2）不支持进一步的非阻塞调用
这个指的是我们通过Future的get方法会一直阻塞到任务完成，但是我还想在获取任务之后，执行额外的任务，因为Future不支持回调函数，所以无法实现这个功能。
（3）不支持链式调用
这个指的是对于Future的执行结果，我们想继续传到下一个Future处理使用，从而形成一个链式的pipline调用，这在Future中是没法实现的。
（4）不支持多个Future合并
比如我们有10个Future并行执行，我们想在所有的Future运行完毕之后，执行某些函数，是没法通过Future实现的。
（5）不支持异常处理
Future的API没有任何的异常处理的api，所以在异步运行时，如果出了问题是不好定位的。
简单的使用CompletableFuture
1，先看一个最简单的例子
在主线程里面创建一个CompletableFuture，然后主线程调用get方法会阻塞，最后我们在一个子线程中 使其终止。
CompletableFuture<String> completableFuture=new CompletableFuture<String>();
Runnable runnable=new Runnable() {
@Override
public void run() {
try {
TimeUnit.SECONDS.sleep(3);
System.out.println(getThreadName()+"执行.....");
completableFuture.complete("success");//在子线程中完成主线程completableFuture的完成
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
Thread t1=new Thread(runnable);
t1.start();//启动子线程
String result=completableFuture.get();//主线程阻塞，等待完成
System.out.println(getThreadName()+"1x: "+result);
输出结果：
Thread-0线程=> 执行.....
main线程=> 1x: success
2，运行一个简单的没有返回值的异步任务
CompletableFuture<Void> future=CompletableFuture.runAsync(new Runnable() {
@Override
public void run() {
try {
System.out.println(getThreadName()+"正在执行一个没有返回值的异步任务。");
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
future.get();
System.out.println(getThreadName()+" 结束。");
输出如下：
ForkJoinPool.commonPool-worker-1线程=> 正在执行一个没有返回值的异步任务。
main线程=> 结束。
从上面我们可以看到CompletableFuture默认运行使用的是ForkJoin的的线程池。当然，你也可以用lambda表达式使得代码更精简。
3，运行一个有返回值的异步任务
CompletableFuture<String> future=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try {
System.out.println(getThreadName()+"正在执行一个有返回值的异步任务。");
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK";
}
});
String result=future.get();
System.out.println(getThreadName()+" 结果："+result);
输出结果：
ForkJoinPool.commonPool-worker-1线程=> 正在执行一个有返回值的异步任务。
main线程=> 结果：OK
当然，上面默认的都是ForkJoinPool我们也可以换成Executor相关的Pool，其api都有支持如下：
static CompletableFuture<Void> runAsync(Runnable runnable)
static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
高级的使用CompletableFuture
前面提到的几种使用方法是使用异步编程最简单的步骤，CompletableFuture.get()的方法会阻塞直到任务完成，这其实还是同步的概念，这对于一个异步系统是不够的，因为真正的异步是需要支持回调函数，这样以来，我们就可以直接在某个任务干完之后，接着执行回调里面的函数，从而做到真正的异步概念。
在CompletableFuture里面，我们通过thenApply(), thenAccept()，thenRun()方法，来运行一个回调函数。
（1）thenApply()
这个方法，其实用过函数式编程的人非常容易理解，类似于scala和spark的map算子，通过这个方法可以进行多次链式转化并返回最终的加工结果。
看下面一个例子：
public static void asyncCallback() throws ExecutionException, InterruptedException {
CompletableFuture<String> task=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
System.out.println(getThreadName()+"supplyAsync");
return "123";
}
});
CompletableFuture<Integer> result1 = task.thenApply(number->{
System.out.println(getThreadName()+"thenApply1");
return Integer.parseInt(number);
});
CompletableFuture<Integer> result2 = result1.thenApply(number->{
System.out.println(getThreadName()+"thenApply2");
return number*2;
});
System.out.println(getThreadName()+" => "+result2.get());
}
输出结果：
ForkJoinPool.commonPool-worker-1线程=> supplyAsync
main线程=> thenApply1
main线程=> thenApply2
main线程=> => 246
（2）thenAccept()
这个方法，可以接受Futrue的一个返回值，但是本身不在返回任何值，适合用于多个callback函数的最后一步操作使用。
例子如下：
public static void asyncCallback2() throws ExecutionException, InterruptedException {
CompletableFuture<String> task=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
System.out.println(getThreadName()+"supplyAsync");
return "123";
}
});
CompletableFuture<Integer> chain1 = task.thenApply(number->{
System.out.println(getThreadName()+"thenApply1");
return Integer.parseInt(number);
});
CompletableFuture<Integer> chain2 = chain1.thenApply(number->{
System.out.println(getThreadName()+"thenApply2");
return number*2;
});
CompletableFuture<Void> result=chain2.thenAccept(product->{
System.out.println(getThreadName()+"thenAccept="+product);
});
result.get();
System.out.println(getThreadName()+"end");
}
结果如下：
ForkJoinPool.commonPool-worker-1线程=> supplyAsync
main线程=> thenApply1
main线程=> thenApply2
main线程=> thenAccept=246
main线程=> end
（3） thenRun()
这个方法与上一个方法类似，一般也用于回调函数最后的执行，但这个方法不接受回调函数的返回值，纯粹就代表执行任务的最后一个步骤：
public static void asyncCallback3() throws ExecutionException, InterruptedException {
CompletableFuture.supplyAsync(()->{
System.out.println(getThreadName()+"supplyAsync: 一阶段任务");
return null;
}).thenRun(()->{
System.out.println(getThreadName()+"thenRun: 收尾任务");
}).get();
}
结果：
ForkJoinPool.commonPool-worker-1线程=> supplyAsync: 一阶段任务
main线程=> thenRun: 收尾任务
这里注意，截止到目前，前面的例子代码只会涉及两个线程，一个是主线程一个是ForkJoinPool池的线程，但其实上面的每一步都是支持异步运行的，其api如下：
// thenApply() variants
<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
我们看下改造后的一个例子：
public static void asyncCallback4() throws ExecutionException, InterruptedException {
CompletableFuture<String> ref1= CompletableFuture.supplyAsync(()->{
try {
System.out.println(getThreadName()+"supplyAsync开始执行任务1.... ");
// TimeUnit.SECONDS.sleep(1);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(getThreadName()+"supplyAsync: 任务1");
return null;
});
CompletableFuture<String> ref2= CompletableFuture.supplyAsync(()->{
try {
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(getThreadName()+"thenApplyAsync: 任务2");
return null;
});
CompletableFuture<String> ref3=ref2.thenApplyAsync(value->{
System.out.println(getThreadName()+"thenApplyAsync: 任务2的子任务");
return null;
});
Thread.sleep(4000);
System.out.println(getThreadName()+ref3.get());
}
输出结果如下：
ForkJoinPool.commonPool-worker-1线程=> supplyAsync开始执行任务1....
ForkJoinPool.commonPool-worker-1线程=> supplyAsync: 任务1
ForkJoinPool.commonPool-worker-1线程=> supplyAsync: 任务2
ForkJoinPool.commonPool-worker-2线程=> thenApplyAsync: 任务2的子任务
main线程=> null
我们可以看到，ForkJoin池的线程1，执行了前面的三个任务，但是第二个任务的子任务，因为我们了使用也异步提交所以它用的线程是ForkJoin池的线程2，最终由于main线程处执行了get是最后结束的。
还有一点需要注意：
ForkJoinPool所有的工作线程都是守护模式的，也就是说如果主线程退出，那么整个处理任务都会结束，而不管你当前的任务是否执行完。如果需要主线程等待结束，可采用ExecutorsThreadPool，如下：
ExecutorService pool = Executors.newFixedThreadPool(5);
final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
... }, pool);
（4）thenCompose合并两个有依赖关系的CompletableFutures的执行结果
CompletableFutures在执行两个依赖的任务合并时，会返回一个嵌套的结果列表，为了避免这种情况我们可以使用thenCompose来返回，直接获取最顶层的结果数据即可：
public static void asyncCompose() throws ExecutionException, InterruptedException {
CompletableFuture<String> future1=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
return "1";
}
});
CompletableFuture<String>nestedResult = future1.thenCompose(value->
CompletableFuture.supplyAsync(()->{
return value+"2";
}));
System.out.println(nestedResult.get());
}
（5）thenCombine合并两个没有依赖关系的CompletableFutures任务
CompletableFuture<Double> d1= CompletableFuture.supplyAsync(new Supplier<Double>() {
@Override
public Double get() {
return 1d;
}
});
CompletableFuture<Double> d2= CompletableFuture.supplyAsync(new Supplier<Double>() {
@Override
public Double get() {
return 2d;
}
});
CompletableFuture<Double> result= d1.thenCombine(d2,(number1,number2)->{
return number1+number2;
});
System.out.println(result.get());
（6）合并多个任务的结果allOf与anyOf
上面说的是两个任务的合并，那么多个任务需要使用allOf或者anyOf方法。
allOf适用于，你有一系列独立的future任务，你想等其所有的任务执行完后做一些事情。举个例子，比如我想下载100个网页，传统的串行，性能肯定不行，这里我们采用异步模式，同时对100个网页进行下载，当所有的任务下载完成之后，我们想判断每个网页是否包含某个关键词。
下面我们通过随机数来模拟上面的这个场景如下：
public static void mutilTaskTest() throws ExecutionException, InterruptedException {
//添加n个任务
CompletableFuture<Double> array[]=new CompletableFuture[3];
for ( int i = 0; i < 3; i++) {
array[i]=CompletableFuture.supplyAsync(new Supplier<Double>() {
@Override
public Double get() {
return Math.random();
}
});
}
//获取结果的方式一
// CompletableFuture.allOf(array).get();
// for(CompletableFuture<Double> cf:array){
// if(cf.get()>0.6){
// System.out.println(cf.get());
// }
// }
//获取结果的方式二，过滤大于指定数字，在收集输出
List<Double> rs= Stream.of(array).map(CompletableFuture::join).filter(number->number>0.6).collect(Collectors.toList());
System.out.println(rs);
}
结果如下：
[0.8228784717152199]
注意其中的join方法和get方法类似，仅仅在于在Future不能正常完成的时候抛出一个unchecked的exception，这可以确保它用在Stream的map方法中，直接使用get是没法在map里面运行的。
anyOf方法，也比较简单，意思就是只要在多个future里面有一个返回，整个任务就可以结束，而不需要等到每一个future结束。
CompletableFuture<String> f1=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try {
TimeUnit.SECONDS.sleep(4);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "wait 4 seconds";
}
});
CompletableFuture<String> f2=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "wait 2 seconds";
}
});
CompletableFuture<String> f3=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try {
TimeUnit.SECONDS.sleep(4);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "wait 10 seconds";
}
});
CompletableFuture<Object> result= CompletableFuture.anyOf(f1,f2,f3);
System.out.println(result.get());
输出结果：
wait 2 seconds
注意由于Anyof返回的是其中任意一个Future所以这里没有明确的返回类型，统一使用Object接受，留给使用端处理。
（7）exceptionally异常处理
异常处理是异步计算的一个重要环节，下面看看如何在CompletableFuture中使用：
int age=-1;
CompletableFuture<String> task= CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
if(age<0){
throw new IllegalArgumentException("性别必须大于0");
}
if(age<18){
return "未成年人";
}
return "成年人";
}
}).exceptionally(ex->{
System.out.println(ex.getMessage());
return "发生 异常"+ex.getMessage();
});
System.out.println(task.get());
结果如下：
java.lang.IllegalArgumentException: 性别必须大于0
发生 异常java.lang.IllegalArgumentException: 性别必须大于0
此外还有另外一种异常捕捉方法handle，无论发生异常都会执行，示例如下：
int age=10;
CompletableFuture<String> task= CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
if(age<0){
throw new IllegalArgumentException("性别必须大于0");
}
if(age<18){
return "未成年人";
}
return "成年人";
}
}).handle((res,ex)->{
System.out.println("执行handle");
if(ex!=null){
System.out.println("发生异常");
return "发生 异常"+ex.getMessage();
}
return res;
});
System.out.println(task.get());
输出结果：
执行handle
发生异常
发生 异常java.lang.IllegalArgumentException: 性别必须大于0
注意上面的方法如果正常执行，也会执行handle方法。
JDK9 CompletableFuture 类增强的主要内容
（1）支持对异步方法的超时调用
orTimeout()
completeOnTimeout()
（2）支持延迟调用
Executor delayedExecutor(long delay, TimeUnit unit, Executor executor)
Executor delayedExecutor(long delay, TimeUnit unit)
详细内容，可以参考Oracle官网文档，这里不再过多介绍。
总结：
本文主要介绍了CompletableFuture的定义，概念及在Java中使用的例子，通过CompletableFuture我们可以实现异步编程的能力，从而使得我们开发的任务可以拥有更强大的能力。
以上是关于理解Java8里面CompletableFuture异步编程的主要内容，如果未能解决你的问题，请参考以下文章