使用 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;
}
具体步骤如下:
- 将原始数据按照每组100条进行拆分。(具体每组拆分多少条需要根据实际的业务情况和服务器性能,多测试一下应该就知道了)
- 多线程组成数据,每个线程组装一组数据(上面拆分的100条原始数据)。packVOs 方法就是组装数据。为了高效,我建议 在组装数据的时候多采用批量,缓存的思想,能批量尽量批量,重复数据就尽量缓存下来。
- CompletableFuture.supplyAsync() 方法说明如下。第一个参数是线程需要执行的动作,第二个参数是线程执行用的Executor,可以填自定义的,也可以不填写,不填写程序会使用默认的执行器。
public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)
返回由给定执行程序中运行的任务异步完成的新CompletableFuture,其中包含通过调用给定供应商获得的值。 - whenCompleteAsync 方法含义和名字一样,将上一步执行的结果或者异常作为参数传给指定的参数。这里我们希望分批组装的结果能过add进result中。
- exceptionally 是用来处理异常。当一个线程执行出现异常的时候应该执行怎样的操作。
- all.join() 这个方法是等待所有的任务(所有的CompletableFuture)完成。组装数据是耗时的,如果我们不等待所有组装任务完成,直接返回result,相信result中不会有数据,或者数据是不完整的。我们期待的结果是所有的数据都正常组装完成,添加进result。
使用了CompletableFuture方式实现多线程分批组装,并且在组装时采用 “批量+缓存” 的思想,原来5000条数据30min缩短为3min。当然还有优化的空间,但是能达到这个效果已经让我非常满意了。
下次遇到类似的情况,我会优先考虑CompletableFuture分批组装的方式,快捷、优雅。你们有好的方法呢?
原文地址:https://www.cnblogs.com/catlkb/p/12250725.html