0
点赞
收藏
分享

微信扫一扫

记一次每日跑批任务耗时性能从六分钟优化到半分钟历程及总结

菜头粿子园 2021-09-30 阅读 83
日记本

我觉得这个跑批任务有优化的空间，伴随着一次次性能优化，最终耗时优化到只需平均20多秒就跑批完毕，鉴于此，写一篇自己的优化过程分享给大家。

一、业务概述

的跑批任务没执行，你调用的时候只能是徒劳，这就是背景。话说，在最初技术方案调研时，通过微信电话本来想沟通一下，通过一些性能压测工具压测一下对方的接口性能，这样对于我们发起方也有 一个指标可以提供在技术实现中有所参考。但是呢，对方口述说，我们只是对决策平台，真正处理的是背后的数据引擎团队，他们的性能才是整个调用链路上的关键。想法很好，但最终没有压测。之所以 这个情况，是由于对方的数据只在生产环境提供，测试环境没有。所以不太好压测，毕竟压测的时间点不一样，结论也不一样。毕竟他们提供很多业务方调用，既然如此，我想着先按照他提供的一些指标 比如，秒级给出结果，不支持批量调用，只支持同步调用，不支持消息通知。这些结论使得我开始设计第一版技术方案，看看上线后的跑批耗时。

上面说了一大堆话，简言之概况如上图。中台的事情不做完，前台即便调度任务想要提前无济于事。所以在每天用户8、9点上班时，跑批任务就得生成工单。

二、上线版本

如上图，这是上线的第一个初步版本，第一次跑批1100多单，耗时了5~6分钟。当我告知给业务方，人家说，"不错了，还以为得跑半个小时"。然而，作为技术研发的我心中默默表示不甘心， 觉得对这个跑批性能，认为有改善的空间，否则随处时间流逝，待处理订单会越来越多，前期不改善，后期肯定耗时越来越长。

第一版方案处理的过程如下

1、调度平台配置一个job，每天凌晨七点调度调用我这边的一个服务（线上部署两个节点，跑批固定IP一台服务器）（假设A），A服务提供一个rest接口，接口实现异步去执行后续处理订单。

rest接口示例如下：

@Autowired
ScanningOrderTask scanningOrderTask;

@PostMapping("/disposeOrders")
@ApiOperation(value = "待处理订单跑批", notes = "待处理订单跑批")
@NoAuthRequired
public SiaResponse disposeOrders() {
    scanningOrderTask.scanningOrder();
    return SiaResponse.newInstance("待处理订单跑批成功!");
}
复制代码

ScanningOrderTask代码示例如下：

@Component
@Slf4j
public class ScanningOrderTask {

    @Autowired
    ExecutorService threadPoolExecutor;

    @Autowired
    DisposeOrderFacade disposeOrderFacade;

    /**
     * 订单跑谛听规则
     */
    public void scanningOrder(){
        log.info("##### [scanningOrder]待处理订单开始执行跑批处理 ####");
        CompletableFuture.runAsync(() -> disposeOrderFacade.scanningOrders(), threadPoolExecutor);
    }
}
复制代码

2、disposeOrderFacade#scanningOrders提供的业务伪代码实现如下：

public void scanningOrders(){
    int start = 0;
    int limit = 100;
    StopWatch stopWatch = new StopWatch("scanningOrders");
    stopWatch.start();
    List<Object> disposeOrderList = 先查询第一页数据;
    do{
        List<DisposeOrder> repeatOrder = Lists.newArrayList();//二次重试需要再次重试订单
        for(DisposeOrder order : list){
            try{
                operateStrategyManager.execute(operateContext);
            }catch(Exception e){
                //异常重试
                operateStrategyManager.execute(operateContext);
                //再次重试失败，则加入失败订单
                if(再次重试失败){
                    repeatOrder(order);
                }
            }
        }
        //查询下一批 待扫描订单
        start += limit;
        param.setStart(start);
        disposeOrderList = disposeOrderService.queryByPage(param);
        if(CollectionUtils.isNotEmpty(repeatOrder)){
            disposeOrderList.addAll(repeatOrder);
        }
    }while(list非空)
    stopWatch.stop();
    int batchSize = longAdder.intValue() - errorOrderSet.size();
    long durationSeconds = (stopWatch.getLastTaskTimeMillis() / 1000);
    log.info("##### [scanningOrder]跑批处理完毕，共计={}单，耗时={}s ####",batchSize,durationSeconds);
}
复制代码

3、决策引擎异步处理相关代码

