一个例子回顾 Future
因为CompletableFuture
实现了Future
接口,先来回顾Future
吧。Future
是Java5新加的一个接口,它提供了一种异步并行计算的功能。如果主线程需要执行一个很耗时的计算任务,就可以通过future
把这个任务放到异步线程中执行。主线程继续处理其他任务,处理完成后,再通过Future
获取计算结果。
来看个简单例子吧,假设有两个任务服务,一个查询用户基本信息,一个是查询用户勋章信息。如下,
public class UserInfoService {
public UserInfo getUserInfo(Long userId) throws InterruptedException {
Thread.sleep(300);//模拟调用耗时
return new UserInfo("666", "Hello", 27); //一般是查数据库,或者远程调用返回的
}
}
public class MedalService {
public MedalInfo getMedalInfo(long userId) throws InterruptedException {
Thread.sleep(500); //模拟调用耗时
return new MedalInfo("666", "守护勋章");
}
}
接下来,来演示下,在主线程中是如何使用Future
来进行异步调用的。
public class FutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
UserInfoService userInfoService = new UserInfoService();
MedalService medalService = new MedalService();
long userId =666L;
long startTime = System.currentTimeMillis();
//调用用户服务获取用户基本信息
FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() {
@Override
public UserInfo call() throws Exception {
return userInfoService.getUserInfo(userId);
}
});
executorService.submit(userInfoFutureTask);
Thread.sleep(300); //模拟主线程其它操作耗时
FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() {
@Override
public MedalInfo call() throws Exception {
return medalService.getMedalInfo(userId);
}
});
executorService.submit(medalInfoFutureTask);
UserInfo userInfo = userInfoFutureTask.get();//获取个人信息结果
MedalInfo medalInfo = medalInfoFutureTask.get();//获取勋章信息结果
System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
}
}
运行结果:
总共用时806ms
如果不使用Future进行并行异步调用,而是在主线程串行进行的话,耗时大约为300+500+300 = 1100 ms。可以发现,future+线程池异步配合,提高了程序的执行效率。
但是Future对于结果的获取,不是很友好,只能通过阻塞或者轮询的方式得到任务的结果。
Future.get()
就是阻塞调用,在线程获取结果之前get
方法会一直阻塞。- Future提供了一个
isDone
方法,可以在程序中轮询这个方法查询执行结果。
阻塞的方式和异步编程的设计理念相违背,而轮询的方式会耗费无谓的CPU资源。因此,JDK8设计出CompletableFuture
。CompletableFuture
提供了一种观察者模式类似的机制,可以让任务执行完成后通知监听的一方。
一个例子走进CompletableFuture
还是基于以上Future
的例子,改用CompletableFuture
来实现
public class FutureTest {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
UserInfoService userInfoService = new UserInfoService();
MedalService medalService = new MedalService();
long userId =666L;
long startTime = System.currentTimeMillis();
//调用用户服务获取用户基本信息
CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId));
Thread.sleep(300); //模拟主线程其它操作耗时
CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId));
UserInfo userInfo = completableUserInfoFuture.get(2,TimeUnit.SECONDS);//获取个人信息结果
MedalInfo medalInfo = completableMedalInfoFuture.get();//获取勋章信息结果
System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
}
}
可以发现,使用CompletableFuture
,代码简洁了很多。CompletableFuture
的supplyAsync
方法,提供了异步执行的功能,线程池也不用单独创建了。实际上,它CompletableFuture
使用了默认线程池是ForkJoinPool.commonPool
。CompletableFuture
提供了几十种方法辅助异步任务场景。这些方法包括创建异步任务、任务异步回调、多个任务组合处理等方面。
CompletableFuture
使用场景
创建异步任务
CompletableFuture
创建异步任务,一般有supplyAsync
和runAsync
两个方法
创建异步任务
supplyAsync
执行CompletableFuture
任务,支持返回值runAsync
执行CompletableFuture
任务,没有返回值。supplyAsync
方法//使用默认内置线程池ForkJoinPool.commonPool(),根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程,根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
runAsync
方法//使用默认内置线程池ForkJoinPool.commonPool(),根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable)
//自定义线程,根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
实例代码如下:
public class FutureTest {
public static void main(String[] args) {
//可以自定义线程池
ExecutorService executor = Executors.newCachedThreadPool();
//runAsync的使用
CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run,Hello"), executor);
//supplyAsync的使用
CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {
System.out.print("supply,Hello");
return "Hello"; }, executor);
//runAsync的future没有返回值,输出null
System.out.println(runFuture.join());
//supplyAsync的future,有返回值
System.out.println(supplyFuture.join());
executor.shutdown(); // 线程池需要关闭
}
}
//输出
run,Hello
null
supply,Hello
任务异步回调
1.
thenRun
/thenRunAsync
public CompletableFuture<Void> thenRun(Runnable action);
public CompletableFuture<Void> thenRunAsync(Runnable action);
CompletableFuture
的thenRun
方法,通俗点讲就是,做完第一个任务后,再做第二个任务。某个任务执行完成后,执行回调方法;但是前后两个任务没有参数传递,第二个任务也没有返回值public class FutureThenRunTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("先执行第一个CompletableFuture方法任务");
return "Hello";
}
);
CompletableFuture thenRunFuture = orgFuture.thenRun(() -> {
System.out.println("接着执行第二个任务");
});
System.out.println(thenRunFuture.get());
}
}
//输出
先执行第一个CompletableFuture方法任务
接着执行第二个任务
null
thenRun
和thenRunAsync
有什么区别呢?可以看下源码: ```java private static final Executor asyncPool = useCommonPool ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
public CompletableFuture
public CompletableFuture
如果执行第一个任务的时候,传入了一个自定义线程池:
- 调用`thenRun`方法执行第二个任务时,则第二个任务和第一个任务是共用同一个线程池。
- 调用`thenRunAsync`执行第二个任务时,则第一个任务使用的是你自己传入的线程池,第二个任务使用的是ForkJoin线程池
TIPS: 后面介绍的`thenAccept`和`thenAcceptAsync`,`thenApply`和`thenApplyAsync`等,它们之间的区别也是这个。
<a name="PxaFH"></a>
#### 2.`thenAccept`/`thenAcceptAsync`
`CompletableFuture`的`thenAccept`方法表示,第一个任务执行完成后,执行第二个回调方法任务,会将该任务的执行结果,作为入参,传递到回调方法中,但是回调方法是没有返回值的。
```java
public class FutureThenAcceptTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("原始CompletableFuture方法任务");
return "Hello";
}
);
CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> {
if ("Hello".equals(a)) {
System.out.println("World");
}
System.out.println("Hi");
});
System.out.println(thenAcceptFuture.get());
}
}
3. thenApply
/thenApplyAsync
CompletableFuture
的thenApply
方法表示,第一个任务执行完成后,执行第二个回调方法任务,会将该任务的执行结果,作为入参,传递到回调方法中,并且回调方法是有返回值的。
public class FutureThenApplyTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("原始CompletableFuture方法任务");
return "Hello";
}
);
CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> {
if ("Hello".equals(a)) {
return "World";
}
return "Hi";
});
System.out.println(thenApplyFuture.get());
}
}
//输出
原始CompletableFuture方法任务
关注了
4. exceptionally
CompletableFuture
的exceptionally
方法表示,某个任务执行异常时,执行的回调方法;并且有抛出异常作为参数,传递到回调方法。
public class FutureExceptionTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("当前线程名称:" + Thread.currentThread().getName());
throw new RuntimeException();
}
);
CompletableFuture<String> exceptionFuture = orgFuture.exceptionally((e) -> {
e.printStackTrace();
return "你的程序异常啦";
});
System.out.println(exceptionFuture.get());
}
}
//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
java.util.concurrent.CompletionException: java.lang.RuntimeException
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.RuntimeException
at cn.eovie.future.FutureWhenTest.lambda$main$0(FutureWhenTest.java:13)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
... 5 more
你的程序异常啦
5. whenComplete
方法
CompletableFuture
的whenComplete
方法表示,某个任务执行完成后,执行的回调方法,无返回值;并且whenComplete
方法返回的CompletableFuture
的result
是上个任务的结果。
public class FutureWhenTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("当前线程名称:" + Thread.currentThread().getName());
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "Hello";
}
);
CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> {
System.out.println("当前线程名称:" + Thread.currentThread().getName());
System.out.println("上个任务执行完啦,还把" + a + "传过来");
if ("Hello".equals(a)) {
System.out.println("666");
}
System.out.println("233333");
});
System.out.println(rstFuture.get());
}
}
//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
当前线程名称:ForkJoinPool.commonPool-worker-1
上个任务执行完啦,还把Hello传过来
666
233333
Hello
6. handle
方法
CompletableFuture
的handle
方法表示,某个任务执行完成后,执行回调方法,并且是有返回值的;并且handle
方法返回的CompletableFuture
的result
是回调方法执行的结果。
public class FutureHandlerTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
()->{
System.out.println("当前线程名称:" + Thread.currentThread().getName());
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "Hello";
}
);
CompletableFuture<String> rstFuture = orgFuture.handle((a, throwable) -> {
System.out.println("上个任务执行完啦,还把" + a + "传过来");
if ("Hello".equals(a)) {
System.out.println("666");
return "关注了";
}
System.out.println("233333");
return null;
});
System.out.println(rstFuture.get());
}
}
//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
上个任务执行完啦,还把Hello传过来
666
关注了
多个任务组合处理
AND组合关系
thenCombine
/ thenAcceptBoth
/ runAfterBoth
都表示:将两个CompletableFuture
组合起来,只有这两个都正常执行完了,才会执行某个任务。
区别在于:
thenCombine
:会将两个任务的执行结果作为方法入参,传递到指定方法中,且有返回值thenAcceptBoth
: 会将两个任务的执行结果作为方法入参,传递到指定方法中,且无返回值runAfterBoth
不会把执行结果当做方法入参,且没有返回值。public class ThenCombineTest {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<String> first = CompletableFuture.completedFuture("第一个异步任务");
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture
//第二个异步任务
.supplyAsync(() -> "第二个异步任务", executor)
// (w, s) -> System.out.println(s) 是第三个任务
.thenCombineAsync(first, (s, w) -> {
System.out.println(w);
System.out.println(s);
return "两个异步任务的组合";
}, executor);
System.out.println(future.join());
executor.shutdown();
}
}
//输出
第一个异步任务
第二个异步任务
两个异步任务的组合
OR 组合的关系
applyToEither
/acceptEither
/runAfterEither
都表示:将两个CompletableFuture
组合起来,只要其中一个执行完了,就会执行某个任务。
区别在于:applyToEither
:会将已经执行完成的任务,作为方法入参,传递到指定方法中,且有返回值acceptEither
: 会将已经执行完成的任务,作为方法入参,传递到指定方法中,且无返回值runAfterEither
:不会把执行结果当做方法入参,且没有返回值。public class AcceptEitherTest {
public static void main(String[] args) {
//第一个异步任务,休眠2秒,保证它执行晚点
CompletableFuture<String> first = CompletableFuture.supplyAsync(()->{
try{
Thread.sleep(2000L);
System.out.println("执行完第一个异步任务");}
catch (Exception e){
return "第一个任务异常";
}
return "第一个异步任务";
});
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<Void> future = CompletableFuture
//第二个异步任务
.supplyAsync(() -> {
System.out.println("执行完第二个任务");
return "第二个任务";}
, executor)
//第三个任务
.acceptEitherAsync(first, System.out::println, executor);
executor.shutdown();
}
}
//输出
执行完第二个任务
第二个任务
AllOf
所有任务都执行完成后,才执行
allOf
返回的CompletableFuture
。如果任意一个任务异常,allOf
的CompletableFuture
,执行get
方法,会抛出异常public class allOfFutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> a = CompletableFuture.runAsync(()->{
System.out.println("我执行完了");
});
CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {
System.out.println("我也执行完了");
});
CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b).whenComplete((m,k)->{
System.out.println("finish");
});
}
}
//输出
我执行完了
我也执行完了
finish
AnyOf
任意一个任务执行完,就执行
anyOf
返回的CompletableFuture
。如果执行的任务异常,anyOf
的CompletableFuture
,执行get
方法,会抛出异常public class AnyOfFutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> a = CompletableFuture.runAsync(()->{
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("我执行完了");
});
CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {
System.out.println("我也执行完了");
});
CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b).whenComplete((m,k)->{
System.out.println("finish");
// return "Hello";
});
anyOfFuture.join();
}
}
//输出
我也执行完了
finish
thenCompose
thenCompose
方法会在某个任务执行完成后,将该任务的执行结果,作为方法入参,去执行指定的方法。该方法会返回一个新的CompletableFuture
实例如果该
CompletableFuture
实例的result
不为null,则返回一个基于该result
新的CompletableFuture
实例;如果该
CompletableFuture
实例为null
,然后就执行这个新任务public class ThenComposeTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> f = CompletableFuture.completedFuture("第一个任务");
//第二个异步任务
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> "第二个任务", executor)
.thenComposeAsync(data -> {
System.out.println(data); return f; //使用第一个任务作为返回
}, executor);
System.out.println(future.join());
executor.shutdown();
}
}
//输出
第二个任务
第一个任务
CompletableFuture
使用有哪些注意点CompletableFuture
使异步编程更加便利的、代码更加优雅的同时,也要关注下它的一些注意点。
1.
Future
需要获取返回值,才能获取异常信息ExecutorService executorService = new ThreadPoolExecutor(5, 10, 5L,
TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));
CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
int a = 0;
int b = 666;
int c = b / a;
return true;
},executorService).thenAccept(System.out::println);
//如果不加 get()方法这一行,看不到异常信息
//future.get();
Future
需要获取返回值,才能获取到异常信息。如果不加get()
/join()
方法,看不到异常信息。使用的时候,注意一下,考虑是否加try...catch...
或者使用exceptionally
方法。2.
CompletableFuture
的get()
方法是阻塞的。CompletableFuture
的get()
方法是阻塞的,如果使用它来获取异步调用的返回值,需要添加超时时间~//反例
CompletableFuture.get();
//正例
CompletableFuture.get(5, TimeUnit.SECONDS);
3. 默认线程池的注意点
CompletableFuture
代码中又使用了默认的线程池,处理的线程个数是电脑CPU核数-1。在大量请求过来的时候,处理逻辑复杂的话,响应会很慢。一般建议使用自定义线程池,优化线程池配置参数。4. 自定义线程池时,注意饱和策略
CompletableFuture
的get()
方法是阻塞的,一般建议使用future.get(3, TimeUnit.SECONDS)
。并且一般建议使用自定义线程池。
但是如果线程池拒绝策略是DiscardPolicy
或者DiscardOldestPolicy
,当线程池饱和时,会直接丢弃任务,不会抛弃异常。因此建议,CompletableFuture
线程池策略最好使用AbortPolicy
,然后耗时的异步线程,做好线程池隔离。