MyBatis 批量操作深度优化——从 N+1 到批处理的全路径

MyBatis 批量操作深度优化——从 N+1 到批处理的全路径

一、从"功能正确"到"性能可接受"——MyBatis 批量操作的三段式进化

MyBatis 在日常增删改查场景中几乎是无感的——实体映射直观、SQL 控制灵活。但当数据量从千级上升到十万级、百万级，许多看似"正确"的写法会成为性能瓶颈。以下是一段真实经历的浓缩：

某订单导入功能，要求支持单次上传 10 万条订单明细。开发同学的实现方案是：循环遍历 Excel 解析出的每一行，调用OrderMapper.insert(order)，外层用@Transactional包裹。功能验证阶段（100 条）一切正常，产线验收阶段（10 万条）耗时 47 分钟，且数据库 CPU 持续飙升。

这个案例暴露了 MyBatis 批量操作中三个递进的问题层次：

1. N+1 问题：循环内逐条 SQL 执行，每次都有网络往返、SQL 解析、事务开销。
2. 忽略 JDBC 批处理能力：即使使用foreach拼接 SQL，也只是减少了网络往返次数，并未利用 JDBC 的addBatch()/executeBatch()机制。
3. 缺乏分片与流控意识：一次性提交 10 万条数据，undo log 膨胀、长事务锁持有时间过长。

本文将按"常规写法 →foreach批量 SQL → JDBC Batch → 分片流控"的路径，逐步展示每个阶段的优化原理和代码实现。

二、底层机制与原理深度剖析

flowchart LR subgraph Input["数据源"] DS[10 万条待入库数据] end subgraph Anti["反模式（请避免）"] L1["循环逐条 insert<br/>10 万次 DB 交互"] end subgraph Stage1["第一阶优化"] L2["foreach 批量 SQL<br/>10 次 DB 交互（每次 1 万条）"] end subgraph Stage2["第二阶优化"] L3["JDBC Batch + 分片<br/>利用 PreparedStatement 批量"] end subgraph Stage3["第三阶优化"] L4["分片 + 流控 + 异步<br/>可中断、可观测"] end DS --> Anti Anti -->|"网络往返：O(n)"| Stage1 Stage1 -->|"SQL 长度/事务膨胀"| Stage2 Stage2 -->|"单次不可中断"| Stage3

阶段分析：

逐条执行时，每条 INSERT 都是一次完整的 Client → DB 往返。以 RTT 0.5ms 计算，仅网络时间就达 50 秒（10 万 × 0.5ms），还不包括 SQL 解析、执行、事务提交的开销。

foreach批量拼接将多条 INSERT 合并为一条多 VALUES 的 SQL（INSERT INTO t VALUES (...), (...), (...)），将网络往返从 O(n) 降低到 O(n/batchSize)。但每批 1 万条意味着一条 SQL 的长度可能达到数 MB，数据库解析开销大，且出错时需要整体回滚。

JDBC Batch 通过PreparedStatement.addBatch()+executeBatch()在 JDBC 驱动层面批量发送，底层可能优化为合并的网络包，同时保持每条语句的独立性。以 MySQL 为例，需要在连接 URL 中加rewriteBatchedStatements=true参数才能将多条独立的 INSERT 合并为一条。

三、生产级代码实现与最佳实践

3.1 分片批处理核心实现

/** * 通用批量插入服务。 * 设计考量：不再使用 MyBatis foreach 做应用层 SQL 拼接， * 而是直接利用 JDBC Batch 能力，配合手动分片和事务控制， * 确保大批量操作可中断、可观测、可恢复。 */ @Service public class BatchInsertService { private final SqlSessionFactory sqlSessionFactory; /** 每批次大小：经实测，MySQL 单次 Batch 在 500~2000 条时性能最优 */ private static final int BATCH_SIZE = 1000; /** * 批量插入，支持进度回调。 * * @param items 待插入数据列表 * @param mapperClass Mapper 接口类 * @param progressCallback 进度回调（可用于日志、前端轮询） */ public <T> BatchResult batchInsert(List<T> items, Class<?> mapperClass, Consumer<Progress> progressCallback) { if (items == null || items.isEmpty()) { return new BatchResult(0, 0, 0); } int total = items.size(); int successCount = 0; int failCount = 0; List<Throwable> errors = new ArrayList<>(); // 手动分片，每片在一个独立事务中提交 for (int i = 0; i < total; i += BATCH_SIZE) { int end = Math.min(i + BATCH_SIZE, total); List<T> batch = items.subList(i, end); try { int affected = executeBatch(batch, mapperClass); successCount += affected; } catch (Exception e) { log.error("批量插入失败: 分片 [{}, {}), 错误: {}", i, end, e.getMessage()); failCount += (end - i); errors.add(e); // 失败后是否继续由调用方决定，这里记录但继续执行 // 如果需要严格一致性，可以将此处的 continue 改为 throw } // 回调进度 if (progressCallback != null) { progressCallback.accept(new Progress( Math.min(end, total), total, successCount, failCount)); } } return new BatchResult(successCount, failCount, errors.size()); } /** * 执行单批次 JDBC Batch 操作。 * 设计考量：使用 BATCH 执行器类型，MyBatis 会委托给 JDBC 的 * PreparedStatement.addBatch/executeBatch，性能优于逐条 execute。 * 事务粒度控制在单批次，避免长事务。 */ private <T> int executeBatch(List<T> batch, Class<?> mapperClass) { try (SqlSession sqlSession = sqlSessionFactory.openSession( ExecutorType.BATCH, false)) { // false 表示不自动提交，由我们手动控制事务边界 Object mapper = sqlSession.getMapper(mapperClass); Method insertMethod = findInsertMethod(mapperClass); for (T item : batch) { insertMethod.invoke(mapper, item); } // 执行批处理并提交事务 List<BatchResult> results = sqlSession.flushStatements(); sqlSession.commit(); // 汇总影响行数 int totalAffected = results.stream() .filter(r -> r instanceof org.apache.ibatis.executor.BatchResult) .flatMap(r -> ((org.apache.ibatis.executor.BatchResult) r) .getUpdateCounts().stream()) .mapToInt(Integer::intValue) .sum(); return totalAffected; } catch (Exception e) { throw new BatchExecutionException("批次执行失败", e); } } /** * 反射获取 Mapper 接口的 insert 方法。 * 设计考量：通过方法名约定而非硬编码类名，保持通用性。 */ private Method findInsertMethod(Class<?> mapperClass) { return Arrays.stream(mapperClass.getDeclaredMethods()) .filter(m -> "insert".equals(m.getName())) .findFirst() .orElseThrow(() -> new IllegalArgumentException( "Mapper 未找到 insert 方法: " + mapperClass.getName())); } // 内部类：进度信息 public record Progress(int completed, int total, int success, int failed) {} public record BatchResult(int success, int failed, int errorCount) {} }

3.2 批量更新与批量删除的优化

/** * 批量更新服务。 * 设计考量：UPDATE 场景与 INSERT 不同，MySQL 的 rewriteBatchedStatements * 只对 INSERT 有效。UPDATE 需要结合 CASE WHEN 语法实现真批量。 */ @Service public class BatchUpdateService { private final SqlSessionFactory sqlSessionFactory; /** * 使用 CASE WHEN 实现单条 SQL 批量更新。 * 适用于需要将多条记录更新为不同值的场景。 */ public <T> int batchUpdateWithCaseWhen(List<T> items, BiFunction<T, Map<String, Object>, Map<String, Object>> updateMapper) { if (items == null || items.isEmpty()) return 0; // 收集所有 ID 和各字段的目标值 List<Long> ids = new ArrayList<>(); Map<String, StringBuilder> fieldCases = new LinkedHashMap<>(); for (T item : items) { Map<String, Object> updates = updateMapper.apply(item, new HashMap<>()); Long id = (Long) updates.remove("id"); ids.add(id); for (Map.Entry<String, Object> entry : updates.entrySet()) { fieldCases.computeIfAbsent(entry.getKey(), k -> new StringBuilder()) .append(String.format("WHEN %d THEN '%s' ", id, entry.getValue())); } } // 构造 SQL StringBuilder sql = new StringBuilder("UPDATE target_table SET "); for (Map.Entry<String, StringBuilder> entry : fieldCases.entrySet()) { sql.append(String.format("%s = CASE id %s END, ", entry.getKey(), entry.getValue())); } sql.setLength(sql.length() - 2); // 去掉最后一个逗号 sql.append(" WHERE id IN (").append( ids.stream().map(String::valueOf) .collect(Collectors.joining(","))) .append(")"); try (SqlSession session = sqlSessionFactory.openSession(true)) { return session.getConnection() .createStatement() .executeUpdate(sql.toString()); } catch (SQLException e) { throw new BatchExecutionException("CASE WHEN 批量更新失败", e); } } }

3.3 避免 N+1 的关联查询优化

/** * 使用 MyBatis 的嵌套结果映射解决 N+1 关联查询。 * * 设计考量：N+1 问题在列表查询中尤为常见——先查主表获取 ID 列表， * 再逐条查关联表。解决方案是使用一条 JOIN SQL + resultMap 的 * collection 关联映射，一次查询带回所有数据。 */ // === Mapper XML 配置 === /* <resultMap id="orderWithItemsMap" type="Order"> <id property="id" column="order_id"/> <result property="orderNo" column="order_no"/> <!-- 嵌套结果映射：使用同一结果集的列 --> <collection property="items" ofType="OrderItem" resultMap="orderItemMap"/> </resultMap> <resultMap id="orderItemMap" type="OrderItem"> <id property="id" column="item_id"/> <result property="productName" column="product_name"/> <result property="quantity" column="quantity"/> </resultMap> <select id="findOrdersWithItems" resultMap="orderWithItemsMap"> SELECT o.id AS order_id, o.order_no, oi.id AS item_id, oi.product_name, oi.quantity FROM orders o LEFT JOIN order_items oi ON o.id = oi.order_id WHERE o.created_at >= #{startDate} ORDER BY o.id </select> */

3.4 MySQL 必须开启的参数

/** * 数据源配置：开启 MySQL JDBC 批处理重写。 * 设计考量：如果没有 rewriteBatchedStatements=true， * JDBC 驱动会将 batch 中的每条 INSERT 单独发送， * 批处理的性能优势大打折扣。 */ @Configuration public class DataSourceConfig { @Bean @ConfigurationProperties(prefix = "spring.datasource") public DataSource dataSource() { return DataSourceBuilder.create() .url("jdbc:mysql://localhost:3306/mydb?" + "rewriteBatchedStatements=true" // 核心：批量 INSERT 多 VALUES 合并 + "&useServerPrepStmts=false" // 关闭服务端预处理（批量场景非必要） + "&cachePrepStmts=true" // 缓存客户端 PreparedStatement + "&useSSL=false" // 内网环境可关闭 + "&allowPublicKeyRetrieval=true") // MySQL 8.0+ 认证 .build(); } }

四、边界分析与架构权衡

批量大小选择对照表：

foreach批量 SQL versus JDBC Batch：

ExecutorType.BATCH的隐形成本：

使用 BATCH 执行器时，MyBatis 不会在每次insert调用后立即返回自增主键。如果业务需要在插入后获取 ID，需要以下两种方案之一：

1. 使用selectKey提前生成 ID（如 Snowflake 分布式 ID），再执行插入。
2. 使用REUSE执行器（复用 PreparedStatement）配合手动flushStatements获取自增 ID。

连接池耗尽与大事务的连锁风险：

大批量操作中一个容易被忽视的问题是连接池耗尽。当批量任务在执行过程中占用了连接，而应用层的其他正常请求仍然需要从同一个连接池获取连接时，可能出现"池中连接被批处理占满、正常请求超时"的雪崩效应。对于这种情况，建议为批量操作使用独立的连接池或数据源，与常规 CRUD 操作进行物理隔离。以 HikariCP 为例，可以将批量数据源的maximumPoolSize控制在较小值（如 5），避免批处理无节制地抢占连接资源。

另一个需要关注的是分片失败后的幂等重试。在逐片提交的策略中，若第 3 片成功提交而第 4 片执行失败，直接从头重试会导致第 3 片数据重复插入。可行的解法有两条路径：一是业务侧引入唯一键约束（如唯一索引 +INSERT IGNORE或ON DUPLICATE KEY UPDATE），让数据库层面处理幂等；二是在应用层维护分片执行状态（已完成/未开始/失败），利用 Redis 或本地内存标记每个分片的执行结果，重试时只处理状态为"失败"或"未开始"的分片。两者的取舍在于：方案一实现成本低但对数据库有写入开销，方案二更精准但需要额外的状态管理逻辑。

五、总结

MyBatis 批量操作优化有清晰的台阶：

1. 第一阶段（避免 N+1）：使用 JOIN 查询或foreach批量拼接替代循环查询。
2. 第二阶段（JDBC Batch）：使用ExecutorType.BATCH替代foreach拼接 SQL，配合rewriteBatchedStatements=true发挥 MySQL 最佳性能。
3. 第三阶段（工程化）：加入分片控制、事务粒度管理、进度回调、异常隔离，确保大批量操作可观测、可恢复。

性能优化不是追求极致数值，而是在吞吐量、事务安全、可维护性之间找到适合当前场景的平衡点。对于大多数场景，本文推荐的 BATCH_SIZE=1000 + JDBC Batch + 手动事务控制的方案已经足够应对。