SpringBoot利用ThreadPoolTaskExecutor批量插入百萬級(jí)數(shù)據(jù)實(shí)測！

前言

• 開發(fā)目的： 提高百萬級(jí)數(shù)據(jù)插入效率。
• 采取方案： 利用ThreadPoolTaskExecutor多線程批量插入。
• 采用技術(shù)： springboot2.1.1+mybatisPlus3.0.6+swagger2.5.0+Lombok1.18.4+postgresql+ThreadPoolTaskExecutor等。

基于 Spring Boot + MyBatis Plus + Vue & Element 實(shí)現(xiàn)的后臺(tái)管理系統(tǒng) + 用戶小程序，支持 RBAC 動(dòng)態(tài)權(quán)限、多租戶、數(shù)據(jù)權(quán)限、工作流、三方登錄、支付、短信、商城等功能

• 項(xiàng)目地址：https://gitee.com/zhijiantianya/ruoyi-vue-pro
• 視頻教程：https://doc.iocoder.cn/video/

具體實(shí)現(xiàn)細(xì)節(jié)

application-dev.properties添加線程池配置信息

> 基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 實(shí)現(xiàn)的后臺(tái)管理系統(tǒng) + 用戶小程序，支持 RBAC 動(dòng)態(tài)權(quán)限、多租戶、數(shù)據(jù)權(quán)限、工作流、三方登錄、支付、短信、商城等功能
>
> * 項(xiàng)目地址：
> * 視頻教程：

# 異步線程配置
# 配置核心線程數(shù)
async.executor.thread.core_pool_size = 30
# 配置最大線程數(shù)
async.executor.thread.max_pool_size = 30
# 配置隊(duì)列大小
async.executor.thread.queue_capacity = 99988
# 配置線程池中的線程的名稱前綴
async.executor.thread.name.prefix = async-importDB-

spring容器注入線程池bean對(duì)象

@Configuration

@EnableAsync

@Slf4j

publicclassExecutorConfig{
@Value("${async.executor.thread.core_pool_size}")
privateintcorePoolSize;
@Value("${async.executor.thread.max_pool_size}")
privateintmaxPoolSize;
@Value("${async.executor.thread.queue_capacity}")
privateintqueueCapacity;
@Value("${async.executor.thread.name.prefix}")
privateStringnamePrefix;

@Bean(name="asyncServiceExecutor")
publicExecutorasyncServiceExecutor(){
log.warn("startasyncServiceExecutor");
//在這里修改
ThreadPoolTaskExecutorexecutor=newVisiableThreadPoolTaskExecutor();
//配置核心線程數(shù)
executor.setCorePoolSize(corePoolSize);
//配置最大線程數(shù)
executor.setMaxPoolSize(maxPoolSize);
//配置隊(duì)列大小
executor.setQueueCapacity(queueCapacity);
//配置線程池中的線程的名稱前綴
executor.setThreadNamePrefix(namePrefix);
// rejection-policy：當(dāng)pool已經(jīng)達(dá)到max size的時(shí)候，如何處理新任務(wù)
// CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是有調(diào)用者所在的線程來執(zhí)行
executor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());
//執(zhí)行初始化
executor.initialize();
returnexecutor;
}
}

創(chuàng)建異步線程 業(yè)務(wù)類

@Service
@Slf4j
publicclassAsyncServiceImplimplementsAsyncService{
@Override
@Async("asyncServiceExecutor")
publicvoidexecuteAsync(ListlogOutputResults,LogOutputResultMapperlogOutputResultMapper,CountDownLatchcountDownLatch){
try{
log.warn("startexecuteAsync");
//異步線程要做的事情
logOutputResultMapper.addLogOutputResultBatch(logOutputResults);
log.warn("endexecuteAsync");
}finally{
countDownLatch.countDown();//很關(guān)鍵,無論上面程序是否異常必須執(zhí)行countDown,否則await無法釋放
}
}
}

創(chuàng)建多線程批量插入具體業(yè)務(wù)方法

@Override
publicinttestMultiThread(){
ListlogOutputResults=getTestData();
//測試每100條數(shù)據(jù)插入開一個(gè)線程
List>lists=ConvertHandler.splitList(logOutputResults,100);
CountDownLatchcountDownLatch=newCountDownLatch(lists.size());
for(ListlistSub:lists){
asyncService.executeAsync(listSub,logOutputResultMapper,countDownLatch);
}
try{
countDownLatch.await();//保證之前的所有的線程都執(zhí)行完成，才會(huì)走下面的；
//這樣就可以在下面拿到所有線程執(zhí)行完的集合結(jié)果
}catch(Exceptione){
log.error("阻塞異常:"+e.getMessage());
}
returnlogOutputResults.size();
}

模擬2000003 條數(shù)據(jù)進(jìn)行測試

多線程 測試 2000003 耗時(shí)如下：耗時(shí)1.67分鐘

本次開啟30個(gè)線程，截圖如下：

單線程測試2000003 耗時(shí)如下：耗時(shí)5.75分鐘

檢查多線程入庫的數(shù)據(jù)，檢查是否存在重復(fù)入庫的問題：

根據(jù)id分組，查看是否有id重復(fù)的數(shù)據(jù)，通過sql語句檢查，沒有發(fā)現(xiàn)重復(fù)入庫的問題

檢查數(shù)據(jù)完整性：通過sql語句查詢，多線程錄入數(shù)據(jù)完整

測試結(jié)果

不同線程數(shù)測試：

總結(jié)

通過以上測試案列，同樣是導(dǎo)入2000003 條數(shù)據(jù)，多線程耗時(shí)1.67分鐘，單線程耗時(shí)5.75分鐘。通過對(duì)不同線程數(shù)的測試，發(fā)現(xiàn)不是線程數(shù)越多越好，具體多少合適，網(wǎng)上有一個(gè)不成文的算法：

CPU核心數(shù)量*2 +2 個(gè)線程。

附：測試電腦配置

審核編輯 ：李倩