核心内容摘要
1秒处理1亿行，这8个Python库彻底让Excel消失
上周接了个数据迁移的活要把10万条数据从老系统导入新系统。
写了个简单的批量插入跑起来一看——5分钟。
领导说太慢了能不能快点折腾了一下午最后优化到3秒记录一下过程。
最初的代码5分钟最开始写的很简单foreach循环插入// 方式1循环单条插入最慢 for (User user : userList) { userMapper.insert(user); }10万条数据每条都要走一次网络请求、一次SQL解析、一次事务提交。
算一下假设每条插入需要3ms10万条就是300秒 5分钟。
这是最蠢的写法但我见过很多项目都这么写。
第一次优化批量SQL30秒把循环插入改成批量SQL!-- Mapper.xml -- insert idbatchInsert INSERT INTO user (name, age, email) VALUES foreach collectionlist itemitem separator, (#{item.name}, #{item.age}, #{item.email}) /foreach /insert// 分批插入每批1000条 int batchSize 1000; for (int i 0; i userList.size(); i batchSize) { int end Math.min(i batchSize, userList.size()); ListUser batch userList.subList(i, end); userMapper.batchInsert(batch); }从5分钟降到30秒提升10倍。
原理一条SQL插入多条数据减少网络往返次数。
但还有问题30秒还是太慢。
第二次优化JDBC批处理8秒MySQL有个参数叫rewriteBatchedStatements开启后可以把多条INSERT合并成一条。
第一步修改数据库连接URLjdbc:mysql://localhost:3306/test?rewriteBatchedStatementstrue第二步使用MyBatis的批处理模式Autowired private SqlSessionFactory sqlSessionFactory; public void batchInsertWithExecutor(ListUser userList) { try (SqlSession sqlSession sqlSessionFactory.openSession(ExecutorType.BATCH)) { UserMapper mapper sqlSession.getMapper(UserMapper.class); int batchSize 1000; for (int i 0; i userList.size(); i) { mapper.insert(userList.get(i)); if ((i
% batchSize
{ sqlSession.flushStatements(); sqlSession.clearCache(); } } sqlSession.flushStatements(); sqlSession.commit(); } }从30秒降到8秒。
原理ExecutorType.BATCH模式下MyBatis会缓存SQL最后一次性发送给数据库执行。
配合rewriteBatchedStatementstrueMySQL驱动会把多条INSERT合并。
第三次优化多线程并行3秒8秒还是不够快上多线程public void parallelBatchInsert(ListUser userList) { int threadCount 4; // 根据数据库连接池大小调整 int batchSize userList.size() / threadCount; ExecutorService executor Executors.newFixedThreadPool(threadCount); ListFuture? futures new ArrayList(); for (int i 0; i threadCount; i) { int start i * batchSize; int end (i threadCount -
? userList.size() : (i
* batchSize; ListUser subList userList.subList(start, end); futures.add(executor.submit(() - { batchInsertWithExecutor(subList); })); } // 等待所有任务完成 for (Future? future : futures) { try { future.get(); } catch (Exception e) { thrownew RuntimeException(e); } } executor.shutdown(); }从8秒降到3秒。
注意事项线程数不要超过数据库连接池大小如果需要事务一致性这个方案不适用要考虑主键冲突的问题优化效果对比方案耗时提升倍数循环单条插入300秒基准批量SQL30秒10倍JDBC批处理8秒37倍多线程并行3秒100倍踩过的坑坑1foreach拼接SQL过长foreach collectionlist itemitem separator,如果一次插入太多条SQL会非常长可能超过max_allowed_packet限制。
解决分批插入每批
条。
坑2rewriteBatchedStatements不生效检查几个点URL参数是否正确rewriteBatchedStatementstrue是否使用了ExecutorType.BATCHMySQL驱动版本是否太旧坑3自增主键返回问题批量插入时想获取自增主键insert idbatchInsert useGeneratedKeystrue keyPropertyid注意rewriteBatchedStatementstrue时自增主键返回可能有问题需要升级MySQL驱动到
8.
17。
坑4内存溢出10万条数据一次性加载到内存可能OOM。
解决分页读取 分批插入。
int pageSize 10000; int total countTotal(); for (int i 0; i total; i pageSize) { ListUser page selectByPage(i, pageSize); batchInsertWithExecutor(page); }最终方案代码Service publicclass BatchInsertService { Autowired private SqlSessionFactory sqlSessionFactory; /** * 高性能批量插入 * 10万条数据约3秒 */ public void highPerformanceBatchInsert(ListUser userList) { if (userList null || userList.isEmpty()) { return; } int threadCount Math.min(4, Runtime.getRuntime().availableProcessors()); int batchSize (int) Math.ceil((double) userList.size() / threadCount); ExecutorService executor Executors.newFixedThreadPool(threadCount); CountDownLatch latch new CountDownLatch(threadCount); for (int i 0; i threadCount; i) { int start i * batchSize; int end Math.min((i