1.分析问题
原程序是分页查询EventAffinityScoreDO表的数据,每次获取2000条在一个个遍历去更新EventAffinityScoreDO表的数据。但是这样耗时比较慢,测试过30万的数据需要2小时
private void eventSubjectHandle(String tenantId, String eventSubject) {// 查询eventAffinityScoreDO表,更新时间小于今天的(今天更新过的不更新)final Integer pageSize = 2000;PageResult<EventAffinityScoreDO> groupPag =eventAffinityScoreDbService.findByTenantIdAndTimePage(tenantId, eventSubject, 1, pageSize);Integer pages = groupPag.getPages();Integer pageNum = groupPag.getPageNum();while (pages >= pageNum) {if (pageNum > 1) {groupPag =eventAffinityScoreDbService.findByTenantIdAndTimePage(tenantId, eventSubject, 1, pageSize);}List<EventAffinityScoreDO> list = groupPag.getList();forEventAffinityScore(tenantId, eventSubject, list);if (list.size() < pageSize) {break;}pageNum++;}}private void forEventAffinityScore(String tenantId, String eventSubject, List<EventAffinityScoreDO> eventAffinityScoreDOS) {eventAffinityScoreDOS.forEach((eventAffinityScoreDO) -> {//更新EventAffinityScoreDO表数据updateOrAddAffinity(tenantId,eventAffinityScoreDO.getChatLabsId(),eventAffinityScoreDO.getEconomyId(),eventAffinityScoreDO.getAttributeValue(),eventSubject,eventAffinityScoreDO.getAttributeName());});}
单个线程一个个遍历去更新表数据太慢了,我想把2000的数据分成多份,每份200条,可以分成10份。每份用一个线程去跑。这样跑2000的时间就大大缩短。大概等于跑200个数据的时间。
这里想到使用CountDownLatch
2.知识点CountDownLatch
CountDownLatch 是 Java 中的一个并发工具类,用于在多线程环境中控制线程的执行顺序。它允许一个或多个线程等待其他线程完成操作后再继续执行。
CountDownLatch 的构造方法接受一个整数作为参数,表示需要等待的线程数量。当一个线程完成了自己的任务后,可以调用 countDown() 方法来将计数器减1。当计数器的值变为0时,所有等待的线程都会被释放,可以继续执行。
3.解决问题
我们使用Lists.partition,把2000的集合拆分成每份200的小份,共10分。
CountDownLatch countDownLatch = new CountDownLatch(partition.size())
设置CountDownLatch需要等待的线程数为拆分后的份数partition.size()
,也就是10份
countDownLatch.countDown();
每跑完一份计数器减一
countDownLatch.await();
计数器减完主程序开始执行,继续循环后面的2000份
private void eventSubjectHandle(String tenantId, String eventSubject)throws InterruptedException {// 查询eventAffinityScoreDO表,更新时间小于今天的(今天更新过的不更新)final Integer pageSize = 2000;PageResult<EventAffinityScoreDO> groupPag =eventAffinityScoreDbService.findByTenantIdAndTimePage(tenantId, eventSubject, 1,pageSize);Integer pages = groupPag.getPages();Integer pageNum = groupPag.getPageNum();while (pages >= pageNum) {if (pageNum > 1) {groupPag =eventAffinityScoreDbService.findByTenantIdAndTimePage(tenantId, eventSubject, 1, pageSize);}List<EventAffinityScoreDO> list = groupPag.getList();//Lists.partition把list进行拆分,没份200个List<List<EventAffinityScoreDO>> partition = Lists.partition(list, 200);//设置需要等待的线程数量,就是我们的集合大小CountDownLatch countDownLatch = new CountDownLatch(partition.size());for (List<EventAffinityScoreDO> eventAffinityScoreDOS : partition) {eventSubjectExecutorPool.execute(() -> {try {forEventAffinityScore(tenantId, eventSubject, eventAffinityScoreDOS);} catch (Exception e) {log.info("AutoAffinityJob updateAffinityByEventSubject error tenantId:{},eventSubject:{}",tenantId,eventSubject,e);}//每处理完200份计数器减一countDownLatch.countDown();});}//计数器减完主程序开始执行,继续循环后面的2000份countDownLatch.await();if (list.size() < pageSize) {break;}pageNum++;}}private void forEventAffinityScore(String tenantId, String eventSubject, List<EventAffinityScoreDO> eventAffinityScoreDOS) {eventAffinityScoreDOS.forEach((eventAffinityScoreDO) -> {// 根据生态中事件属性属性值更新or新增影响到的内容亲和力updateOrAddAffinity(tenantId,eventAffinityScoreDO.getChatLabsId(),eventAffinityScoreDO.getEconomyId(),eventAffinityScoreDO.getAttributeValue(),eventSubject,eventAffinityScoreDO.getAttributeName());});}
这里需要注意的是如果线程池设置的太小,会导致触发拒绝策略。如果触发了拒绝策略countDownLatch.countDown()
就不会执行了。就会导致countDownLatch.await()
一直等待。所以这里我把线程池的队列设置的很大Integer.MAX_VALUE
,这样不会触发拒绝策略。因为我们最多就10个线程,也不会导致出现OOM
@Configuration
@Slf4j
public class CalculateAffinityThreadPool {@Bean(name = "eventSubjectExecutorPool")public ExecutorService eventSubjectExecutorPool() {int poolSize = ThreadExecutorUtils.getNormalCoreSize();return ThreadExecutorUtils.createNormalThreadPool(poolSize,poolSize,0L,TimeUnit.MILLISECONDS,Integer.MAX_VALUE,"eventSubject-pool",false);}}
经过测试跑30万的数据只需要20分钟了。