使用 CompletableFuture 异步组装数据

Posted 2021-03-12 catlkb

使用 CompletableFuture 异步组装数据

一种快捷、优雅的异步组装数据方式

实际项目中经常遇到这种情况： 从多个表中查找到数据然后拼装成一个VO返回给前端。
这个过程有可能会非常耗时。因为最终每一条返回的VO数据是由多个表中的数据拼装而成，如果项目还是微服务需要从其他服务获取数据，那将会更加耗时，更加麻烦。简单的几十条、几百条数据单个线程跑起来可能没有什么压力，但是当数量达到成千上万，几十万，几百万，组装的逻辑也变得非常复杂时，这个操作就非常耗时。

最近我在项目中就遇到这个的情况。项目中我们需要做一个相关流程数据的下载功能。
最初版本使用单线程，因为业务的复杂性，5000多条数据完全下载下来需要30min。以为是从数据库分拣数据比较耗时，查询日志后发现数据库查询并没有耗时多久，反而是组装数据占用了大多数时间。

因此机智的我就想起之前同组小伙伴分享的Java8一个新的类CompletableFuture。

CompletableFuture 简介

CompletableFuture 是Java 8 新增加的Api,该类实现，Future和CompletionStage两个接口，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

具体大家可以查看Java Api 文档，或者阅读网上一些博客。

CompletableFuture 异步组装数据

代码示例如下

/**
     * 功能描述: 拼装数据
     * @author lkb
     * @date 2019/12/25
     * @param
     * @return java.util.List<com.laidian.erp.crm.vo.DeviceProcessListExportVO>
     */
    private List<DeviceProcessListExportVO> listByFlowJobIds(List<String> flowJobIds, Map<String, ProcessInfoVo> map, Map<Integer,UserInfoDTO> userInfoDTOMap, Map<Integer,HatCity> cityMap){
        //result 列表保存组装完成的数据
        List<DeviceProcessListExportVO> result = new LinkedList<>();
        //每次组装100条数据
        List<List<String>> partition = Lists.partition(flowJobIds,100);
        List<CompletableFuture> futures = partition.stream().map(subList -> CompletableFuture.supplyAsync(() -> {
            //packVOs 方法就是组装数据
            return packVOs(subList,map,userInfoDTOMap,cityMap);
        },ASYNC_IO_POOL).whenCompleteAsync((r,e)->result.addAll(r))
                        .exceptionally(e->{
                            log.error(e.getMessage(),e);
                            log.error("listByFlowJobIds error.");
                            return result;
                        })).collect(Collectors.toList());

        CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
        log.info("任务阻塞 ");
        Instant start = Instant.now();
        //阻塞，直到所有任务结束。
        all.join();
        log.info("任务阻塞结束 耗时 = {}",ChronoUnit.MILLIS.between(start, Instant.now()));
        return result;
    }

具体步骤如下：

1. 将原始数据按照每组100条进行拆分。（具体每组拆分多少条需要根据实际的业务情况和服务器性能，多测试一下应该就知道了）
2. 多线程组成数据，每个线程组装一组数据（上面拆分的100条原始数据）。packVOs 方法就是组装数据。为了高效，我建议 在组装数据的时候多采用批量，缓存的思想，能批量尽量批量，重复数据就尽量缓存下来。

3. CompletableFuture.supplyAsync() 方法说明如下。第一个参数是线程需要执行的动作，第二个参数是线程执行用的Executor，可以填自定义的，也可以不填写，不填写程序会使用默认的执行器。

   public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)
   返回由给定执行程序中运行的任务异步完成的新CompletableFuture，其中包含通过调用给定供应商获得的值。

4. whenCompleteAsync 方法含义和名字一样，将上一步执行的结果或者异常作为参数传给指定的参数。这里我们希望分批组装的结果能过add进result中。
5. exceptionally 是用来处理异常。当一个线程执行出现异常的时候应该执行怎样的操作。

6. all.join() 这个方法是等待所有的任务（所有的CompletableFuture）完成。组装数据是耗时的，如果我们不等待所有组装任务完成，直接返回result，相信result中不会有数据，或者数据是不完整的。我们期待的结果是所有的数据都正常组装完成，添加进result。

使用了CompletableFuture方式实现多线程分批组装，并且在组装时采用 “批量+缓存” 的思想，原来5000条数据30min缩短为3min。当然还有优化的空间，但是能达到这个效果已经让我非常满意了。

下次遇到类似的情况，我会优先考虑CompletableFuture分批组装的方式，快捷、优雅。你们有好的方法呢？