使用 CompletableFuture 异步组装数据

Java基础系列1:深入理解Java数据类型

使用 CompletableFuture 异步组装数据

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

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

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

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

CompletableFuture 简介

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

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

Linux系统实时数据同步inotify+rsync

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分批组装的方式,快捷、优雅。你们有好的方法呢?

负载均衡算法 – 基本实现

© 版权声明
THE END
喜欢就支持一下吧
点赞0 分享