CompletableFuture的类关系
对于 Future接口都比较熟悉了.主要介绍下CompletionStage接口.
CompletionStage接口代表异步计算过程中的某一个阶段,一个阶段完成后会进入另一个阶段.一个阶段可以理解为一个子任务,每个子任务会包装一个Java函数式接口实例,表示该子任务要执行的操作.
CompletionStage接口
顾名思义,Stage是阶段的意思.CompletionStage代表某个同步或者异步计算的一个阶段,或者是一系列异步任务中的一个子任务.
每个CompletionStage子任务所包装的可以是一个Function Consumer Runnable函数式接口.
1:Function
Function接口的特点,有输入有输出,包装了Function实例的CompletionStage的子任务需要输入一个参数,并会产生一个输出结果到下一步.
2:Runnable
Runnable接口的特点,是无输入无输出,包装了Runnable实例的CompletionStage的子任务既不需要任何输入也不会产生任何输出.
3:Consumer
Consumer接口的特点是,有输入无输出.包装了Consumer实例的CompletionStage的子任务需要输入一个参数,但不会产生任何输出.
runAsync和supplyAsync创建子任务
CompletionStage子任务的创建是通过CompletableFuture完成的.CompletableFuture定义了一组方法用于创建CompletionStage子任务.基础方法如下.
//包装一个Runnable实例,使用ForkJoinPool.commonPool()线程池来执行public static CompletableFuture<Void> runAsync(Runnable runnable)//包装一个Runnable实例,指定线程池执行.public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor) //包装一个Supplier实例,使用ForkJoinPool.commonPool()线程池来执行public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)//包装一个Supplier实例,指定线程池执行.public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor)实例如下
public class CompletableFutureTest {public static void main(String[] args) throws Exception {//创建一个无输入值无返回值的异步子任务.CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();System.out.println("运行结束");}});//等待异步任务执行完成.future.get(2, TimeUnit.SECONDS);//创建一个无输入有返回值的异步子任务.CompletableFuture<Long> future1 = CompletableFuture.supplyAsync(() -> {long start = System.currentTimeMillis();try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();System.out.println("运行结束");}return System.currentTimeMillis() - start;});//等待异步任务执行完成.Long aLong = future1.get(2, TimeUnit.SECONDS);System.out.println("异步执行耗时--" + aLong / 1000);}
}设置子任务的回调钩子
可以为CompletionStage子任务设置特定的回调钩子,当计算计算结果完成或者抛出异常的时候执行这些特定的回调钩子.
//设置子任务完成的回调钩子.
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)//设置子任务完成的回调钩子,可能不在同一个线程.
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action)//设置子任务完成的回调钩子,提交给线程池执行.
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor)实例如下
public class CompletableFutureTest2 {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("发生异常");throw new RuntimeException("我是异常..");});//异步执行完成的回调钩子.future.whenComplete(new BiConsumer<Void, Throwable>() {@Overridepublic void accept(Void unused, Throwable throwable) {System.out.println("执行完成.");}});//抛出异常回调钩子.future.exceptionally(new Function<Throwable, Void>() {@Overridepublic Void apply(Throwable throwable) {System.out.println("执行失败"+throwable.getMessage());return null;}});//异步获取结果.System.out.println(future.get());}
}调用Handle()方法统一处理异常和结果
除了可以通过回调函数处理以外,还可以通过handle方法统一处理结果和异常.
handle方法有三个重载版本.
//在执行任务的同一个线程处理异常和结果.
public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
//可能不在同一个线程处理异常和结果.
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
//在指定线程池处理异常和结果
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)案例如下
public class CompletableFutureTest3 {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("发生异常");throw new RuntimeException("我是异常..");});future.handle(new BiFunction<Void, Throwable, Object>() {@Overridepublic Object apply(Void unused, Throwable throwable) {if (null == throwable) {System.out.println("没有发生异常.");}else {System.out.println("发生了异常.");}return null;}});System.out.println(future.get());}
}