springboot使用ThreadPoolTaskExecutor多執行緒批次插入百萬級資料的實現方法

# 非同步執行緒設定
# 設定核心執行緒數
async.executor.thread.core_pool_size = 30
# 設定最大執行緒數
async.executor.thread.max_pool_size = 30
# 設定佇列大小
async.executor.thread.queue_capacity = 99988
# 設定執行緒池中的執行緒的名稱字首
async.executor.thread.name.prefix = async-importDB-

@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();
        //設定核心執行緒數
        executor.setCorePoolSize(corePoolSize);
        //設定最大執行緒數
        executor.setMaxPoolSize(maxPoolSize);
        //設定佇列大小
        executor.setQueueCapacity(queueCapacity);
        //設定執行緒池中的執行緒的名稱字首
        executor.setThreadNamePrefix(namePrefix);
        // rejection-policy：當pool已經達到max size的時候，如何處理新任務
        // CALLER_RUNS：不在新執行緒中執行任務，而是有呼叫者所在的執行緒來執行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //執行初始化
        executor.initialize();
        return executor;
    }
}

@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();// 很關鍵, 無論上面程式是否異常必須執行countDown,否則await無法釋放
        }
    }
}

@Override
    public int testMultiThread() {
        List<LogOutputResult> logOutputResults = getTestData();
        //測試每100條資料插入開一個執行緒
        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(); //保證之前的所有的執行緒都執行完成，才會走下面的；
            // 這樣就可以在下面拿到所有執行緒執行完的集合結果
        } catch (Exception e) {
            log.error("阻塞異常:"+e.getMessage());
        }
        return logOutputResults.size();
    }