public void operation(OperateContext context) {
    //1、调用决策引擎
    Result<AlarmDecisionRe> alarmDecisionResult = alarmDecisionManager.call(AlarmDecisionDTO.builder()
            .appCode(context.getDisposeOrder().getAppCode())
            .scenePhase(ScenePhaseEnum.LOAN_PROCESS.getIndex())
            .build());
    //2、决策输出的结果，再交由另外一个线程池 threadPoolExecutor（通用线程池）异步去处理。
    CompletableFuture.runAsync(() -> handleAlarmResult(alarmDecisionResult.getData()), threadPoolExecutor);
}
复制代码

4、决策输出结果的异步处理相关代码

/**
 * 处理规则校验结果`
 * @param decisionResult
 */
void handleAlarmResult(AlarmDecisionRe decisionResult){
    String appCode = decisionResult.getAppCode();
    StopWatch stopWatch = new StopWatch("handleAlarmResult");
    stopWatch.start();
    if(Objects.isNull(decisionResult) || CollectionUtils.isEmpty(decisionResult.getDecisionRuleList())){
        log.info("[每日跑批]决策输出为空,appCode={}",appCode);
        return;
    }
    //存detail 优先级、编码
    AlarmDetail detail = AlarmDetail.builder()
            .appCode(appCode)
            .alarmTime(new Date())
            .alarmRules(JSON.toJSONString(decisionResult.getDecisionRuleList()))
            .scenePhase(ScenePhaseEnum.LOAN_PROCESS.getIndex())
            .build();
    alarmDetailService.insertRecord(detail);
    List<WorkOrder> oldWorkOrder = workOrderService.queryList(WorkOrderForm.builder()
            .appCode(decisionResult.getAppCode())
            .isFinished(ConstEnum.YesOrNoEnum.NO.getIndex()).build());
    if(CollectionUtils.isEmpty(oldWorkOrder)){
        saveWorkOrder(detail,decisionResult,0);
    }else{
        log.info("[每日跑批]该订单未处理完结无需创建,appCode={}",decisionResult.getAppCode());
    }
    stopWatch.stop();
    long duration = stopWatch.getLastTaskTimeMillis();
    log.info("[每日跑批]handleAlarmResult,appCode={},duration={}ms",appCode,duration);
}
复制代码

决策引擎返回的结果然后异步交由一个线程池去处理。跑批的待处理订单大概只有300多单会命中业务规则，对于命中规则的订单会生成工单外，还需要做相应 一系列业务逻辑处理，包括通知调用其他系统业务处理。而未命中规则，仅需插入一个预警记录外。无需其他操作，这说明，整体耗时在这300多单阻塞同步循环 调用决策引擎耗时占比非常大。

[图片上传失败...(image-4847ba-1608706829493)]

三、优化方案

[图片上传失败...(image-2ab7c3-1608706829493)]

从上图可以看到最明显与初版差异的是，这次优化方案是，把循环处理待处理订单的阻塞操作，改为异步处理，交由一个线程池去处理，串行等待变为并行处理。

disposeOrderFacade#scanningOrders的方法前后变化

//异步处理，把每个待处理订单放在disposeOrderHandleThreadPool（待处理线程池）去处理
CompletableFuture.runAsync(() -> operateStrategyManager.execute(operateContext), disposeOrderHandleThreadPool);
复制代码

/**
 * @description: 线程池配置
 * @Date : 2019/4/25 下午2:36
 * @Author : 石冬冬-Seig Heil
 */
@Configuration
@EnableAsync
@Slf4j
public class ThreadPoolConfig {

    static final int blockingQueueCapacity = 5000;

    static final int keepAliveTime = 60;

    static final int availableProcessors = Runtime.getRuntime().availableProcessors();

    static final int corePoolSize  = availableProcessors * 8;

    static final int maximumPoolSize = 50;

    static {
        log.info("[availableProcessors]={},corePoolSize={},maximumPoolSize={}",availableProcessors,corePoolSize,maximumPoolSize);
    }

    /**
     * 通用线程池配置
     * @return
     */
    @Bean
    public ExecutorService threadPoolExecutor() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("common-pool-%d").build();
        return new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(blockingQueueCapacity), threadFactory);
    }

    /**
     * 跑批待处理订单线程池配置
     * blockingQueueCapacity 需要设置合理大小，目前上线一周，每日跑批 1175单左右；即便线程池已满，使用拒绝策略；有补偿机制。
     * @return
     */
    @Bean
    public ExecutorService disposeOrderHandleThreadPool() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("disposeOrder-pool-%d").build();
        final int corePoolSize = availableProcessors * 6,maxPoolSize = corePoolSize * 2,keepAliveTime = 60,blockingQueueCapacity = 5000;
        return new ThreadPoolExecutor(corePoolSize,maxPoolSize,keepAliveTime, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(blockingQueueCapacity), threadFactory);
    }

}
复制代码

