转转门店商详页异步编程的转转实践

作者：郑晴 2023-07-27 07:00:01开发 转转门店商详页为用户在小程序、APP等终端提供下单、门店预约看机的商详实践能力。

• 1 背景
• 2 CompletableFuture的页异发展

• 2.1 Future简介
• 2.2 Future的局限性

• 3 关于CompletableFuture
• 3.1 CompletableFuture提供的能力
• 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的常用方法和使用。

2 CompletableFuture的发展

常见的线程创建方式有两种，一是直接继承Thread，另一种是实现Runnable接口。但这两种方式有个缺点，不支持获取线程执行结果，所以在JDK1.5之后，提供了Callable和Future，可以在任务执行后获取执行结果。

2.1 Future简介

Future类位于java.util.concurrent包下，从下面的源码可以看出，Future主要提供了三种能力：

• 关闭执行中的任务
• 判断任务是否执行完成
• 获取任务执行的结果

package java.util.concurrent;public interface Future<V> {     // 取消执行中的任务    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.2 Future的局限性

Future通过isDone()判断任务是否执行完成，get()获取任务执行的结果，解决了创建线程异步执行任务不能获取执行结果的问题，在任务异步执行中，主线程在等待过程中可以做其他事，但其本身也存在一定的局限性。

• 并行执行多任务获取结果主线程长时间阻塞：当需要将多个模块的任务异步执行时，使用for循环遍历任务列表，通过isDone()轮询判断任务是否执行完成，通过get()方法获取任务执行结果，且结果获取依赖任务执行顺序。但因Future的get()方法是主线程阻塞等待获取执行结果，所以在结果返回前，主线程不能处理其他任务，长时间阻塞，可能会产生block，在使用时考虑用超时时间的get()方法。
• 不能链式执行任务：如上述场景，希望在获取商品基础信息后执行获取优惠信息任务，Future没有提供这种能力，不能实现链式调用。
• 不能将多个任务执行的结果组合：在上述场景中，希望获取商详所需的各个模块信息后，组合成调用方需要的结果，但Future不支持。
• 不能处理异常：Future没有异常处理的能力。

综上所述，阻塞主线程获取结果的方式与异步编程的初衷相违背，轮询判断任务是否执行完成会耗费不必要的CPU资源，为优化上述问题，在JDK1.8时引入了CompletableFuture实现类，提供异步链式编程的能力。

3 关于CompletableFuture

CompletableFuture扩展了Future接口，实现了CompletionStage，提供了函数式编程的能力，通过回调的方式处理计算结果，并提供了转换和组合CompletableFuture的方法，从而简化异步编程的复杂性。下图为CompletableFuture类图。

图片

3.1 CompletableFuture提供的能力

CompletableFuture依然有Future的能力，可以通过get()获取结果，但不推荐使用，那么CompletableFuture提供了哪些能力呢？

3.1.1 创建异步任务

CompletableFuture提供了四种创建异步对象的方法，如下：

public static CompletableFuture<Void>  runAsync(Runnable runnable)public static CompletableFuture<Void>  runAsync(Runnable runnable, Executor executor)public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier)public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier, Executor executor)

其中runAsync表示无返回值的异步对象，supplyAsync为有返回值的异步对象，仍然可以通过如果不指定线程池会默认使用ForkJoinPool.commonPool()，建议使用自定义的线程池，和其他场景区分开，避免线程资源竞争，也易于监控线程使用情况。

在上述获取商品详情页信息的示例中，想获取验机评估报告信息，可以使用无返回值的runAsync创建对象，使用自定义的线程池testThreadPool，结果集使用自定义的对象ResultDTO接收，里面定义各个模块结果的成员变量，如果使用Map接收结果，需要注意线程安全问题。

ExecutorService testThreadPool = Executors.newFixedThreadPool(10);    ResultDTO resultDTO = new ResultDTO();    CompletableFuture<Void> 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<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) //当上一个任务执行结束时，回调方法接收上一段任务执行结果或异常，返回回调方法的结果<U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)//异步回调CompletableFuture<T>  whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)//当上一个任务执行出现异常时，返回回调中的结果CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)

3.1.3 异步编排

CompletionStage接口定义了异步编排的各种场景，CompletableFuture实现了该接口，每个CompletionStage表示异步编排的某个阶段，任务执行可能在上一个阶段完成时触发，也可能后面几个阶段全部完成或某个阶段完成时触发。

