首页 > 代码库 > ibatis批量执行分析

ibatis批量执行分析

2024-10-23 12:05:39   211人阅读

最近做pos数据文件解析及入库的开发，其中pos的流水文件一个文件中就包含8000多条数据，每次插入数据库执行的sql都是相同的。因此考虑到使用批量插入来提升效率。查看ibatis的文档，看到提供了startBatch和executBatch两个方法，看名字大概就知道这两个方法和批量执行有关。我立马在之前的for循环外面加上了这两个方法：

sqlMap.getSqlMapClient().startBatch();
for (Map<String, Object> map : list) {
    sqlMap.insert(sqlId, map);
}
sqlMap.getSqlMapClient().executeBatch();
执行之后发现，效率跟没加一样，毫无提升。到网上查阅资料说是要在这外边再加一层事务处理代码才行。
技术分享
或者是使用spring提供的SqlMapClientTemplate中的方法来执行：
sqlMap.execute(new SqlMapClientCallback() {
    @Override
    public Object doInSqlMapClient(SqlMapExecutor executor) throws SQLException {
       
        int len = list.size();
        executor.startBatch();
       
        for(int i = 0; i < len; i++){
            executor.insert(test, list.get(i));
            if(i != 0 && i%500 == 0){
                executor.executeBatch();
                executor.startBatch();
            }
        }
       
        executor.executeBatch();
        return null;
    }
});
 
针对ibatis的批量操作必须加事务的做法感到有点奇怪(虽然JDBC批量操作一般会加事务，但不是必须加)。于是大致的研究了一下ibatis批量操作有关的代码。
涉及到的类大概有如下几个：
SqlMapClientImpl ：SqlMapClient的实现类
SqlMapSessionImpl ：SqlMapSession的实现类
SqlMapExecutorDelegate ：代理SqlMapSession中的数据库操作
MappedStatement ：映射的sql语句配置对象
SqlExecutor：真正执行sql语句
SessionScope ：SqlMapSession中相关的设置信息
StatementScope ：Statemente中相关的设置信息
 
需要明确的是SqlMapClient中的大部分方法实际上是调用了SqlMapSession和SqlMapExecutorDelegate的方法，而SqlMapSession中的一些方法实际还是调用了SqlMapExecutorDelegate的方法，也就是说大部分我们操作的方法都实际有delegate来完成的。
直接与我们打交道的是SqlMapClient，因此首先来看SqlMapClientImpl的startBatch方法：
public void startBatch() throws SQLException {
    getLocalSqlMapSession().startBatch();
}
public int executeBatch() throws SQLException {
    return getLocalSqlMapSession().executeBatch();
}
protected SqlMapSessionImpl getLocalSqlMapSession() {
    SqlMapSessionImpl sqlMapSession = (SqlMapSessionImpl)this.localSqlMapSession.get();
    if ((sqlMapSession == null) || (sqlMapSession.isClosed())) {
      sqlMapSession = new SqlMapSessionImpl(this);
      this.localSqlMapSession.set(sqlMapSession);
    }
    return sqlMapSession;
}
可以看到这两个方法实际上都委派给了SqlMapSession来操作的，那么接下来的线索就在SqlMapSessionImpl类中了：
public void startBatch() throws SQLException {
    this.delegate.startBatch(this.sessionScope);
}
public int executeBatch() throws SQLException {
    return this.delegate.executeBatch(this.sessionScope);
}
这里又可以看到这两个操作又委托给了delegate(SqlMapExecutorDelegate)来执行，最重要的操作都在delegate中了，这里还涉及到了SessionScope，它用于传递一些配置信息。
public void startBatch(SessionScope sessionScope)
{
    sessionScope.setInBatch(true);
}
public int executeBatch(SessionScope sessionScope)
throws SQLException
{
    sessionScope.setInBatch(false);
    return this.sqlExecutor.executeBatch(sessionScope);
}
startBatch方法仅仅修改了SessionScope中的inBatch属性的值，executeBatch也会重置SessionScope中inBatch的值，并且调用sqlExecutor的executeBatche方法来执行批量操作，注意这里传递的参数只有sessionScope，那么也就说明所有批量操作有关的信息都放置在SessionScope中。这里我们先来看sqlExecutor中是怎么执行这些批量操作的。
public int executeBatch(SessionScope sessionScope)
    throws SQLException
{
    int rows = 0;
    Batch batch = (Batch)sessionScope.getBatch();
    if (batch != null) {
      try {
        rows = batch.executeBatch();
      } finally {
        batch.cleanupBatch(sessionScope);
      }
    }
    return rows;
}
这里可以看到，批量操作的信息是存放在sessionScope的batch属性(Object类型)中的。是一个Batch类型的变量，获取到之后直接调用Batch的executeBatch方法来完成批量操作的执行，然后再做一些清理操作。
public int executeBatch()
      throws SQLException
{
  int totalRowCount = 0;
  int i = 0; for (int n = this.statementList.size(); i < n; ++i) {
    PreparedStatement ps = (PreparedStatement)this.statementList.get(i);
    int[] rowCounts = ps.executeBatch();
    for (int j = 0; j < rowCounts.length; ++j) {
      if (rowCounts[j] == -2)
        continue;
      if (rowCounts[j] == -3) {
        throw new SQLException("The batched statement at index " + j + " failed to execute.");
      }
      totalRowCount += rowCounts[j];
    }
  }
  return totalRowCount;
}
这里就可以看到底层实际还是使用了JDBC的PreparedStatement来实现批量操作。这里有个statementList对象，里面存放了一些PreparedStatement，一个PreparedStatement对象就对应了同一个sql的批量操作处理。也就是说ibatis的批量操作中还支持多个不同sql的批量操作。
 
