使用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);
}

// 注：分片新增任意一个事务操作失败将不对已提交事务产生任何影响，需自行保证数据在业务上的一致性