原理大致是每个新建的CompletableFuture对象，会将当前方法需要执行的任务封装成Completion对象，将其压入到上个方法中创建的异步任务栈中，Completion为一个链表结构，保存依赖的任务，CAS出链表后，tryFire()执行依赖任务，从而实现任务串联。

比较常用的场景有在上一阶段完成时触发的串行场景，多个阶段都执行完成触发的AND，或者多个阶段中某一阶段执行完触发的OR。下面为方法示例。

//异步接收上一阶段任务结果，且有返回值<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) //异步接收上一阶段任务结果，但无返回值CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)//等待全部异步任务执行结束static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)//某个任务执行结束即返回static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

在上述获取商详单示例中，可以使用thenAcceptAsync在获取商品基础信息后异步获取优惠信息，如下代码。

CompletableFuture<Void> productBaseInfoFuture = CompletableFuture.runAsync(() -> {         ProductInfoDTO base = new ProductInfoDTO();        BaseInfoDTO baseInfoDTO = rpcxx;        resultDTO.setBaseInfoDTO(baseInfoDTO);    }, testThreadPool).thenAcceptAsync(() -> {         CouponInfoDTO couponInfoDTO = rpcxx;        resultDTO.setCouponInfoDTO(couponInfoDTO);    }, testThreadPool);

返回结果：

图片

3.2 示例场景的解决方案

通过上述分析，可以将获取商品详情页的步骤分为三步，分别为：

1. 获取商品基础信息、商品验机评估报告信息、商品标品参数信息、门店信息；
2. 获取商品优惠信息，需要等1结束；
3. 将上述信息RPC结果组装返回，需要等1，2结束。

因为多任务异步并行执行，最终耗时将取决于耗时最长的链路。如下图所示：

图片

代码示例：

ExecutorService testThreadPool = Executors.newFixedThreadPool(10);    ResultDTO resultDTO = new ResultDTO();    //基础信息    CompletableFuture<Void> productBaseInfoFuture = CompletableFuture.runAsync(() -> {         BaseInfoDTO baseInfoDTO = rpcxx;        resultDTO.setBaseInfoDTO(baseInfoDTO);    }, testThreadPool);    //优惠信息    CompletableFuture<Void> couponInfoFuture = productBaseInfoFuture.thenAcceptAsync(() -> {         CouponInfoDTO couponInfoDTO = rpcxx;        resultDTO.setCouponInfoDTO(couponInfoDTO);    }, testThreadPool);    //验机评估报告信息    CompletableFuture<Void> qcInfoFuture = CompletableFuture.runAsync(() -> {         QcInfoDTO qcInfoDTO = rpcxx;        resultDTO.setQcInfoDTO(qcInfoDTO);    }, testThreadPool);    //门店信息    CompletableFuture<Void> storeInfoFuture = CompletableFuture.runAsync(() -> {         StoreInfoDTO storeInfoDTO = rpcxx;        resultDTO.setStoreInfoDTO(storeInfoDTO);    }, testThreadPool);    //标品参数信息    CompletableFuture<Void> spuInfoFuture = CompletableFuture.runAsync(() -> {         SpuInfoDTO spuInfoDTO = rpcxx;        resultDTO.setSpuInfoDTO(spuInfoDTO);    }, testThreadPool);    //组装结果    CompletableFuture<Void> allQuery = CompletableFuture.allOf(couponInfoFuture, qcInfoFuture, storeInfoFuture, spuInfoFuture);    CompletableFuture<Void> buildFuture = allQuery.thenAcceptAsync((result) -> {         //组装逻辑        return null;    }).join();

以上即为获取门店商品详情页异步编排的实现逻辑，但也发现，该方案创建多个异步任务，执行逻辑不一样但流程大致相同，类似的代码重复写多遍，不便于扩展和阅读。

在此基础上可以优化为使用CompletableFuture+简单工厂+策略模式，将上述步骤中的每个模块都作为策略handler，且策略之间有权重依赖关系，模块类型作为工厂类型，将模块类型放进列表中，使用CompletableFuture.allOf()异步执行列表中的任务。

伪代码如下：

List<String> 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.html
• https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html

责任编辑：庞桂玉 来源： 转转技术 转转门店商编程

(责任编辑：知识)