多线程读58万数据处理后写入到另外一张表中性能调优实例

怼烎@ 2022-11-13 00:52 52阅读 0赞

场景是设置线程池，从51万数据读取数据，将数据拆成60波，经过drools引擎处理后，将数据写入到另外一张表中。  
**1 事务**  
`@Transactional(readOnly = true)`，只读事务针对一次性多条查询语句，例如统计查询，报表查询等情况，多条查询SQL必须保证整体的读一致性，所以一般情况下是不需要指定只读事务，因为数据库默认支持SQL执行期间的读一致性.  
`readOnly=false`表明所注解的方法或类是增加，删除，修改数据。

@Override
        public BaziInfo selectByPrimaryKey(java.lang.Integer id) { 
            return baziInfoMapper.selectByPrimaryKey(id);
        }
    
    @Transactional(readOnly = false)
        @Override
        public boolean updateByPrimaryKeySelective(BaziInfo record) { 
            int flag = baziInfoMapper.updateByPrimaryKeySelective(record);
            return flag > 0;
        }

没播数据8140条，如果等到所有数据都写完，再做事务提交，那就等待猴年马月，这里再写一下spring的事务传播。在外层再套一个@Transactional(readOnly = false),事务将传递下去。因为默认的是`propagation=Propagation.REQUIRED`  
**2 多线程**  
**2.1 线程池**

package com.dzmsoft.qingqiu.server.config;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /** * 线程池 */
    @Configuration
    @EnableAsync
    @Slf4j
    public class ExecutorConfig { 
    
        @Bean
        public Executor asyncServiceExecutor(){ 
            log.info("start asyncServiceExecutor");
    // ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
            // 核心线程数
            executor.setCorePoolSize(50);
            executor.setAllowCoreThreadTimeOut(true);
            // 最大线程数
            executor.setMaxPoolSize(50);
            // 配置队列大小
            executor.setQueueCapacity(200);
            // 线程名称
            executor.setThreadNamePrefix("qingqiu-");
            // rejection policy ，pool达到max size，采取caller_runs策略，不在新线程中执行任务，而是在有调用者所在的线程中执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //
            executor.initialize();
            return executor;
        }
    }

**2.2 controller**

package com.dzmsoft.qingqiu.analysis.controller;
    
    import com.dzmsoft.framework.web.view.ResponseResult;
    import com.dzmsoft.qingqiu.analysis.service.ShyAnalysisService;
    import com.dzmsoft.qingqiu.base.enums.LiuShiJiaZiEnum;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.lang.reflect.InvocationTargetException;
    import java.util.ArrayList;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    import java.util.List;
    
    @Slf4j
    @RestController
    @RequestMapping("shy")
    public class ShyController { 
        @Autowired
        private ShyAnalysisService shyAnalysisService;
    
    
    
        @PostMapping("genMatchExcel/all")
        @ResponseBody
        public ResponseResult genMatchExcel(){ 
            try { 
                List<Future<Boolean>> tasks = genTasks();
                for (Future<Boolean> task:tasks){ 
                    task.get();
                }
                log.info("all tasks finished");
                return new ResponseResult(true);
            } catch (InterruptedException | ExecutionException | InvocationTargetException | IllegalAccessException e) { 
                e.printStackTrace();
                return new ResponseResult(false,e.getMessage());
            }
        }
    
        private List<Future<Boolean>> genTasks() throws InterruptedException, InvocationTargetException, IllegalAccessException { 
            List<Future<Boolean>> tasks = new ArrayList<>();
            for (LiuShiJiaZiEnum jiazi: LiuShiJiaZiEnum.values()){ 
                tasks.add(shyAnalysisService.matchQrByDay(jiazi));
            }
            return tasks;
        }
    }

**2.3 service**

