当前位置：首页 > news >正文

java 线程池 + 分批次批量插入代码

news 2026/4/24 9:26:48

线程池 + 分批次批量插入代码

import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.util.ArrayList; import java.util.List; import java.util.concurrent.*; /** * 批量导入工具类 * 线程池分批插入 + 安全等待 */ @Component public class UserBatchInsert { @Resource private UserMapper userMapper; // 核心线程数：根据服务器CPU核心数设置（推荐 CPU核心数*2） private static final int CORE_THREAD_NUM = 8; // 线程池拒绝策略：调用者线程执行，保证任务不丢失 private static final RejectedExecutionHandler HANDLER = new ThreadPoolExecutor.CallerRunsPolicy(); /** * 线程池分批批量插入 * @param data 总数据集合 * @param batchNum 每批次插入条数（推荐 500~1000） * @throws InterruptedException 中断异常 */ public void batchInsertUser(List<User> data, int batchNum) throws InterruptedException { // 1. 参数校验 if (data == null || data.isEmpty()) { return; } if (batchNum <= 0) { batchNum = 500; // 默认批次 } int totalSize = data.size(); // 计算总批次 int batchCount = (totalSize + batchNum - 1) / batchNum; // 自定义线程池（固定核心线程，拒绝OOM） ThreadPoolExecutor executor = new ThreadPoolExecutor( CORE_THREAD_NUM, CORE_THREAD_NUM, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), HANDLER ); CountDownLatch countDownLatch = new CountDownLatch(batchCount); try { for (int i = 0; i < batchCount; i++) { int start = i * batchNum; int end = Math.min(start + batchNum, totalSize); // 关键：subList 转为新集合，避免并发异常 List<User> subList = new ArrayList<>(data.subList(start, end)); // 提交任务 executor.execute(new InsertTask(subList, countDownLatch)); } // 等待所有批次插入完成 countDownLatch.await(); } finally { // 优雅关闭线程池 executor.shutdown(); if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { executor.shutdownNow(); } } } /** * 静态内部类：插入任务 * 必须加 static，避免外部类引用泄漏 */ static class InsertTask implements Runnable { private final List<User> userList; private final CountDownLatch countDownLatch; public InsertTask(List<User> userList, CountDownLatch countDownLatch) { this.userList = userList; this.countDownLatch = countDownLatch; } @Override public void run() { try { // 执行业务：批量插入 if (userList != null && !userList.isEmpty()) { // 调用MyBatis批量插入 userList.forEach(System.out::println); // userMapper.insertUserList(userList); } } catch (Exception e) { // 捕获异常，避免任务失败导致计数器不执行 e.printStackTrace(); } finally { // 必须放在finally：无论成功/失败，都计数-1 countDownLatch.countDown(); } } } }

Mapper 批量插入接口

import org.apache.ibatis.annotations.Param; import java.util.List; public interface UserMapper { /** * 批量新增用户 */ void insertUserList(@Param("list") List<User> list); }

XML SQL（MyBatis）

<insert id="insertUserList"> INSERT INTO user(id,name,phone) VALUES <foreach collection="list" item="item" separator=","> (#{item.id},#{item.name},#{item.phone}) </foreach> </insert>

使用示例

@Autowired private UserBatchInsert userBatchInsert; public void test() throws InterruptedException { // 模拟10000条数据 List<User> userList = new ArrayList<>(); // 每500条一批，多线程插入 userBatchInsert.batchInsertUser(userList, 500); }