到这里，批量操作的执行过程基本就分析完了，接下来就要分析，怎么将这些PreparedStatement对象放置到Batch的statementList中去的。
 
通过文章最前面的代码，可以看到在startBatch和executeBatch之前执行的全部是insert操作，那么PreparedStatement对象肯定就是在insert方法执行的过程中放置到Batch对象中去的。虽然调用的SqlMapClient的insert,但实际上会执行delegate的insert方法，因此我们直接看SqlMapExecutorDelegate的insert方法：
public Object insert(SessionScope sessionScope, String id, Object param)
    throws SQLException
{
    Object generatedKey = null;

    MappedStatement ms = getMappedStatement(id); //获取映射的sql配置信息
    Transaction trans = getTransaction(sessionScope);//获取当前session的事务(SessionScope在一个session中唯一)
    boolean autoStart = trans == null;//判断事务是否为空
    try
    {
      trans = autoStartTransaction(sessionScope, autoStart, trans);//为空则使用自动事务

      SelectKeyStatement selectKeyStatement = null;
      if (ms instanceof InsertStatement) {
        selectKeyStatement = ((InsertStatement)ms).getSelectKeyStatement();
      }

      Object oldKeyValue = null;
      String keyProperty = null;
      boolean resetKeyValueOnFailure = false;
      if ((selectKeyStatement != null) && (!(selectKeyStatement.isRunAfterSQL()))) {
        keyProperty = selectKeyStatement.getKeyProperty();
        oldKeyValue = PROBE.getObject(param, keyProperty);
        generatedKey = executeSelectKey(sessionScope, trans, ms, param);
        resetKeyValueOnFailure = true;
      }

      StatementScope statementScope = beginStatementScope(sessionScope, ms);//生成StatementScope信息，其中包含sessionScope对象
      try {
        ms.executeUpdate(statementScope, trans, param);//使用MappedStatement对象执行，批量操作处理在这里实现
      }
      catch (SQLException e)
      {
        if (resetKeyValueOnFailure);
        throw e;
      } finally {
        endStatementScope(statementScope);
      }

      if ((selectKeyStatement != null) && (selectKeyStatement.isRunAfterSQL())) {
        generatedKey = executeSelectKey(sessionScope, trans, ms, param);
      }
      //注意这里，相当关键。如果是使用自动事务，那么会自动提交事务
      autoCommitTransaction(sessionScope, autoStart);
    } finally {
      autoEndTransaction(sessionScope, autoStart);
    }

    return generatedKey;
}
这里相当关键的地方就是自动事务的处理，批处理的和正常操作的区处理在MappedStatement中的executeUpdate方法中实现，这其中由调用了另外一个名为sqlExecuteUpdate的方法。
rows = sqlExecuteUpdate(statementScope, trans.getConnection(), sqlString, parameters);
这个方法中判断是执行批量操作还是普通操作：
protected int sqlExecuteUpdate(StatementScope statementScope, Connection conn, String sqlString, Object[] parameters) throws SQLException {
    if (statementScope.getSession().isInBatch()) {
      getSqlExecutor().addBatch(statementScope, conn, sqlString, parameters);
      return 0;
    }
    return getSqlExecutor().executeUpdate(statementScope, conn, sqlString, parameters);
}
这里就使用到了之前在startBatch中向SessionScope中设置的inBatch属性，如果当前是在执行批处理操作，那么就调用sqlExecutor的addBatch方法，如果是普通操作，那么就调用sqlExecutor的executeUpdate来直接执行。
Batch是SqlExecutor的内部类，在SqlExecutor的addBatch中会从SessionScope中获取batch属性，如果为空，则创建一个，并且设置到SessionScope中，然后在调用Batch对象的addBatch方法：
public void addBatch(StatementScope statementScope, Connection conn, String sql, Object[] parameters)
      throws SQLException
{
      PreparedStatement ps = null;
      if ((currentSql != null) && (currentSql.equals(sql)))
      {
        int last = statementList.size() - 1;
        ps = (PreparedStatement)statementList.get(last);
      }
      else
      {
        ps = SqlExecutor.prepareStatement(statementScope.getSession(), conn, sql);
        SqlExecutor.setStatementTimeout(statementScope.getStatement(), ps);
        currentSql = sql;
        statementList.add(ps);
        batchResultList.add(new BatchResult(statementScope.getStatement().getId(), sql));
      }
      statementScope.getParameterMap().setParameters(statementScope, ps, parameters);
      ps.addBatch();
      size += 1;
}
在addBatch方法中，就回归到了JDBC的PreparedStatement，将PreparedStatement对象加入到了Batch的statementList中，以供后面的executeBatch来调用。
 
