|
1背景2CompletableFuture的发展2.1Future简介2.2Future的局限性3关于CompletableFuture3.1CompletableFuture提供的能力3.2示例场景的解决方案3.3实际应用效果4总结5参考资料1背景转转门店商详页为用户在小程序、APP等终端提供下单、预约看机的能力。下图为转转门店小程序的商详页。起初业务体量小,QPS低,内容简单,接口采用简单的串行实现,但随着门店业务的发展,页面信息的丰富,QPS的提升,串行实现的缺点也显露出来:访问耗时高,代码耦合严重。假设门店商详页有以下几个模块,每个模块需要如下时间才能完成:获取商品基础信息0.5s获取优惠信息1s获取门店信息1s获取验机评估报告信息1s获取标品参数信息1s组装返回商品信息0.5s串行执行每个模块,那么一共需要5s才能返回给调用方,如果RPC接口产生超时等,会比5s还要长,显然是不能接受的。如果有多个线程并行完成各个模块,可能2s内就能返回信息。在JDK1.8中引入了CompletableFuture实现类,扩展了Future和CompletionStage,实现异步多线程任务执行。本文主要以获取门店商详页信息为例介绍CompletableFuture的常用方法和使用。2CompletableFuture的发展常见的线程创建方式有两种,一是直接继承Thread,另一种是实现Runnable接口。但这两种方式有个缺点,不支持获取线程执行结果,所以在JDK1.5之后,提供了Callable和Future,可以在任务执行后获取执行结果。2.1Future简介Future类位于java.util.concurrent包下,从下面的源码可以看出,Future主要提供了三种能力:关闭执行中的任务判断任务是否执行完成获取任务执行的结果package java.util.concurrent;public interface Future { // 取消执行中的任务 boolean cancel(boolean mayInterruptIfRunning); // 判断任务是否被取消成功 boolean isCancelled(); // 判断任务是否执行完成 boolean isDone(); // 获取任务执行结果 V get() throws InterruptedException, ExecutionException; // 在规定时间内获取任务执行结果,若规定时间任务还没执行完,则返回null,而非抛异常 V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;}2.2Future的局限性Future通过isDone()判断任务是否执行完成,get()获取任务执行的结果,解决了创建线程异步执行任务不能获取执行结果的问题,在任务异步执行中,主线程在等待过程中可以做其他事,但其本身也存在一定的局限性。并行执行多任务获取结果主线程长时间阻塞:当需要将多个模块的任务异步执行时,使用for循环遍历任务列表,通过isDone()轮询判断任务是否执行完成,通过get()方法获取任务执行结果,且结果获取依赖任务执行顺序。但因Future的get()方法是主线程阻塞等待获取执行结果,所以在结果返回前,主线程不能处理其他任务,长时间阻塞,可能会产生block,在使用时考虑用超时时间的get()方法。不能链式执行任务:如上述场景,希望在获取商品基础信息后执行获取优惠信息任务,Future没有提供这种能力,不能实现链式调用。不能将多个任务执行的结果组合:在上述场景中,希望获取商详所需的各个模块信息后,组合成调用方需要的结果,但Future不支持。不能处理异常:Future没有异常处理的能力。综上所述,阻塞主线程获取结果的方式与异步编程的初衷相违背,轮询判断任务是否执行完成会耗费不必要的CPU资源,为优化上述问题,在JDK1.8时引入了CompletableFuture实现类,提供异步链式编程的能力。3关于CompletableFutureCompletableFuture扩展了Future接口,实现了CompletionStage,提供了函数式编程的能力,通过回调的方式处理计算结果,并提供了转换和组合CompletableFuture的方法,从而简化异步编程的复杂性。下图为CompletableFuture类图。3.1CompletableFuture提供的能力CompletableFuture依然有Future的能力,可以通过get()获取结果,但不推荐使用,那么CompletableFuture提供了哪些能力呢?3.1.1创建异步任务CompletableFuture提供了四种创建异步对象的方法,如下:public static CompletableFuture runAsync(Runnable runnable)public static CompletableFuture runAsync(Runnable runnable, Executor executor)public static CompletableFuture supplyAsync(Supplier supplier)public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)其中runAsync表示无返回值的异步对象,supplyAsync为有返回值的异步对象,仍然可以通过如果不指定线程池会默认使用ForkJoinPool.commonPool(),建议使用自定义的线程池,和其他场景区分开,避免线程资源竞争,也易于监控线程使用情况。在上述获取商品页信息的示例中,想获取验机评估报告信息,可以使用无返回值的runAsync创建对象,使用自定义的线程池testThreadPool,结果集使用自定义的对象ResultDTO接收,里面定义各个模块结果的成员变量,如果使用Map接收结果,需要注意线程安全问题。 ExecutorService testThreadPool = Executors.newFixedThreadPool(10); ResultDTO resultDTO = new ResultDTO(); CompletableFuture baseInfoFuture = CompletableFuture.runAsync(() -> { System.out.println("获取商品验机评估报告任务当前线程" + Thread.currentThread().getId()); //获取报告rpc接口逻辑 resultMap.put("baseInfo", ""); }, testThreadPool).exceptionally(e -> { System.out.println(e.getMessage()); return null; });3.1.2异步回调CompletableFuture可以在任务处理结束异步回调,主要提供了以下回调方法://当上一个任务执行结束时,回调方法接收上一段任务执行结果或异常,返回上一段任务结果CompletableFuture whenComplete(BiConsumer action) //当上一个任务执行结束时,回调方法接收上一段任务执行结果或异常,返回回调方法的结果 CompletableFuture handle(BiFunction fn)//异步回调CompletableFuture whenCompleteAsync(BiConsumer action)//当上一个任务执行出现异常时,返回回调中的结果CompletableFuture exceptionally(Function fn)3.1.3异步编排CompletionStage接口定义了异步编排的各种场景,CompletableFuture实现了该接口,每个CompletionStage表示异步编排的某个阶段,任务执行可能在上一个阶段完成时触发,也可能后面几个阶段全部完成或某个阶段完成时触发。原理大致是每个新建的CompletableFuture对象,会将当前方法需要执行的任务封装成Completion对象,将其压入到上个方法中创建的异步任务栈中,Completion为一个链表结构,保存依赖的任务,CAS出链表后,tryFire()执行依赖任务,从而实现任务串联。比较常用的场景有在上一阶段完成时触发的串行场景,多个阶段都执行完成触发的AND,或者多个阶段中某一阶段执行完触发的OR。下面为方法示例。//异步接收上一阶段任务结果,且有返回值 CompletableFuture thenApplyAsync(Function fn) //异步接收上一阶段任务结果,但无返回值CompletableFuture thenAcceptAsync(Consumer action, Executor executor)//等待全部异步任务执行结束static CompletableFuture allOf(CompletableFuture... cfs)//某个任务执行结束即返回static CompletableFuture anyOf(CompletableFuture... cfs)在上述获取商详单示例中,可以使用thenAcceptAsync在获取商品基础信息后异步获取优惠信息,如下代码。 CompletableFuture productBaseInfoFuture = CompletableFuture.runAsync(() -> {  roductInfoDTO base = new roductInfoDTO(); BaseInfoDTO baseInfoDTO = rpcxx; resultDTO.setBaseInfoDTO(baseInfoDTO); }, testThreadPool).thenAcceptAsync(() -> { CouponInfoDTO couponInfoDTO = rpcxx; resultDTO.setCouponInfoDTO(couponInfoDTO); }, testThreadPool);返回结果:3.2示例场景的解决方案通过上述分析,可以将获取商品页的步骤分为三步,分别为:获取商品基础信息、商品验机评估报告信息、商品标品参数信息、门店信息;获取商品优惠信息,需要等1结束;将上述信息RPC结果组装返回,需要等1,2结束。因为多任务异步并行执行,最终耗时将取决于耗时最长的链路。如下图所示:代码示例: ExecutorService testThreadPool = Executors.newFixedThreadPool(10); ResultDTO resultDTO = new ResultDTO(); //基础信息 CompletableFuture productBaseInfoFuture = CompletableFuture.runAsync(() -> { BaseInfoDTO baseInfoDTO = rpcxx; resultDTO.setBaseInfoDTO(baseInfoDTO); }, testThreadPool); //优惠信息 CompletableFuture couponInfoFuture = productBaseInfoFuture.thenAcceptAsync(() -> { CouponInfoDTO couponInfoDTO = rpcxx; resultDTO.setCouponInfoDTO(couponInfoDTO); }, testThreadPool); //验机评估报告信息 CompletableFuture qcInfoFuture = CompletableFuture.runAsync(() -> { QcInfoDTO qcInfoDTO = rpcxx; resultDTO.setQcInfoDTO(qcInfoDTO); }, testThreadPool); //门店信息 CompletableFuture storeInfoFuture = CompletableFuture.runAsync(() -> { StoreInfoDTO storeInfoDTO = rpcxx; resultDTO.setStoreInfoDTO(storeInfoDTO); }, testThreadPool); //标品参数信息 CompletableFuture spuInfoFuture = CompletableFuture.runAsync(() -> { SpuInfoDTO spuInfoDTO = rpcxx; resultDTO.setSpuInfoDTO(spuInfoDTO); }, testThreadPool); //组装结果 CompletableFuture allQuery = CompletableFuture.allOf(couponInfoFuture, qcInfoFuture, storeInfoFuture, spuInfoFuture); CompletableFuture buildFuture = allQuery.thenAcceptAsync((result) -> { //组装逻辑 return null; }).join();以上即为获取门店商品页异步编排的实现逻辑,但也发现,该方案创建多个异步任务,执行逻辑不一样但流程大致相同,类似的代码重复写多遍,不便于扩展和阅读。在此基础上可以优化为使用CompletableFuture+简单工厂+策略模式,将上述步骤中的每个模块都作为策略handler,且策略之间有权重依赖关系,模块类型作为工厂类型,将模块类型放进列表中,使用CompletableFuture.allOf()异步执行列表中的任务。伪代码如下: List eventList = Arrays.asList("xx", "xxx"); CompletableFuture.allOf(eventList.stream().map(event -> CompletableFuture.runAsync(() -> { //通过工厂类型获取策略实现handler if (Objects.nonNull(handler)) { //如果存在则执行 } }, testThreadPool)).toArray(CompletableFuture[]::new)).join();3.3实际应用效果最终转转门店商详页接口重构采用CompletableFuture+简单工厂+策略模式的方案,解耦堆叠的代码块,使功能更易扩展,可阅读性更高,在重构代码上线后,效果明显。4总结以上就是关于使用CompletableFuture进行异步编程的全部内容,在此基础上我们还要注意以下几点:当有多个任务可以异步并行执行时,使用CompletableFuture,任务越多效果越明显;使用CompletableFuture可以将多个任务串联执行,也可以利用组合方式将任务排列由列表变成树结构;在使用集合接收多线程处理任务的结果时,需要考虑线程安全问题;当任务执行有相互依赖关系时,需考虑任务超时主动结束,避免系统block。5参考资料https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.htmlhttps://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html关于作者郑晴,转转门店技术部后端研发工程师
|
|