package com.dzmsoft.qingqiu.analysis.service.impl;
    
    import com.dzmsoft.qingqiu.analysis.service.ShyAnalysisService;
    import com.dzmsoft.qingqiu.base.enums.LiuShiJiaZiEnum;
    import com.dzmsoft.qingqiu.base.pojo.BaZi;
    import com.dzmsoft.qingqiu.db.pojo.BaziInfo;
    import com.dzmsoft.qingqiu.db.pojo.ShyGenyinQr;
    import com.dzmsoft.qingqiu.db.service.BaziInfoService;
    import com.dzmsoft.qingqiu.db.service.ShyGenyinQrService;
    import com.dzmsoft.qingqiu.rules.dto.ShyGenYinDto;
    import com.dzmsoft.qingqiu.rules.service.IQiangRuoService;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.beanutils.BeanUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.scheduling.annotation.AsyncResult;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    
    import java.lang.reflect.InvocationTargetException;
    import java.util.concurrent.Future;
    import java.util.List;
    
    @Slf4j
    @Service
    public class ShyAnalysisServiceImpl implements ShyAnalysisService { 
        @Autowired
        private BaziInfoService baziInfoService;
        @Autowired
        private ShyGenyinQrService shyGenyinQrService;
        @Autowired
        private IQiangRuoService qiangRuoService;
    
        @Async("asyncServiceExecutor")
        @Override
        public Future<Boolean> matchQrByDay(LiuShiJiaZiEnum day) throws InterruptedException, InvocationTargetException, IllegalAccessException { 
            log.info(String.format("task %s start", day.display()));
            long start = System.currentTimeMillis();
            List<BaziInfo> baziInfos = baziInfoService.getByDay(day.display());
            // 生成
            ShyGenYinDto dto = null;
            ShyGenyinQr shyGenyinQr = null;
            for (BaziInfo baziInfo: baziInfos){ 
                BaZi bazi = new BaZi.Builder().setYear(baziInfo.getYear())
                        .setMonth(baziInfo.getMonth()).setDay(baziInfo.getDay())
                        .setHour(baziInfo.getHour()).build();
                dto = qiangRuoService.getQiangRuo(bazi);
                shyGenyinQr = convert2Shy(dto);
                //
                shyGenyinQrService.insertSelective(shyGenyinQr);
            }
    // Thread.sleep(1000);
            long end = System.currentTimeMillis();
            log.info("日柱{}:数量{},查询耗时{}", day.display(),baziInfos.size(), end-start);
            return new AsyncResult<>(true);
        }
    
        /** * 转换可识别的对象 * @param dto * @return */
        private ShyGenyinQr convert2Shy(ShyGenYinDto dto) throws InvocationTargetException, IllegalAccessException { 
            ShyGenyinQr record = new ShyGenyinQr();
            BeanUtils.copyProperties(record,dto);
            return record;
        }
    }

**2.4 drools的service**

import com.drools.core.KieTemplate;
    import com.dzmsoft.qingqiu.base.enums.ShengKeBiEnum;
    import com.dzmsoft.qingqiu.base.pojo.BaZi;
    import com.dzmsoft.qingqiu.base.util.HeUtil;
    import com.dzmsoft.qingqiu.base.util.TianGanUtil;
    import com.dzmsoft.qingqiu.base.util.WuXingUtil;
    import com.dzmsoft.qingqiu.rules.dto.ShyGenYinDto;
    import com.dzmsoft.qingqiu.rules.service.IQiangRuoService;
    import org.kie.api.runtime.KieSession;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    /** * 强弱 */
    @Service
    public class QiangRuoService implements IQiangRuoService { 
    @Autowired
        private KieTemplate kieTemplate;
     @Override
        public ShyGenYinDto getQiangRuo(ShyGenYinDto dto) { 
            KieSession session = kieTemplate.getKieSession("shy3.xlsx");
    // KieSession session = kieTemplate.getKieSession("test01.drl");
            session.insert(dto);
            session.fireAllRules();
            logger.debug(dto.getResult());
            return dto;
        }
    }

**2.5 线程阻塞**  
通过jprofier可以看到红色阻塞的线程`ForkJoinPoll`,看到阻塞了，我们能做什么呢？  
![1][]  
出了问题先看CPU.因为如果CPU使用率较小，可能是因为线程池的数量太小了，导致请求在排队等待进入线程池。但是下图看CPU使用率并不低，参考[性能监控诊断linux系统笔记][linux]CPU利用率超过75，就有问题需要优化。可以这里有两个值，`Process load`和`System load`，`System load`要高很多，那么说明什么问题呢？  
![1][1 1]

**3 数据库**

[1]: /images/20221022/9ffabcbacdf844a7aee5f66172e25bfc.png
[linux]: https://blog.csdn.net/warrah/article/details/114945254?spm=1001.2014.3001.5501
[1 1]: /images/20221022/5317ff17747d419e8e2e309c21839ecd.png