利用 Spring Boot 3.4 结合 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-impontDB配置类
package com.icoderoad.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
@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("启动线程池 asyncServiceExecutor");
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(corePoolSize);
executor.setMaxPoolSize(maxPoolSize);
executor.setQueueCapacity(queueCapacity);
executor.setThreadNamePrefix(namePrefix);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}异步任务执行
package com.icoderoad.service.impl;
import com.icoderoad.mapper.LogOutputResultMapper;
import com.icoderoad.model.LogOutputResult;
import com.icoderoad.service.AsyncService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.concurrent.CountDownLatch;
@Slf4j
@Service
public class AsyncServiceImpl implements AsyncService {
@Async("asyncServiceExecutor")
@Override
public void executeAsync(List < LogOutputResult > logOutputResults, LogOutputResultMapper logOutputResultMapper, CountDownLatch countDownLatch) {
try {
log.warn("执行异步插入任务");
logOutputResultMapper.addLogOutputResultBatch(logOutputResults);
} finally {
countDownLatch.countDown();
}
}
}业务调用
package com.icoderoad.service.impl;
import com.icoderoad.mapper.LogOutputResultMapper;
import com.icoderoad.model.LogOutputResult;
import com.icoderoad.service.AsyncService;
import com.icoderoad.utils.ConvertHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.concurrent.CountDownLatch;
@Slf4j
@Service
public class LogOutputService {
private final AsyncService asyncService;
private final LogOutputResultMapper logOutputResultMapper;
public LogOutputService(AsyncService asyncService, LogOutputResultMapper logOutputResultMapper) {
this.asyncService = asyncService;
this.logOutputResultMapper = logOutputResultMapper;
}
public int testMultiThread() {
List<LogOutputResult> logOutputResults = getTestData();
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();
}
private List<LogOutputResult> getTestData() {
return ConvertHandler.generateTestData(3000000);
}
}工具类
package com.icoderoad.utils;
import com.icoderoad.model.LogOutputResult;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class ConvertHandler {
public static <T> List<List<T>> splitList(List<T> list, int size) {
List<List<T>> parts = new ArrayList<>();
for (int i = 0; i < list.size(); i += size) {
parts.add(new ArrayList<>(list.subList(i, Math.min(list.size(), i + size))));
}
return parts;
}
public static List<LogOutputResult> generateTestData(int count) {
return IntStream.range(0, count)
.mapToObj(i -> new LogOutputResult((long) i, "TestLog " + i))
.collect(Collectors.toList());
}
}数据访问层
package com.icoderoad.mapper;
import com.icoderoad.model.LogOutputResult;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import java.util.List;
@Mapper
public interface LogOutputResultMapper {
@Insert("INSERT INTO log_output_result (id, message) VALUES (#{id}, #{message})")
void addLogOutputResultBatch(List<LogOutputResult> logOutputResults);
}测试结果
单线程：插入 300万 数据，耗时 5.75分钟。
30个线程：并发插入 300万 数据，耗时 1.67分钟，效率提升 3.4倍。
数据完整性检查无误，无重复数据。
结论
在高并发、大数据量插入的场景下，传统的单线程批量插入方式已经无法满足性能需求。
通过SpringBoot3.4+ThreadPoolTaskExecutor，我们可以充分利用多线程并发处理，显著提升数据库写入性能。
在本次实验中，我们成功地将300万数据的插入时间从8.62分钟缩短到2.50分钟，多线程（30线程）耗时约：2.50分钟，单线程耗时约：8.62分钟。
此外，我们通过SQL语句检查数据完整性，确保所有数据均成功写入且无重复问题。由此可见，采用ThreadPoolTaskExecutor进行多线程优化是提升大数据量插入效率的有效方案，适用于日志存储、批量数据导入、业务数据初始化等场景。
优化
未来，我们可以进一步优化方案，例如：
动态调整线程池大小，以适应不同负载的插入任务。
异步批量提交事务，减少数据库锁竞争，提高吞吐量。
结合Kafka/RabbitMQ进行异步解耦，进一步优化数据处理架构。
总的来说，合理使用线程池技术，可以大幅度提升应用的性能，优化数据处理的效率，为系统带来显著的收益！
评论