按理说到这里为止，看上去没什么问题，思路也比较清晰，但是为什么我加了startBatch和executeBatch之后效率没有提升呢。我们还得回到前面SqlMapExecutorDelegate类中的insert方法中去，前面已经分析过了，如果insert操作发现没有事务，那么就会使用自动事务，会创建一个事务对象，设置到sessionScope中去，在insert方法的最后几行，调用了一个autoCommitTransaction方法：
protected void autoCommitTransaction(SessionScope sessionScope, boolean autoStart) throws SQLException
{
    if (autoStart)
      sessionScope.getSqlMapTxMgr().commitTransaction();
}
也就说如果是自动事务管理，那么在这个方法中就会调用sessionScope中的事务事务管理器(SqlMapClientImpl本身，他实现了SqlMapTransactionManager接口)的commitTransaction方法：
public void commitTransaction() throws SQLException {
    getLocalSqlMapSession().commitTransaction();
}

 

他又调用了SqlMapSessionImpl的commitTransaction方法，最中调用到了SqlMapExecutorDelegate的commitTransaction方法：
public void commitTransaction(SessionScope sessionScope) throws SQLException{
    try
    {
      if (sessionScope.isInBatch()) {
        executeBatch(sessionScope);
      }
      this.sqlExecutor.cleanup(sessionScope);
      this.txManager.commit(sessionScope);
    } catch (TransactionException e) {
      throw new NestedSQLException("Could not commit transaction.  Cause: " + e, e);
    }
}
这里就可以看到，如果发现实在批处理中，那么就直接把批处理执行了再提交。这也就是我们不加事务的情况下，使用ibatis的startBatch和executeBatch方法无效的原因：因为没一次操作(insert等)添加到批量操作中之后，又在自动事务提交的时候把批量操作给执行了，因此实质上还是单条单条的操作在执行。
 
使用Spring提供的SqlMapClientTempalte的execute方法执行时，传入了一个SqlMapClientCallback类型的对象，我们将执行批处理的代码写在了该对象的doInSqlMapClient方法中，这样就不需要我们在去写事务处理代码了，这时候会发现批处理就生效了。Spring也没做什么神奇的事情，就是给SesccionScope设置了一个Transaction对象，导致我们在执行操作(insert)时，自动事务就不会启用了。
public Object execute(SqlMapClientCallback action) throws DataAccessException {
   
    SqlMapSession session = this.sqlMapClient.openSession(); //这里打开了session,后续操作使用这个SqlMapSession对象完成
   
    Connection ibatisCon = null;

    try {
        Connection springCon = null;
        DataSource dataSource = getDataSource();
        boolean transactionAware = (dataSource instanceof TransactionAwareDataSourceProxy);

        // Obtain JDBC Connection to operate on...
        try {
            ibatisCon = session.getCurrentConnection();
            if (ibatisCon == null) {
                springCon = (transactionAware ?
                        dataSource.getConnection() : DataSourceUtils.doGetConnection(dataSource));
                session.setUserConnection(springCon);//这一步操作很关键
            }
        }
        catch (SQLException ex) {
            throw new CannotGetJdbcConnectionException("Could not get JDBC Connection", ex);
        }

        // Execute given callback...
        try {
            return action.doInSqlMapClient(session);//传入了打开的SqlSession对象
        }
}
这里省略部分其他代码，仅看看我们关心的部分。首先创建了一个SqlMapSession，并且获取了一个Spring可管理的Connection,设置到了SqlMapSession中，正是这一操作使得ibatis的自动事务关闭了。我们看SqlMapSession的setUserConnection方法，调用了delegate的setUserProviededTransaction方法：
public void setUserProvidedTransaction(SessionScope sessionScope, Connection userConnection)
{
    if (sessionScope.getTransactionState() == TransactionState.STATE_USER_PROVIDED) {
      sessionScope.recallTransactionState();
    }
    if (userConnection != null) {
      Connection conn = userConnection;
      sessionScope.saveTransactionState();
      sessionScope.setTransaction(new UserProvidedTransaction(conn));
      sessionScope.setTransactionState(TransactionState.STATE_USER_PROVIDED);
    } else {
      sessionScope.setTransaction(null);
      sessionScope.closePreparedStatements();
      sessionScope.cleanup();
    }
}
这里就给SessionScope设置了一个UserProvidedTransactiond对象。ibatis就不会再去关注事务了，由用户自己去管理事务了，这里相当于就是把事务交给了spring来管理了。如果我们没用通过spring给执行批量操作的方法加事务操作，那么实际上就相当于这段代码没有使用事务。
 

ibatis批量执行分析


联系
我们