【Java异步编程】CompletableFuture基础（1）：创建不同线程的子任务、子任务链式调用与异常处理

文章目录1. 三种实现接口2. 链式调用保证链的顺序性与异步性3. CompletableFuture创建CompletionStage子任务4. 处理异常a. 创建回调钩子b. 调用handle()方法统一处理异常和结果5. 如何选择线程池不同的业务选择不同的线程池CompletableFuture是JDK 1.8引入的实现类该类实现了Future和CompletionStage两个接口。该类的实例作为一个异步任务可以在自己异步执行完成之后触发一些其他的异步任务从而达到异步回调的效果。CompletionStage代表异步计算过程中的某一个阶段一个阶段完成以后可能会进入另一个阶段。一个阶段可以理解为一个子任务每一个子任务会包装一个Java函数式接口实例表示该子任务所要执行的操作。1. 三种实现接口每个CompletionStage子任务所包装的可以是一个Function、Consumer或者Runnable函数式接口实例。这三个常用的函数式接口的特点如下被包装接口功能描述FunctionFunction接口的特点是有输入、有输出。包装了Function实例的CompletionStage子任务需要一个输入参数并会产生一个输出结果到下一步。RunnableRunnable接口的特点是无输入、无输出。包装了Runnable实例的CompletionStage子任务既不需要任何输入参数又不会产生任何输出。ConsumerConsumer接口的特点是有输入、无输出。包装了Consumer实例的CompletionStage子任务需要一个输入参数但不会产生任何输出。2. 链式调用保证链的顺序性与异步性多个CompletionStage构成了一条任务流水线一个环节执行完成了可以将结果移交给下一个环节子任务​。多个CompletionStage子任务之间可以使用链式调用。下面是一个顺序调用的例子使用CompletionStage及其方法构建了一个异步任务链thenApply用于对前一个阶段的结果进行计算并传递结果thenAccept用于消费前一个阶段的结果并执行操作thenRun用于执行无输入输出的操作。oneStage//被thenApply包装CompletionStage子任务,由输入输出.thenApply(x-square(x))//消耗上游输出但是没有输出.thenAccept(y-System.out.println(y))//不消耗上一个子任务的输出又不产生结果.thenRun(()-System.out.println())这种链式操作可以方便地将多个异步操作连接起来同时保证了操作的顺序性和异步性提高了代码的可维护性和并发性能。接下来是一个异步调用的例子在这个例子中task2 和 task3 都依赖于 task1 完成后执行但它们可能并行执行也就是说task2 和 task3 的执行顺序是不确定的它们不一定会按照 thenRunAsync 的顺序执行。CompletableFutureVoidtask1CompletableFuture.runAsync(()-{System.out.println(Task 1);});CompletableFutureVoidtask2task1.thenRunAsync(()-{System.out.println(Task 2);});CompletableFutureVoidtask3task1.thenRunAsync(()-{System.out.println(Task 3);});3. CompletableFuture创建CompletionStage子任务CompletableFuture定义了一组方法用于创建CompletionStage子任务或者阶段性任务​基础的方法如下//子任务包装一个Runnable实例并调用ForkJoinPool.commonPool()线程池来执行publicstaticCompletableFutureVoidrunAsync(Runnablerunnable)//子任务包装一个Runnable实例并调用指定的executor线程池来执行publicstaticCompletableFutureVoidrunAsync(Runnablerunnable,Executorexecutor)//子任务包装一个Supplier实例并调用ForkJoinPool.commonPool()线程池来执行publicstaticUCompletableFutureUsupplyAsync(SupplierUsupplier)//子任务包装一个Supplier实例并使用指定的executor线程池来执行publicstaticUCompletableFutureUsupplyAsync(SupplierUsupplier,Executorexecutor)其中主要注意的信息是Supplier 表示一个无参数但有返回值的函数Runnable表示无惨无返回值的函数在使用CompletableFuture创建CompletionStage子任务时如果没有指定Executor线程池在默认情况下CompletionStage会使用公共的ForkJoinPool线程池。它们都会交给线程池执行get方法会堵塞主线程等待执行结果。给一个例子//无返回值异步调用TestpublicvoidrunAsyncDemo()throwsException{CompletableFutureVoidfutureCompletableFuture.runAsync(()-{sleepSeconds(1);//模拟执行1秒Print.tco(run end ...);});//等待异步任务执行完成,最多等待2秒future.get(2,TimeUnit.SECONDS);}//有返回值异步调用TestpublicvoidsupplyAsyncDemo()throwsException{CompletableFutureLongfutureCompletableFuture.supplyAsync(()-{longstartSystem.currentTimeMillis();sleepSeconds(1);//模拟执行1秒Print.tco(run end ...);returnSystem.currentTimeMillis()-start;});//等待异步任务执行完成,现时等待2秒longtimefuture.get(2,TimeUnit.SECONDS);Print.tco(异步执行耗时秒 time/1000);}4. 处理异常a. 创建回调钩子可以为CompletionStage子任务设置特定的回调钩子当计算结果完成或者抛出异常的时候执行这些特定的回调钩子。//设置子任务完成时的回调钩子publicCompletableFutureTwhenComplete(BiConsumer?superT,?superThrowableaction)//设置子任务完成时的回调钩子可能不在同一线程执行publicCompletableFutureTwhenCompleteAsync(BiConsumer?superT,?superThrowableaction)//设置子任务完成时的回调钩子提交给线程池executor执行publicCompletableFutureTwhenCompleteAsync(BiConsumer?superT,?superThrowableaction,Executorexecutor)//设置异常处理的回调钩子publicCompletableFutureTexceptionally(FunctionThrowable,?extendsTfn)TestpublicvoidwhenCompleteDemo()throwsException{CompletableFutureVoidfutureCompletableFuture.runAsync(()-{sleepSeconds(1);//模拟执行1秒Print.tco(抛出异常);thrownewRuntimeException(发生异常);//Print.tco(run end ...);});//设置执行完成后的回调钩子future.whenComplete(newBiConsumerVoid,Throwable(){Overridepublicvoidaccept(Voidt,Throwableaction){Print.tco(执行完成);}});//设置发生异常后的回调钩子future.exceptionally(newFunctionThrowable,Void(){OverridepublicVoidapply(Throwablet){Print.tco(执行失败t.getMessage());returnnull;}});future.get();}[ForkJoinPool.commonPool-worker-9]抛出异常[main]执行完成[ForkJoinPool.commonPool-worker-9]执行失败java.lang.RuntimeException:发生异常java.util.concurrent.ExecutionException:java.lang.RuntimeException:发生异常有如下几个注意点调用cancel()方法取消CompletableFuture时任务被视为异常完成completeExceptionally()方法所设置的异常回调钩子也会被执行。如果没有设置异常回调钩子发生内部异常时会有两种情况发生在调用get()时如果遇到内部异常get()方法就会抛出ExecutionException执行异常​。在调用join()和getNow(T)启动任务时如果遇到内部异常join()和getNow(T)方法就会抛出CompletionException。b. 调用handle()方法统一处理异常和结果//在执行任务的同一个线程中处理异常和结果publicUCompletionStageUhandle(BiFunction?superT,Throwable,?extendsUfn);//可能不在执行任务的**同一个线程**中处理异常和结果publicUCompletionStageUhandleAsync(BiFunction?superT,Throwable,?extendsUfn);//在指定线程池executor中处理异常和结果publicUCompletionStageUhandleAsync(BiFunction?superT,Throwable,?extendsUfn,Executorexecutor);TestpublicvoidhandleDemo()throwsException{CompletableFutureVoidfutureCompletableFuture.runAsync(()-{sleepSeconds(1);//模拟执行1秒Print.tco(抛出异常);thrownewRuntimeException(发生异常);//Print.tco(run end ...);});//设置执行完成后的回调钩子future.handle(newBiFunctionVoid,Throwable,Void(){OverridepublicVoidapply(Voidinput,Throwablethrowable){if(throwablenull){Print.tcfo(没有发生异常);}else{Print.tcfo(sorry,发生了异常);}returnnull;}});future.get();}////[ForkJoinPool.commonPool-worker-1]抛出异常//[ForkJoinPool.commonPool-worker-1|CompletableFutureDemo$3.apply]: sorry,发生了异常5. 如何选择线程池不同的业务选择不同的线程池默认情况下通过静态方法runAsync()、supplyAsync()创建的CompletableFuture任务会使用公共的ForkJoinPool线程池默认的线程数是CPU的核数。当然它的线程数可以通过以下JVM参数设置option:-Djava.util.concurrent.ForkJoinPool.common.parallelism如果所有CompletableFuture共享一个线程池那么一旦有任务执行一些很慢的IO操作就会导致线程池中的所有线程都阻塞在IO操作上造成线程饥饿进而影响整个系统的性能。所以强烈建议大家根据不同的业务类型创建不同的线程池以避免互相干扰。