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文章来源:https://www.uudwc.com/A/gVRMo/
@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分钟了。文章来源地址https://www.uudwc.com/A/gVRMo/