使用 CompletableFuture 异步组装数据

使用 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分批组装的方式，快捷、优雅。大家有好的方法呢？