ibatis批量执行分析
最近做pos数据文件解析及入库的开发，其中pos的流水文件一个文件中就包含8000多条数据，每次插入数据库执行的sql都是相同的。因此考虑到使用批量插入来提升效率。查看ibatis的文档，看到提供了startBatch和executBatch两个方法，看名字大概就知道这两个方法和批量执行有关。我立马在之前的for循环外面加上了这两个方法：
sqlMap.getSqlMapClient().startBatch();
for (Map<String, Object> map : list) {
sqlMap.insert(sqlId, map);
}
sqlMap.getSqlMapClient().executeBatch();
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;
}
});
@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;
}
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);
}
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);
}
{
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;
}
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;
}
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;
}
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);
}
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;
}
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();
}
{
if (autoStart)
sessionScope.getSqlMapTxMgr().commitTransaction();
}
也就说如果是自动事务管理，那么在这个方法中就会调用sessionScope中的事务事务管理器(SqlMapClientImpl本身，他实现了SqlMapTransactionManager接口)的commitTransaction方法：
public void commitTransaction() throws SQLException {
getLocalSqlMapSession().commitTransaction();
}
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);
}
}
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 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();
}
}
{
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给执行批量操作的方法加事务操作，那么实际上就相当于这段代码没有使用事务。
文章来自：http://www.cnblogs.com/jdluojing/p/4201832.html