使用Mybatis批量插入大量数据的实践

前言

在项目开发过程中，我们经常会有批量插入的需求，例如：定时统计任务

但是受限于MySQL中 max_allowed_packet 参数限制，5.7版本默认值为4M，这显然不太符合我们的需求，当然我们也可以通过修改此值来适应我们业务，今天分享在不修改此值的情况下，如何在客户端优雅的处理此场景

常用基础版

@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
public void submit(List<GroupDayStatistics> dataList) {// 清理统计数据mapper.delete();int batchNum = 1000;List<Object> tempList = new ArrayList<>(batchNum);for (int i = 0, size = dataList.size(); i < size; i++) {tempList.add(dataList.get(i));if (tempList.size() == 1000 || i + 1 == size) {mapper.batchInsert(tempList);tempList.clear();}}
}

这种写法有个缺点：其他地方要用都得copy一份过去，麻烦

改进版

@Service
public class BatchInsertUtil<T> {/*** 批次新增(删除函数 以及 分片新增都将在同一个事务中进行处理)** @param batchSize*            批次新增大小* @param deleteSupplier*            清理函数* @param insertFunction*            批次新增调用函数* @param data*            需新增数据* @return left：删除影响行 right：新增行数*/@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)public ImmutablePair<Integer, Integer> batchInsert(int batchSize, IntSupplier deleteSupplier,Function<List<T>, Integer> insertFunction, List<T> data) {int deleteRow = 0;// 清理数据if (Objects.nonNull(deleteSupplier)) {deleteRow = deleteSupplier.getAsInt();}// 分片分事务插入数据List<List<T>> partition = ListUtils.partition(data, batchSize);int insertRow = partition.stream().mapToInt(insertFunction::apply).sum();return new ImmutablePair<>(deleteRow, insertRow);}
}

使用第一步

// 注入我们的工具类
@Resource
private BatchInsertUtil<需要插入的实体类> batchInsertUtil;

使用第二步

IntSupplier deleteSupplier = () -> mapper.delete();
Function<List<需要插入的实体类>, Integer> insertFunction = arg -> mapper.batchInsert(arg);
batchInsertUtil.batchInsert(1000, deleteSupplier, insertFunction, dataList);

本来到这就结束了，问题很多的小明就说了：

你这个怎么删除与分片新增都在一个事务中，如果我一次性插入的数据过多，这不就是一个大事务了嘛？虽然在一个事务中可以保证原子性，但是我有的场景就是想要他们分别处于不同事务，业务上的一致性我自己保证，你就说能不能做吧！

......

......

......

安排！

进一步优化

// 这里我们在第二步的基础上引入 编程式事务：
@Resource
private PlatformTransactionManager transactionManager;/*** 批次新增（删除函数 以及 分片新增函数都将采取独立事务commit）** @param batchSize*            批次新增大小* @param deleteSupplier*            清理函数* @param insertFunction*            批次新增调用函数* @param data*            需新增数据* @return left：删除影响行 right：新增行数*/
@Transactional(rollbackFor = Exception.class, propagation = Propagation.NOT_SUPPORTED)
public ImmutablePair<Integer, Integer> batchInsertAloneTransaction(int batchSize, IntSupplier deleteSupplier,Function<List<T>, Integer> insertFunction, List<T> data) {TransactionStatus transactionStatus = buildTransactionStatus();int deleteRow = 0;try {// 清理数据if (Objects.nonNull(deleteSupplier)) {deleteRow = deleteSupplier.getAsInt();}transactionManager.commit(transactionStatus);} catch (Exception e) {transactionManager.rollback(transactionStatus);}// 分片分事务插入数据List<List<T>> partition = ListUtils.partition(data, batchSize);int insertRow = partition.stream().mapToInt(list -> insertAndCommit(insertFunction, list)).sum();return new ImmutablePair<>(deleteRow, insertRow);
}/*** 新增分片数据并提交事务* * @param insertFunction*            新增函数* @param dataList*            分片数据* @return 新增行数*/
private int insertAndCommit(Function<List<T>, Integer> insertFunction, List<T> dataList) {TransactionStatus transactionStatus = buildTransactionStatus();try {Integer apply = insertFunction.apply(dataList);transactionManager.commit(transactionStatus);return apply;} catch (Exception e) {transactionManager.rollback(transactionStatus);throw e;}
}/*** 构建事务状态* * @return 结果*/
private TransactionStatus buildTransactionStatus() {DefaultTransactionDefinition defaultTransactionDefinition = new DefaultTransactionDefinition();defaultTransactionDefinition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);return transactionManager.getTransaction(defaultTransactionDefinition);
}// 注：分片新增任意一个事务操作失败将不对已提交事务产生任何影响，需自行保证数据在业务上的一致性