快捷導(dǎo)航

springboot使用ThreadPoolTaskExecutor多線程批量插入百萬級數(shù)據(jù)的實現(xiàn)方法

更新時間：2023年02月04日 09:26:03 作者：阿啄debugIT

這篇文章主要介紹了springboot利用ThreadPoolTaskExecutor多線程批量插入百萬級數(shù)據(jù),本文通過示例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下

前言

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

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

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

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

spring容器注入線程池bean對象

@Configuration
 
@EnableAsync
 
@Slf4j
 
public class ExecutorConfig {
    @Value("${async.executor.thread.core_pool_size}")
    private int corePoolSize;
    @Value("${async.executor.thread.max_pool_size}")
    private int maxPoolSize;
    @Value("${async.executor.thread.queue_capacity}")
    private int queueCapacity;
    @Value("${async.executor.thread.name.prefix}")
    private String namePrefix;
 
    @Bean(name = "asyncServiceExecutor")
    public Executor asyncServiceExecutor() {
        log.warn("start asyncServiceExecutor");
        //在這里修改
        ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
        //配置核心線程數(shù)
        executor.setCorePoolSize(corePoolSize);
        //配置最大線程數(shù)
        executor.setMaxPoolSize(maxPoolSize);
        //配置隊列大小
        executor.setQueueCapacity(queueCapacity);
        //配置線程池中的線程的名稱前綴
        executor.setThreadNamePrefix(namePrefix);
        // rejection-policy：當(dāng)pool已經(jīng)達(dá)到max size的時候，如何處理新任務(wù)
        // CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是有調(diào)用者所在的線程來執(zhí)行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //執(zhí)行初始化
        executor.initialize();
        return executor;
    }
}

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

@Service
 
@Slf4j
 
public class AsyncServiceImpl implements AsyncService {
@Override
    @Async("asyncServiceExecutor")
    public void executeAsync(List<LogOutputResult> logOutputResults, LogOutputResultMapper logOutputResultMapper, CountDownLatch countDownLatch) {
        try{
            log.warn("start executeAsync");
            //異步線程要做的事情
            logOutputResultMapper.addLogOutputResultBatch(logOutputResults);
            log.warn("end executeAsync");
        }finally {
            countDownLatch.countDown();// 很關(guān)鍵, 無論上面程序是否異常必須執(zhí)行countDown,否則await無法釋放
        }
    }
}

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

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

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

多線程測試 2000003 耗時如下：耗時1.67分鐘

本次開啟30個線程，截圖如下：

單線程測試2000003 耗時如下：耗時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ù)，多線程耗時1.67分鐘，單線程耗時5.75分鐘。通過對不同線程數(shù)的測試，發(fā)現(xiàn)不是線程數(shù)越多越好，具體多少合適，網(wǎng)上有一個不成文的算法：
CPU核心數(shù)量*2 +2 個線程。
附：測試電腦配置

到此這篇關(guān)于springboot使用ThreadPoolTaskExecutor多線程批量插入百萬級數(shù)據(jù)的實現(xiàn)方法的文章就介紹到這了,更多相關(guān)springboot ThreadPoolTaskExecutor多線程內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: