MyBatis-Plus saveBatch异步事务问题分析与解决方案

1. 问题背景与现象描述最近在开发一个电商订单处理系统时遇到了一个棘手的问题使用MyBatis-Plus的saveBatch方法在异步线程中批量插入数据时发现事务没有正常提交。具体表现为系统使用Spring Boot MyBatis-Plus架构订单创建后需要异步处理库存扣减和日志记录使用Async注解标记的异步方法中调用了saveBatch批量插入操作日志日志表中有部分记录插入成功部分记录丢失没有抛出任何异常但数据不完整这个问题在测试环境偶发出现但在高并发压测时几乎必现。经过排查发现这与MyBatis-Plus的批量操作机制、Spring事务管理以及异步线程处理有密切关系。2. MyBatis-Plus saveBatch原理剖析2.1 saveBatch的默认实现MyBatis-Plus的saveBatch方法默认实现是这样的// MyBatis-Plus 3.x版本的默认实现 Transactional(rollbackFor Exception.class) Override public boolean saveBatch(CollectionT entityList, int batchSize) { String sqlStatement sqlStatement(SqlMethod.INSERT_ONE); return executeBatch(entityList, batchSize, (sqlSession, entity) - { sqlSession.insert(sqlStatement, entity); }); }关键点方法本身带有Transactional注解使用MyBatis的SqlSession执行批量插入默认batchSize为10002.2 批量操作的执行流程saveBatch的实际执行流程可以分为以下几个步骤开启事务由Spring管理对集合进行分片处理根据batchSize对每个分片执行批量插入提交事务如果成功或回滚如果失败问题在于当这个方法在异步线程中执行时第4步的事务提交可能不会按预期工作。3. 异步环境中的事务问题3.1 Spring事务管理机制Spring的事务管理是基于ThreadLocal实现的关键点包括事务上下文存储在ThreadLocal中Transactional注解的事务传播行为默认是REQUIRED异步方法会使用新的线程执行无法继承原有的事务上下文3.2 Async与事务的交互当我们在异步方法中使用Transactional时会遇到以下问题异步方法本身需要Async注解如果异步方法内部有Transactional会创建新的事务这两个注解的执行顺序和交互需要特别注意3.3 典型的问题场景在我们的案例中代码结构大致如下Service public class OrderService { Autowired private AsyncLogService asyncLogService; Transactional public void createOrder(OrderDTO dto) { // 订单创建逻辑... asyncLogService.saveOperationLog(logs); } } Service public class AsyncLogService { Async Transactional(propagation Propagation.REQUIRES_NEW) public void saveOperationLog(ListOperationLog logs) { logMapper.saveBatch(logs); // 使用MyBatis-Plus的saveBatch } }这种情况下虽然两个方法都有Transactional注解但由于异步执行事务可能无法正常提交。4. 问题排查过程4.1 复现问题为了准确复现问题我们设计了以下测试方案准备1000条测试数据在异步方法中调用saveBatch观察数据库中的记录数量检查日志是否有异常测试结果有时插入全部成功有时部分成功如插入300条没有异常日志4.2 日志分析通过增加事务相关的日志配置logging.level.org.springframework.transactionDEBUG logging.level.org.mybatisTRACE从日志中可以观察到主线程的事务正常开启和提交异步线程中的事务有时没有提交日志没有回滚日志4.3 线程池配置检查发现项目中配置了自定义的线程池Configuration EnableAsync public class AsyncConfig implements AsyncConfigurer { Override public Executor getAsyncExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(5); executor.setMaxPoolSize(10); executor.setQueueCapacity(100); executor.setThreadNamePrefix(async-); executor.initialize(); return executor; } }线程池配置可能导致任务被拒绝或线程被回收影响事务提交。5. 解决方案5.1 方案一使用编程式事务管理改造异步方法使用TransactionTemplateService public class AsyncLogService { Autowired private TransactionTemplate transactionTemplate; Async public void saveOperationLog(ListOperationLog logs) { transactionTemplate.execute(status - { try { logMapper.saveBatch(logs); return Boolean.TRUE; } catch (Exception e) { status.setRollbackOnly(); throw e; } }); } }优点明确控制事务边界避免注解方式的问题缺点代码稍显冗长5.2 方案二调整事务传播行为修改Transactional的传播行为Async Transactional(propagation Propagation.NESTED) public void saveOperationLog(ListOperationLog logs) { logMapper.saveBatch(logs); }注意NESTED需要数据库支持保存点不是所有场景都适用5.3 方案三使用同步批量插入如果不必须异步可以改为同步执行Service public class OrderService { Transactional public void createOrder(OrderDTO dto) { // 订单创建逻辑... logMapper.saveBatch(logs); // 同步执行 } }最简单可靠但可能影响性能。5.4 最终采用的方案我们最终选择了方案一结合以下优化增加事务超时设置添加重试机制完善日志记录完整实现Async public void saveOperationLog(ListOperationLog logs) { transactionTemplate.setTimeout(30); // 30秒超时 transactionTemplate.execute(status - { try { int retryCount 0; while (retryCount 3) { try { logMapper.saveBatch(logs); return Boolean.TRUE; } catch (Exception e) { retryCount; if (retryCount 3) { throw e; } Thread.sleep(1000 * retryCount); } } return Boolean.FALSE; } catch (Exception e) { log.error(保存操作日志失败, e); status.setRollbackOnly(); throw new RuntimeException(保存操作日志失败, e); } }); }6. 深入原理为什么会出现这个问题6.1 MyBatis-Plus批量操作的本质虽然叫批量插入但默认实现其实是循环单条插入不是真正的JDBC批量addBatch/executeBatch每条insert都是独立的SQL语句依赖事务保证原子性6.2 Spring异步执行的原理Async的工作机制通过AOP代理拦截方法调用提交到线程池执行原始线程继续执行新线程中方法执行6.3 事务失效的根本原因综合来看问题根源在于异步线程可能被突然终止如线程池回收事务提交发生在异步线程中Spring无法保证异步线程的事务一定会提交MyBatis-Plus的批量不是原子操作7. 性能优化建议7.1 使用真正的批量插入可以重写saveBatch方法使用JDBC的批量操作public class CustomServiceImplM extends BaseMapperT, T extends ServiceImplM, T { Override Transactional public boolean saveBatch(CollectionT entityList, int batchSize) { try (SqlSession batchSqlSession sqlSessionBatch()) { int i 0; for (T entity : entityList) { batchSqlSession.insert(sqlStatement(SqlMethod.INSERT_ONE), entity); if (i 1 i % batchSize 0) { batchSqlSession.flushStatements(); } i; } batchSqlSession.flushStatements(); return true; } } }7.2 调整批量大小根据数据库性能调整batchSizeMySQL建议500-1000Oracle建议100-200SQL Server建议1000-20007.3 使用多线程批量插入对于大数据量可以结合多线程public void batchInsertConcurrent(ListData dataList) { int threadCount 4; int batchSize dataList.size() / threadCount; ExecutorService executor Executors.newFixedThreadPool(threadCount); ListFuture? futures new ArrayList(); for (int i 0; i threadCount; i) { int from i * batchSize; int to (i threadCount - 1) ? dataList.size() : (i 1) * batchSize; ListData subList dataList.subList(from, to); futures.add(executor.submit(() - { transactionTemplate.execute(status - { customService.saveBatch(subList); return null; }); })); } for (Future? future : futures) { try { future.get(); } catch (Exception e) { // 处理异常 } } executor.shutdown(); }8. 其他注意事项8.1 事务隔离级别的影响在高并发下还需要考虑隔离级别READ_COMMITTED可能导致幻读SERIALIZABLE性能影响大建议根据业务需求选择合适的隔离级别8.2 连接池配置确保连接池配置合理足够大的最大连接数合理的超时设置适当的验证查询例如HikariCP配置spring.datasource.hikari.maximum-pool-size20 spring.datasource.hikari.connection-timeout30000 spring.datasource.hikari.validation-timeout5000 spring.datasource.hikari.leak-detection-threshold600008.3 监控与告警建议添加以下监控事务执行时间监控批量操作成功率监控线程池使用情况监控可以使用Micrometer Prometheus Grafana实现。9. 常见问题解答9.1 为什么部分数据插入成功了这是因为MyBatis-Plus的saveBatch默认不是原子操作。在事务提交前部分插入已经执行但如果事务最终没有提交这些已执行的插入可能会被保留取决于数据库的具体实现。9.2 如何确定是事务问题可以通过以下方法验证在方法结束后手动抛出异常看是否回滚检查数据库事务日志使用Spring的TransactionSynchronizationManager.isActualTransactionActive()9.3 除了saveBatch还有其他方法吗可以考虑使用MyBatis的 标签实现批量插入使用JDBC的addBatch/executeBatch使用存储过程处理批量数据9.4 异步事务的最佳实践是什么建议避免在异步方法中进行复杂的多步骤事务如果必须使用事务确保有完善的错误处理和重试机制考虑使用消息队列实现最终一致性监控异步任务的执行情况10. 总结与个人建议经过这次问题排查我总结了以下几点经验不要想当然地认为批量操作就是原子的要了解框架的具体实现异步和事务结合使用时需要格外小心生产环境中的事务问题往往在高压下才会暴露完善的日志和监控是快速定位问题的关键在实际项目中我建议对于关键业务操作优先考虑同步执行如果必须异步考虑使用消息队列等更可靠的机制对批量操作进行充分的压力测试编写详细的文档记录这些坑避免团队成员重复踩坑最后记住一个原则分布式系统没有完美的事务解决方案我们需要根据业务特点在一致性和性能之间找到平衡点。