从 7 分钟到 10 秒，Mybatis 批处理真的很强！

点击上方“码农突围”，马上关注
这里是码农充电第一站，回复“666”，获取一份专属大礼包
真爱，请设置“星标”或点个“在看”

来源：juejin.cn/post/7078237987011559460

问题原因

简单了解一下批处理背后的秘密，BatchExecutor

每次向数据库发送的 SQL 语句的条数是有上限的，如果批量执行的时候超过这个上限值，数据库就会抛出异常，拒绝执行这一批 SQL 语句，所以我们需要控制批量发送 SQL 语句的条数和频率。

版本1-呱呱坠地

@Resource
private 某Mapper类 mapper实例对象;

private int BATCH = 1000;


  private void doUpdateBatch(Date accountDate, List<某实体类> data) {
    SqlSession batchSqlSession = null;
    try {
      if (data == null || data.size() == 0) {
        return;
      }
      batchSqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
      for (int index = 0; index < data.size(); index++) {
        mapper实例对象.更新/插入Method(accountDate, data.get(index).getOrderNo());
        if (index != 0 && index % BATCH == 0) {
          batchSqlSession.commit();
          batchSqlSession.clearCache();
        }
      }
      batchSqlSession.commit();
    } catch (Exception e) {
      batchSqlSession.rollback();
      log.error(e.getMessage(), e);
    } finally {
      if (batchSqlSession != null) {
        batchSqlSession.close();
      }
    }
  }

你真的懂commit、clearCache、flushStatements嘛？

  @Override
  public void commit() {
    commit(false);
  }  

  @Override
  public void commit(boolean force) {
    try {
      executor.commit(isCommitOrRollbackRequired(force));
      dirty = false;
    } catch (Exception e) {
      throw ExceptionFactory.wrapException("Error committing transaction.  Cause: " + e, e);
    } finally {
      ErrorContext.instance().reset();
    }
  }

  private boolean isCommitOrRollbackRequired(boolean force) {
    // autoCommit默认为false，调用过插入、更新、删除之后的dirty值为true
    return (!autoCommit && dirty) || force;
  }

  @Override
  public void commit(boolean required) throws SQLException {
    if (closed) {
      throw new ExecutorException("Cannot commit, transaction is already closed");
    }
    clearLocalCache();
    flushStatements();
    if (required) {
      transaction.commit();
    }
  }

反例

不具备通用性

版本2-初具雏形

import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.function.ToIntFunction;

@Slf4j
@Component
public class MybatisBatchUtils {

    /**
     * 每次处理1000条
     */
    private static final int BATCH = 1000;

    @Resource
    private SqlSessionFactory sqlSessionFactory;

    /**
     * 批量处理修改或者插入
     *
     * @param data     需要被处理的数据
     * @param function 自定义处理逻辑
     * @return int 影响的总行数
     */
    public   int batchUpdateOrInsert(List data, ToIntFunction function) {
        int count = 0;
        SqlSession batchSqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH);
        try {
            for (int index = 0; index < data.size(); index++) {
                count += function.applyAsInt(data.get(index));
                if (index != 0 && index % BATCH == 0) {
                    batchSqlSession.flushStatements();
                }
            }
            batchSqlSession.commit();
        } catch (Exception e) {
            batchSqlSession.rollback();
            log.error(e.getMessage(), e);
        } finally {
            batchSqlSession.close();
        }
        return count;
    }
}

@Resource
private 某Mapper类 mapper实例对象;

batchUtils.batchUpdateOrInsert(数据集合, item -> mapper实例对象.insert方法(item));

版本3-标准写法

  public Executor newExecutor(Transaction transaction, ExecutorType executorType) {
    executorType = executorType == null ? defaultExecutorType : executorType;
    executorType = executorType == null ? ExecutorType.SIMPLE : executorType;
    Executor executor;
    if (ExecutorType.BATCH == executorType) {
      executor = new BatchExecutor(this, transaction);
    } else if (ExecutorType.REUSE == executorType) {
      executor = new ReuseExecutor(this, transaction);
    } else {
      executor = new SimpleExecutor(this, transaction);
    }
    if (cacheEnabled) {
      executor = new CachingExecutor(executor);
    }
    executor = (Executor) interceptorChain.pluginAll(executor);
    return executor;
  }

public class MybatisBatchUtils {
    
    /**
    * 每次处理1000条
    */
    private static final int BATCH_SIZE = 1000;
    
    @Resource
    private SqlSessionFactory sqlSessionFactory;
    
    /**
    * 批量处理修改或者插入
    *
    * @param data     需要被处理的数据
    * @param mapperClass  Mybatis的Mapper类
    * @param function 自定义处理逻辑
    * @return int 影响的总行数
    */
    public   int batchUpdateOrInsert(List data, Class mapperClass, BiFunction function) {
        int i = 1;
        SqlSession batchSqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH);
        try {
            U mapper = batchSqlSession.getMapper(mapperClass);
            int size = data.size();
            for (T element : data) {
                function.apply(element,mapper);
                if ((i % BATCH_SIZE == 0) || i == size) {
                    batchSqlSession.flushStatements();
                }
                i++;
            }
            // 非事务环境下强制commit，事务情况下该commit相当于无效
            batchSqlSession.commit(!TransactionSynchronizationManager.isSynchronizationActive());
        } catch (Exception e) {
            batchSqlSession.rollback();
            throw new CustomException(e);
        } finally {
            batchSqlSession.close();
        }
        return i - 1;
    }
}

batchUtils.batchUpdateOrInsert(数据集合, xxxxx.class, (item, mapper实例对象) -> mapper实例对象.insert方法(item));

附：Oracle批量插入优化

<selectKey keyProperty="id" order="BEFORE" resultType="java.lang.Long">
  select XXX.nextval from dual
selectKey>

<insert id="insert" parameterType="user">
        insert into table_name(id, username, password)
        values(SEQ_USER.NEXTVAL,#{username},#{password})
insert>

（完）
码农突围资料链接
1、卧槽！字节跳动《算法中文手册》火了，完整版 PDF 开放下载！
2、计算机基础知识总结与操作系统 PDF 下载
3、艾玛，终于来了！《LeetCode Java版题解》.PDF
4、Github 10K+，《LeetCode刷题C/C++版答案》出炉.PDF
欢迎添加鱼哥个人微信：smartfish2020，进粉丝群或围观朋友圈