如何对网站的图片做cdn,淘客做的领券网站,开发网站用什么软件,点击器文章目录 一次业务的批量数据任务的处理优化业务背景1.0版本 分批处理模式2.0版本 平衡任务队列模式3.0版本 优化调度平衡任务队列模式总结 一次业务的批量数据任务的处理优化
业务背景
一个重新生成所有客户的财务业务指标数据的批量数据处理任务。
1.0版本 分批处理模式
… 文章目录 一次业务的批量数据任务的处理优化业务背景1.0版本 分批处理模式2.0版本 平衡任务队列模式3.0版本 优化调度平衡任务队列模式总结 一次业务的批量数据任务的处理优化
业务背景
一个重新生成所有客户的财务业务指标数据的批量数据处理任务。
1.0版本 分批处理模式
根据要处理的客户数量按照最大线程数切分成多个段尽量保证每个线程处理相同的客户数量。 private void updateForRegenerateByCustomer(ListInteger customerIdList,SystemUserCommonDTO user, LocalDateTime now) {ListCustomerBaseInfo baseInfoList CollectionUtils.isEmpty(customerIdList)?customerInfoService.listAll():customerInfoService.listByIdList(customerIdList);//先清理客户的数据updateForCleanByCustomerIdList(baseInfoList,user,now);int maxSize baseInfoList.size();//计算当前任务数量int currentMaxPoolSize maxPoolSizemaxSize?maxPoolSize:maxSize;CompletableFuture[] tasks new CompletableFuture[currentMaxPoolSize];//计算每个任务分段的数量int size maxSize / currentMaxPoolSize;for(int i0;icurrentMaxPoolSize;i){final int begin i * size;final int end icurrentMaxPoolSize-1?maxSize:(i1)*size;//创建异步处理的分段任务tasks[i] CompletableFuture.runAsync(()-updateForGenerateByCustomerIdList(baseInfoList,begin,end,user,now),executorService).whenCompleteAsync((k,v)- log.info(重新生成财务业务指标客户的所有数据-线程【{}】完成,Thread.currentThread().getName()));}// 向线程池提交任务CompletableFuture.allOf(tasks).whenComplete((v, th) - log.info(重新生成财务业务指标客户的所有数据-【{}】个子线程处理完成,tasks.length)).join();}/*** 生成指定客户列表的所有数据**/private void updateForGenerateByCustomerIdList(ListCustomerBaseInfo baseInfoList,int begin,int end,SystemUserCommonDTO user, LocalDateTime now){//每个线程只处理自己的分段的数据for(int ibegin;iend;i){CustomerBaseInfo baseInfo baseInfoList.get(i);//每个客户独立事务TransactionalUtils.runWithNewTransactional(()-updateForGenerateByCustomerId(baseInfo.getId(),user,now));}}/*** 生成指定客户的所有数据**/private void updateForGenerateByCustomerId(Integer customerId,SystemUserCommonDTO user, LocalDateTime now){//1、重新生成客户的所有业务类型的数据ListFinanceBiMaintainDto maintainDtoList financeBiBusinessTypeSupport.getMaintainListByCustomerId(customerId);if(CollectionUtils.isEmpty(maintainDtoList)){return ;}//生成每个指标的数据MapBusinessIndicatorEnum,ListFinanceBiMaintainDto indicatorMaintainDtoMap maintainDtoList.stream().collect(Collectors.groupingBy(FinanceBiMaintainDto::getIndicator));indicatorMaintainDtoMap.forEach((k,v)-{log.info(重新生成财务业务指标指定客户【{}】的【{}】支持处理开始,customerId,k);financeBiManager.updateForBiMaintain(k, v,user,now);});}运行耗时1420.145秒
2.0版本 平衡任务队列模式
1.0 版本 由于不同客户的数据量不同导致生成数据的耗时不同因此按照客户数量均分任务的的方式对于每个线程来说任务量是不一样的因此可能会导致部分线程太忙部分线程太空的情况。因此调整为使用队列方式来解决任务分配的问题每个线程自己取队列中取要处理的客户直到所有队列中的客户都被处理完所有的线程结束。这样就避免的线程任务量不平衡问题。
updateForGenerateByCustomerId 方法不需要改造只需要调整任务分配的相关方法就可以。
private void updateForRegenerateByCustomer(ListInteger customerIdList, SystemUserCommonDTO user,LocalDateTime now) {ListCustomerBaseInfo baseInfoList CollectionUtils.isEmpty(customerIdList) ? customerInfoService.listAll() :customerInfoService.listByIdList(customerIdList);//先清理客户的数据updateForCleanByCustomerIdList(baseInfoList, user, now);int maxSize baseInfoList.size();int currentMaxPoolSize Math.min(maxPoolSize, maxSize);//根据线程数构建固定的任务数量CompletableFuture?[] tasks new CompletableFuture?[currentMaxPoolSize];//构建待处理的客户队列由于这里没有并发读写的情况因此用ConcurrentLinkedQueue效率会更高一点。ConcurrentLinkedQueueInteger queue new ConcurrentLinkedQueue(baseInfoList.stream().map(CustomerBaseInfo::getId).collect(Collectors.toList()));//创建多个线程去消耗客户队列for (int i 0; i currentMaxPoolSize; i) {tasks[i] CompletableFuture.runAsync(() - updateForGenerateByCustomerIdList(queue, user, now), executorService).whenCompleteAsync((k, v) - {if (v ! null) {log.error(String.format(重新生成财务业务指标客户的所有数据-线程【%s】发生异常,Thread.currentThread().getName()), v);} else {log.info(重新生成财务业务指标客户的所有数据-线程【{}】完成,Thread.currentThread().getName());}});}// 向线程池提交任务CompletableFuture.allOf(tasks).whenComplete((v, th) - log.info(重新生成财务业务指标客户的所有数据-【{}】个子线程处理完成, tasks.length)).join();}/*** 生成指定客户列表的所有数据**/private void updateForGenerateByCustomerIdList(ConcurrentLinkedQueueInteger queue, SystemUserCommonDTO user,LocalDateTime now) {Integer customerId queue.poll();//循环从客户队列中取出待处理的客户直到所有客户都处理完毕。while (customerId ! null) {final Integer currentCustomerId customerId;TransactionalUtils.runWithNewTransactional(() - updateForGenerateByCustomerId(currentCustomerId, user, now));customerId queue.poll();}}
优化后的耗时:1037.059秒
3.0版本 优化调度平衡任务队列模式
2.0版本虽然解决的了每个线程任务量不平衡的问题但可能出现某个数据量很大的客户在队列的尾部导致当其他线程都处理完所有的客户时取到最大数据量的客户的线程仍在运行任务整体的耗时被增加。因此需要优化调度将耗时高的客户调度到队列头部保证耗时最长的客户的优先处理从而避免最后等待耗时长的线程。
updateForGenerateByCustomerIdList 方法不需要改造只需要队列构造处理就可以。 private void updateForRegenerateByCustomer(ListInteger customerIdList, SystemUserCommonDTO user,LocalDateTime now) {ListCustomerBaseInfo baseInfoList CollectionUtils.isEmpty(customerIdList) ? customerInfoService.listAll() :customerInfoService.listByIdList(customerIdList);//先清理客户的数据updateForCleanByCustomerIdList(baseInfoList, user, now);//获取客户的统计数据MapInteger, CustomerStatisticsInfo customerStatisticsInfoMap customerStatisticsInfoService.listAll().stream().collect(Collectors.toMap(CustomerStatisticsInfo::getCustomerId, Function.identity()));int maxSize baseInfoList.size();int currentMaxPoolSize Math.min(maxPoolSize, maxSize);CompletableFutureString[] tasks new CompletableFuture[currentMaxPoolSize];//根据客户的统计数据构建待处理的客户队列ConcurrentLinkedQueueInteger queue baseInfoList.stream().map(item - customerStatisticsInfoMap.get(item.getId())).filter(Objects::nonNull)
//队列按照客户数据量倒序排列 .sorted(Comparator.comparing(CustomerStatisticsInfo::getNumberOfCheckedSatisfactoryActivitys,Comparator.reverseOrder())).map(CustomerStatisticsInfo::getCustomerId).collect(Collectors.toCollection(ConcurrentLinkedQueue::new));for (int i 0; i currentMaxPoolSize; i) {tasks[i] CompletableFuture.supplyAsync(() - {updateForGenerateByCustomerIdList(queue, user, now);return Thread.currentThread().getName();}, executorService).whenCompleteAsync((k, ex) - {if (ex ! null) {log.error(String.format(重新生成财务业务指标客户的所有数据-线程【%s】发生异常, k), ex);} else {log.info(重新生成财务业务指标客户的所有数据-线程【{}】完成, k);}});}// 向线程池提交任务CompletableFuture.allOf(tasks).whenComplete((v, th) - log.info(重新生成财务业务指标客户的所有数据-【{}】个子线程处理完成, tasks.length)).join();}耗时:726.725秒
总结
最终的耗时从1400多秒 降低到700多秒。降低了一半左右。