这是上述线程池配置，也是目前跑到线上的配置。其中 disposeOrderHandleThreadPool （待处理订单线程池）也是经过几次优化后最终设置的参数配置。

  • 第一次优化，出现队列拒绝现象。
/**
 * 跑批待处理订单线程池配置
 * @return
 */
@Bean
public ExecutorService disposeOrderHandleThreadPool() {
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("carthage-disposeOrder-pool-%d").build();
    final int corePoolSize = 5,maxPoolSize = 12;
    return new ThreadPoolExecutor(corePoolSize,maxPoolSize,60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(500), threadFactory);
}
复制代码

异常信息如下所图(好在待处理有重试机制，跑批任务数据没有问题)：

  • 第二次优化，无队列拒绝现象。
/**
 * 跑批待处理订单线程池配置
 * blockingQueueCapacity 需要设置合理大小，目前上线一周，每日跑批 1175单左右；即便线程池已满，使用拒绝策略；有补偿机制。
 * @return
 */
@Bean
public ExecutorService disposeOrderHandleThreadPool() {
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("carthage-disposeOrder-pool-%d").build();
    final int corePoolSize = availableProcessors * 4,maxPoolSize = corePoolSize,keepAliveTime = 60,blockingQueueCapacity = 5000;
    return new ThreadPoolExecutor(corePoolSize,maxPoolSize,keepAliveTime, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(blockingQueueCapacity), threadFactory);
}
复制代码

跑批情况（共计=1210单，主线程：耗时=1s；子线程最后一单最后完成时间 2020-08-16 07:00:27.794；），跑批任务正常，无线程任务队列拒绝现象。

  • 第三次优化，进一步调整待处理和通用线程池核心数大小。
static final int blockingQueueCapacity = 5000;

static final int keepAliveTime = 60;

static final int availableProcessors = Runtime.getRuntime().availableProcessors();

static final int corePoolSize  = availableProcessors * 8;

static final int maximumPoolSize = 50;

/**
 * 通用线程池配置
 * @return
 */
@Bean
public ExecutorService threadPoolExecutor() {
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("common-pool-%d").build();
    return new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(blockingQueueCapacity), threadFactory);
}

/**
 * 跑批待处理订单线程池配置
 * blockingQueueCapacity 需要设置合理大小，目前上线一周，每日跑批 1175单左右；即便线程池已满，使用拒绝策略；有补偿机制。
 * @return
 */
@Bean
public ExecutorService disposeOrderHandleThreadPool() {
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("disposeOrder-pool-%d").build();
    final int corePoolSize = availableProcessors * 6,maxPoolSize = corePoolSize * 2,keepAliveTime = 60,blockingQueueCapacity = 5000;
    return new ThreadPoolExecutor(corePoolSize,maxPoolSize,keepAliveTime, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(blockingQueueCapacity), threadFactory);
}
复制代码

上图是今日（8月21日）跑批容器实例（单机）的CPU以及堆内存使用情况。

上图是今日（8月21日）跑批容器实例（单机）的CPU以及线程数。

上图是今日（8月21日）跑批容器实例（单机）的堆内存和GC情况（JDK8，使用G1垃圾收集器，跑批期间发生一次MGC）。

四、归纳总结

举报

相关推荐

记一次Spring|项目启动耗时超级长的优化【从7分钟优化到1分多钟】

javaspring开发语言

实现页面每隔10分钟刷新一次

编程语言

记一次To B开发普通的性能优化历程......报表优化

数据结果集sqlJava编程语言

java icon表达式 每五分钟执行一次

定时任务javaJava后端开发

记一次接口性能优化实践总结：优化接口性能的八个建议

随笔面试学习Java 杂谈Spring-...开发技巧Java

记一次效果“非常”显著的性能优化[五秒到一秒】

性能优化java开发语言后端

crontab设置定时任务脚本文件每秒运行一次，每个月最后一天的21点到22点 每20分钟（1200秒）执行一次...

pythonshelljarbashphpHarmonyOS后端开发

如何把一个报表从2分钟优化到1秒?---GaussDB数据库

OLTP金融行业OLTP数据库选型SQL数据查询Html/CSS前端开发

使用 SpringAOP和SpringIOC一个bean调用另一个bean耗时超过1分钟什么情况？获取一次请求流经方法的调用次数和调用耗时

SpringIOC调用耗时过长JAVA调用Bean的使用耗时d3kafka数据Java编程语言

记一次 .NET 程序的性能优化实战（4）—— .NET 源码查看及使用 vs 调试

正则表达式githubmicrosoft虚拟化云计算
0 条评论