前面两篇文章已经整理了CompletableFuture大部分的特性,本文会整理完CompletableFuture余下的特性,以及将它跟RxJava进行比较。
3.6 Either
Either 表示的是两个CompletableFuture,当其中任意一个CompletableFuture计算完成的时候就会执行。
方法名 | 描述 |
---|---|
acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) | 当任意一个CompletableFuture完成的时候,action这个消费者就会被执行。 |
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) | 当任意一个CompletableFuture完成的时候,action这个消费者就会被执行。使用ForkJoinPool |
acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) | 当任意一个CompletableFuture完成的时候,action这个消费者就会被执行。使用指定的线程池 |
Random random = new Random();
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{
try {
Thread.sleep(random.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return "from future1";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{
try {
Thread.sleep(random.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return "from future2";
});
CompletableFuture<Void> future = future1.acceptEither(future2,str->System.out.println("The future is "+str));
try {
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
执行结果:The future is from future1 或者 The future is from future2。
因为future1和future2,执行的顺序是随机的。
applyToEither 跟 acceptEither 类似。
方法名 | 描述 |
---|---|
applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn) | 当任意一个CompletableFuture完成的时候,fn会被执行,它的返回值会当作新的CompletableFuture的计算结果。 |
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn) | 当任意一个CompletableFuture完成的时候,fn会被执行,它的返回值会当作新的CompletableFuture的计算结果。使用ForkJoinPool |
applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor) | 当任意一个CompletableFuture完成的时候,fn会被执行,它的返回值会当作新的CompletableFuture的计算结果。使用指定的线程池 |
Random random = new Random();
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{
try {
Thread.sleep(random.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return "from future1";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{
try {
Thread.sleep(random.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return "from future2";
});
CompletableFuture<String> future = future1.applyToEither(future2,str->"The future is "+str);
try {
System.out.println(future.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
执行结果也跟上面的程序类似。
3.7 其他方法
allOf、anyOf是CompletableFuture的静态方法。
3.7.1 allOf
方法名 | 描述 |
---|---|
allOf(CompletableFuture<?>… cfs) | 在所有Future对象完成后结束,并返回一个future。 |
allOf()方法所返回的CompletableFuture,并不能组合前面多个CompletableFuture的计算结果。于是我们借助Java 8的Stream来组合多个future的结果。
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "tony");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "cafei");
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "aaron");
CompletableFuture.allOf(future1, future2, future3)
.thenApply(v ->
Stream.of(future1, future2, future3)
.map(CompletableFuture::join)
.collect(Collectors.joining(" ")))
.thenAccept(System.out::print);
执行结果:
tony cafei aaron
3.7.2 anyOf
方法名 | 描述 |
---|---|
anyOf(CompletableFuture<?>… cfs) | 在任何一个Future对象结束后结束,并返回一个future。 |
Random rand = new Random();
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(rand.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return "from future1";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(rand.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return "from future2";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(rand.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
return "from future3";
});
CompletableFuture<Object> future = CompletableFuture.anyOf(future1,future2,future3);
try {
System.out.println(future.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
使用anyOf()时,只要某一个future完成,就结束了。所以执行结果可能是”from future1”、”from future2”、”from future3”中的任意一个。
anyOf 和 acceptEither、applyToEither的区别在于,后两者只能使用在两个future中,而anyOf可以使用在多个future中。
3.8 CompletableFuture异常处理
CompletableFuture在运行时如果遇到异常,可以使用get()并抛出异常进行处理,但这并不是一个最好的方法。CompletableFuture本身也提供了几种方式来处理异常。
3.8.1 exceptionally
方法名 | 描述 |
---|---|
exceptionally(Function |
只有当CompletableFuture抛出异常的时候,才会触发这个exceptionally的计算,调用function计算值。 |
CompletableFuture.supplyAsync(() -> "hello world")
.thenApply(s -> {
s = null;
int length = s.length();
return length;
}).thenAccept(i -> System.out.println(i))
.exceptionally(t -> {
System.out.println("Unexpected error:" + t);
return null;
});
执行结果:
Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException
对上面的代码稍微做了一下修改,修复了空指针的异常。
CompletableFuture.supplyAsync(() -> "hello world")
.thenApply(s -> {
// s = null;
int length = s.length();
return length;
}).thenAccept(i -> System.out.println(i))
.exceptionally(t -> {
System.out.println("Unexpected error:" + t);
return null;
});
执行结果:
11
3.8.2 whenComplete
whenComplete 在上一篇文章其实已经介绍过了,在这里跟exceptionally的作用差不多,可以捕获任意阶段的异常。如果没有异常的话,就执行action。
CompletableFuture.supplyAsync(() -> "hello world")
.thenApply(s -> {
s = null;
int length = s.length();
return length;
}).thenAccept(i -> System.out.println(i))
.whenComplete((result, throwable) -> {
if (throwable != null) {
System.out.println("Unexpected error:"+throwable);
} else {
System.out.println(result);
}
});
执行结果:
Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException
跟whenComplete相似的方法是handle,handle的用法在上一篇文章中也已经介绍过。
四. CompletableFuture VS Java8 Stream VS RxJava1 & RxJava2
CompletableFuture 有很多特性跟RxJava很像,所以将CompletableFuture、Java 8 Stream和RxJava做一个相互的比较。
各种对象 | composable | lazy | resuable | async | cached | push | back pressure |
---|---|---|---|---|---|---|---|
CompletableFuture | 支持 | 不支持 | 支持 | 支持 | 支持 | 支持 | 不支持 |
Stream | 支持 | 支持 | 不支持 | 不支持 | 不支持 | 不支持 | 不支持 |
Observable(RxJava1) | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
Observable(RxJava2) | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 | 不支持 |
Flowable(RxJava2) | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
五. 总结
Java 8提供了一种函数风格的异步和事件驱动编程模型CompletableFuture,它不会造成堵塞。CompletableFuture背后依靠的是fork/join框架来启动新的线程实现异步与并发。当然,我们也能通过指定线程池来做这些事情。
CompletableFuture特别是对微服务架构而言,会有很大的作为。举一个具体的场景,电商的商品页面可能会涉及到商品详情服务、商品评论服务、相关商品推荐服务等等。获取商品的信息时(/productdetails?productid=xxx),需要调用多个服务来处理这一个请求并返回结果。这里可能会涉及到并发编程,我们完全可以使用Java 8的CompletableFuture或者RxJava来实现。