tips：解释下主线程等待，子线程执行，如果子线程发生异常会被捕捉，主线程也会发生异常而终止，不会一直阻塞下去
ExecutorService executorService= Executors.newSingleThreadExecutor();
Future cf = executorService.submit(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(false){//改成ture后就抛异常了，cf.get()也不会一直阻塞主线程
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
return 1.2;
}
});
System.out.println("run result->"+cf.get());
supplyAsync异步带有返回值
CompletableFuture completableFuture= CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
return "我是返回值";
});
System.out.println("completableFuture result->"+completableFuture.get());
runAsync异步没有返回值
CompletableFuture cf = CompletableFuture.runAsync(()->{
System.out.println(Thread.currentThread()+" start,time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
if(false){
throw new RuntimeException("test");
}else{
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
}
});
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任务执行完成
System.out.println("run result->"+cf.get());
CompletableFuture里面执行的其实还是Executor的实现类，默认执行的是ForkJoinPool
源码如下：
public static CompletableFuture supplyAsync(Supplier supplier,
Executor executor) {
return asyncSupplyStage(screenExecutor(executor), supplier);
}
static Executor screenExecutor(Executor e) {
if (!useCommonPool && e == ForkJoinPool.commonPool())//这里是ForfJoinPool
return asyncPool;
if (e == null) throw new NullPointerException();
return e;
}
如果是机器是单核的，就会创建一个ThreadPerTaskExecutor的执行器：
源码如下：
private static final boolean useCommonPool =
(ForkJoinPool.getCommonPoolParallelism() > 1);//判断单核多核返回boolean值
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool()
: new ThreadPerTaskExecutor();//单核就new一个执行器
ThreadPerTaskExecutor 实现Executor接口并且创建一个单线程
static final class ThreadPerTaskExecutor implements Executor {
public void execute(Runnable r) { new Thread(r).start(); }
}
基于上述源码可以看到CompletableFuture应该有两个重载Executor的方法
于是乎有如下静态构造方法：
public static CompletableFuture supplyAsync(Supplier supplier,
Executor executor) {
return asyncSupplyStage(screenExecutor(executor), supplier);
}
/111111111111111111111111111
ForkJoinPool forkJoinPool=new ForkJoinPool();
CompletableFuture completableFuture=CompletableFuture.supplyAsync(()->{
//do yourself
return xxxxx;
},forkJoinPool);
System.out.println("completableFuture result->"+completableFuture.get());
/22222222222
ExecutorService executorService= Executors.newSingleThreadExecutor();
// 创建异步执行任务:
CompletableFuture cf = CompletableFuture.runAsync(()->{
//do yourself
},executorService);
System.out.println("main thread start,time->"+System.currentTimeMillis());
//等待子任务执行完成
System.out.println("run result->"+cf.get());
异步回调
1、thenApply:表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中
2.thenApplyAsync
ForkJoinPool pool=new ForkJoinPool();
// 创建异步执行任务:
CompletableFuture cf = CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread()+" 任务一,开始time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" 任务一,结束time->"+System.currentTimeMillis());
return "123456789";
},pool);
Object o = cf.get();
System.out.println(o);
CompletableFuture completableFuture = cf.thenApplyAsync((result)->{
System.out.println(Thread.currentThread()+" 任务二,开始time->"+System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread()+" 任务二,结束time->"+System.currentTimeMillis());
return "456789123"+result;
});
System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
System.out.println(completableFuture.get());
pool.shutdown();
thenAccept:接收上一个任务的执行结果作为入参，但是没有返回值
thenRun:没入参也没返回值
ForkJoinPool pool=new ForkJoinPool();
// 创建异步执行任务:
CompletableFuture cf = CompletableFuture.supplyAsync(()->{
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "123";
},pool);
CompletableFuture completableFuture = cf.thenApply((result)->{
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "456";
}).thenAccept((result)->{//接收上一个任务的执行结果作为入参，但是没有返回值
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("接收："+result);
}).thenRun(()->{//没入参也没返回值
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("啥也不干");
});
System.out.println(":::::::"+cf.get());
System.out.println("等待最后一个thenRun执行完成completableFuture才输出："+completableFuture.get());
exceptionally方法指定某个任务执行异常时执行的回调方法，
会将抛出异常作为参数传递到回调方法中，
如果该任务正常执行,exceptionally
方法返回的CompletionStage的result就是该任务正常执行的结果( CompletableFuture implements Future, CompletionStage)
ForkJoinPool pool=new ForkJoinPool();
// 创建异步执行任务:
CompletableFuture cf = CompletableFuture.supplyAsync(()->{
System.out.println("任务一");
if(true){
throw new RuntimeException("test");
}else{
return 1.23;
}
},pool);
//cf执行异常时，将抛出的异常作为入参传递给回调方法
CompletableFuture cf2= cf.exceptionally((param)->{
System.out.println("任务二");
System.out.println("error stack trace->");
param.printStackTrace();
System.out.println(Thread.currentThread()+" exit,time->"+System.currentTimeMillis());
return -1.0;
});
//cf正常执行时执行的逻辑，如果执行异常则不调用此逻辑
CompletableFuture cf3=cf.thenAccept((param)->{
System.out.println("任务三");
System.out.println("param->"+param);
});
//等待子任务执行完成,此处无论是任务二和任务三都可以实现job2退出，主线程才退出
// 如果是cf，则主线程不会等待任务二执行完成自动退出了
//cf2.get时，没有异常，但是依然有返回值，就是cf的返回值
System.out.println("》》》》》》》》》》->"+cf2.get());