今天我们一起来分析下mybatis中的Executor,我们暂且翻译为执行器吧,在mybatis中有以下执行器
1:BaseExecutor;
2:CachingExecutor;
3:SimpleExecutor;
4:ReuseExecutor;
5:BatchExecutor;
BasExecutor就是个基类,是一个抽象类,提供了执行的增删改查,关闭资源等等一列类方法,真正做事情的都是各个实现类
CachingExecutor:就是增加了缓存的执行器,
SimpleExecutor:mybatis的默认执行器,就是相当于一套sql语句执行一次数据库交互操作
ReuseExecutor:重用执行器:在同一个sqlSession反的范围下,同一个sql语句会缓存起来,重复利用,
BatchExecutor:顾名思义,批量执行器,就是我们执行批量更新或者批量插入的时候可以使用这个执行器,单挑语句执行,最后批量提交
鉴于此,这本篇文章,我就只对批量的执行器执行下分析,其他就不做具体的分析了,大家伙有兴趣的就自行看看源码
因为可能我们在项目中,会经常遇到的场景就是会有批量新增,或者批量删除,批量更新,的场景,一般我们会怎嘛处理这些那,大家肯定会想到一种处理方式,那就是sql语句中利用<foreach>标签,这个我就不做案例了,网上案例太多,可能在数据量小的时候,我们这样处理,没有什么问题,但是当数据量大的时候,比如一次性插入上千条,上万条,等等,会有可能让数据库宕机,所以,鉴于此,希望大家尽量不要再项目中使用循环的批量插入,这次,我们使用一种新的模式,来处理批量的数据处理,这里我们就要借用到mybatis的Batch执行器了,
我会从两个方面来分析,第一种方案,使用原生的DefaultSqlSession来操作,第二种方案,使用spring提供的SqlsessionTemplete来操作
话不多说,我们通过案例,然后窥探下myabtis的源码是怎样实现批量的哈
我们先看一下SqlSessionFactory这个接口申明的方法
public interface SqlSessionFactory {
SqlSession openSession();
SqlSession openSession(boolean autoCommit);
SqlSession openSession(Connection connection);
SqlSession openSession(TransactionIsolationLevel level);
SqlSession openSession(ExecutorType execType);
SqlSession openSession(ExecutorType execType, boolean autoCommit);
SqlSession openSession(ExecutorType execType, TransactionIsolationLevel level);
SqlSession openSession(ExecutorType execType, Connection connection);
Configuration getConfiguration();
可以看到定义很多重载方法,一看明了,不用在多做介绍了,有获取连接的,定义事物的,定义执行器类型的等等
接下来,进入主题,我这边写了个demo,我们通过demo来分析
@RunWith(SpringRunner.class)
@SpringBootTest
public class SqlBatch {
@Autowired
private SqlSessionFactory sqlSessionFactory;
@Test
public void testBatch(){
//通过sqlsessionFactory指定执行器的类型为BATCH,自动提交设置为false
SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
//获取mapper接口类
RyxAccountMapper mapper = sqlSession.getMapper(RyxAccountMapper.class);
//循环插入10条记录
for (int i =0;i<10;i++){
RyxAccount ryxAccount = new RyxAccount();
ryxAccount.setName(i+"小五");
ryxAccount.setMoney(new BigDecimal(3));
mapper.insert(ryxAccount);
}
//执行提交
sqlSession.commit();
//关闭资源
sqlSession.close();
}
}
案例很简单,就个简单的向数据库插入10条记录,我们运行,然后跟着源码一起来看,mybatis到底是怎样执行批量插入的
当我们运行的时候,会先进入DefaultSqlSession类,这个是myabtis的默认处理sqlSession的类,所有的主要逻辑都在这个类中开始实现的,终于怎样通过mapper接口找到这个的,就是通过动态代理生成代理类,这个我已经分析过了,就不在分析了
我们运行下代码
获取代理类后,执行mapperProxy,将Mapper方法放入缓存,执行mapperMethod.execute,确定执行语句的种类
接下来,进入我们的DefaultSqlSession类,的update方法(insert也是定义为update执行方法)
@Override
public int update(String statement, Object parameter) {
try {
dirty = true;
MappedStatement ms = configuration.getMappedStatement(statement);
//调用基类执行器
return executor.update(ms, wrapCollection(parameter));
} catch (Exception e) {
throw ExceptionFactory.wrapException("Error updating database. Cause: " + e, e);
} finally {
ErrorContext.instance().reset();
}
}
重点关注executor.update防方法,这里由于我们定义了执行器的类型,就不会再走默认的Simple了,而是会走Batch方法
@Override
public int update(MappedStatement ms, Object parameter) throws SQLException {
ErrorContext.instance().resource(ms.getResource()).activity("executing an update").object(ms.getId());
if (closed) {
throw new ExecutorException("Executor was closed.");
}
clearLocalCache();
return doUpdate(ms, parameter);
}
#BatchExecutor类
@Override
public int doUpdate(MappedStatement ms, Object parameterObject) throws SQLException {
final Configuration configuration = ms.getConfiguration();
final StatementHandler handler = configuration.newStatementHandler(this, ms, parameterObject, RowBounds.DEFAULT, null, null);
final BoundSql boundSql = handler.getBoundSql();
final String sql = boundSql.getSql();
final Statement stmt;
//当第一次执行的时候,currentSql中还没有值,会进入if分支,当执行到第二次的时候,就会进入else分支
if (sql.equals(currentSql) && ms.equals(currentStatement)) {
int last = statementList.size() - 1;
stmt = statementList.get(last);
applyTransactionTimeout(stmt);
handler.parameterize(stmt);//fix Issues 322
//将当前的statement,sql,参数对象放入batchResultList集合中,如下图所示,这里有个问题,
BatchResult batchResult = batchResultList.get(last);
batchResult.addParameterObject(parameterObject);
} else {
//获取连接
Connection connection = getConnection(ms.getStatementLog());
//创建statement对象
stmt = handler.prepare(connection, transaction.getTimeout());
//设置映射参数
handler.parameterize(stmt); //fix Issues 32
currentSql = sql;
currentStatement = ms;
//将statement对象放入集合
statementList.add(stmt);
//将当前的statement,sql,参数对象放入batchResultList集合中,如下图所示
batchResultList.add(new BatchResult(ms, sql, parameterObject));
}
// handler.parameterize(stmt);
//执行批量方法
handler.batch(stmt);
return BATCH_UPDATE_RETURN_VALUE;
}
@Override
public void batch(Statement statement) throws SQLException {
delegate.batch(statement);
}
@Override
public void batch(Statement statement) throws SQLException {
PreparedStatement ps = (PreparedStatement) statement;
//这里会根据我们使用的不同的连接池调用不同的方法,我目前使用的是阿里的druid连接池,所以会去调用
//DruidPooledPreparedStatement(这块暂时没有看的特别明白,所以下面的分析可能就不是特别正确)
//大概可能就是通过myabtis的拦截器,自定义druid的处理逻辑
ps.addBatch();
}
@Override
public void addBatch() throws SQLException {
//检查statement对象
checkOpen();
try {
//执行com.alibaba.druid.proxy.jdbc.PreparedStatementProxyImpl方法
stmt.addBatch();
} catch (Throwable t) {
throw checkException(t);
}
}
@Override
public void addBatch() throws SQLException {
createChain().preparedStatement_addBatch(this);
}
@Override
public void preparedStatement_addBatch(PreparedStatementProxy statement) throws SQLException {
if (this.pos < filterSize) {
nextFilter().preparedStatement_addBatch(this, statement);
return;
}
//将参数添加到statement对象中
statement.getRawObject().addBatch();
}
当我们处理完数据的时候,会有如下结果,paremetorObjects会存储我们需要新增的数据列表,
现在咱们数据已经构造完毕,sql语句也已经准备好,接下来,就是要提交逻辑了,咱们会调用sqlSession.commit
@Override
public void commit(boolean force) {
try {
//最终调用BatchExecutor的commit
executor.commit(isCommitOrRollbackRequired(force));
dirty = false;
} catch (Exception e) {
throw ExceptionFactory.wrapException("Error committing transaction. Cause: " + e, e);
} finally {
ErrorContext.instance().reset();
}
}
@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();
}
}
最终的执行逻辑在doFlushStatements,继续往下看
public List<BatchResult> flushStatements(boolean isRollBack) throws SQLException {
if (closed) {
throw new ExecutorException("Executor was closed.");
}
return doFlushStatements(isRollBack);
}
@Override
public List<BatchResult> doFlushStatements(boolean isRollback) throws SQLException {
try {
List<BatchResult> results = new ArrayList<BatchResult>();
if (isRollback) {
return Collections.emptyList();
}
//简历statement集合
for (int i = 0, n = statementList.size(); i < n; i++) {
//获取statement对象
Statement stmt = statementList.get(i);
//设置事物超时时间
applyTransactionTimeout(stmt);
//获取batchResult对象
BatchResult batchResult = batchResultList.get(i);
try {
//执行完这句代码后,数据库的数据就已经插入完成了,主要逻辑在stmt.executeBatch()中
batchResult.setUpdateCounts(stmt.executeBatch());
MappedStatement ms = batchResult.getMappedStatement();
//获取参数对象集合,这个时候,还是没有主键的,以下代码主要是为参数对象填充主键
//也就是说,这里有10个参数对象,就会循环10次,为每一个对象生成同一个主键,但是一直没有想到这样的设计目的
//假如这里我们批量插入1000条数据的时候,这里就循环1000次,返回的对象不同,但是主键是一样的,
最后循环调用keyGenerator.processAfter方法,完成主键的查询和赋值,这段代码就不具体分析了,
List<Object> parameterObjects = batchResult.getParameterObjects();
KeyGenerator keyGenerator = ms.getKeyGenerator();
if (Jdbc3KeyGenerator.class.equals(keyGenerator.getClass())) {
Jdbc3KeyGenerator jdbc3KeyGenerator = (Jdbc3KeyGenerator) keyGenerator;
jdbc3KeyGenerator.processBatch(ms, stmt, parameterObjects);
} else if (!NoKeyGenerator.class.equals(keyGenerator.getClass())) { //issue #141
for (Object parameter : parameterObjects) {
keyGenerator.processAfter(this, ms, stmt, parameter);
}
}
} catch (BatchUpdateException e) {
StringBuilder message = new StringBuilder();
message.append(batchResult.getMappedStatement().getId())
.append(" (batch index #")
.append(i + 1)
.append(")")
.append(" failed.");
if (i > 0) {
message.append(" ")
.append(i)
.append(" prior sub executor(s) completed successfully, but will be rolled back.");
}
throw new BatchExecutorException(message.toString(), e, results, batchResult);
}
//将结果添加到results集合中,包含了,当前的sql,已经入库的有主键的参数对象,statement对象
results.add(batchResult);
}
return results;
} finally {
for (Statement stmt : statementList) {
closeStatement(stmt);
}
currentSql = null;
statementList.clear();
batchResultList.clear();
}
}
后面的逻辑那,提交完了,之后,就开始关闭连接,这里咱们用的是连接池,所以不是真的关闭连接,而是将连接的资源放回到连接池中,下次继续使用.
接下来,我们使用spring的方式看看myabtis的如何操作的批量sql插入,我也是写了个demo,咱们一起通过demo来分析源码,这里大家注意下,我这里加上了事物注解,如果没有事物注解,这里不是批量提交的,而是,插入一行,提交一行,千万注意,咱们在使用spring整合mybatis的时候,如果要使用mybatis的批量插入,在修改执行器的类型后,一定要加入事物注解否则,是没有意义的,达不到一行一行插入,批量提交的结果,而是会插入一行,提交一行,还有一点注意下,这里不需要我们手动提价,而是spring还自动帮助我们提交,如果我们手动调用提交方法,会直接报错的,然后其他就没有什么区别了,至于为什么会在事物的情况下,才有用,我前面在分析sqlSession线程安全的问题的时候,已经描述过了,如果没有事物注解,spring每次会重新new一个DefaultSqlSession,不管执行器是什么类型,再执行完相应的SQL语句后,spring会自动提交,源码里写的很清楚,我们先看看没有添加事物注解的情况下过程
@Test
public void testBatch(){
sqlSessionTemplate = new SqlSessionTemplate(sqlSessionFactory, ExecutorType.BATCH);
final RyxAccountMapper mapper = sqlSessionTemplate.getMapper(RyxAccountMapper.class);
for (int i =0;i<10;i++){
RyxAccount ryxAccount = new RyxAccount();
ryxAccount.setName(i+"小新");
ryxAccount.setMoney(new BigDecimal(3));
mapper.insert(ryxAccount);
}
}
demo很简单,就不做分析了,我们直接进入源码,重点地方我都会添加注释
private class SqlSessionInterceptor implements InvocationHandler {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//获取sqlSession
SqlSession sqlSession = getSqlSession(
SqlSessionTemplate.this.sqlSessionFactory,
SqlSessionTemplate.this.executorType,
SqlSessionTemplate.this.exceptionTranslator);
try {
//执行invoke方法,调用DefaultSqlSession的相关逻辑
Object result = method.invoke(sqlSession, args);
//判断当前的sqlSession是否是被事物管理,如果不是事物管理,直接提交,如果,是事物管理,则执行下一步,
//由于未被事物管理,就失去了批量提交的意义,这里会直接提交,
if (!isSqlSessionTransactional(sqlSession, SqlSessionTemplate.this.sqlSessionFactory)) {
// force commit even on non-dirty sessions because some databases require
// a commit/rollback before calling close()
sqlSession.commit(true);
}
return result;
} catch (Throwable t) {
Throwable unwrapped = unwrapThrowable(t);
if (SqlSessionTemplate.this.exceptionTranslator != null && unwrapped instanceof PersistenceException) {
// release the connection to avoid a deadlock if the translator is no loaded. See issue #22
closeSqlSession(sqlSession, SqlSessionTemplate.this.sqlSessionFactory);
sqlSession = null;
Throwable translated = SqlSessionTemplate.this.exceptionTranslator.translateExceptionIfPossible((PersistenceException) unwrapped);
if (translated != null) {
unwrapped = translated;
}
}
throw unwrapped;
} finally {
//这里关闭session也是一样的,如果存在事务的话,是将当前占用的资源的计数器减一,如果没有事物才是将
//sqlSession关闭
if (sqlSession != null) {
closeSqlSession(sqlSession, SqlSessionTemplate.this.sqlSessionFactory);
}
}
}
}
后续的代码已经分析过了我就不做具体的分析了,从以上我们可以总结下
当使用sqlsessionFactry获取的sqlSession是DefaultSqlsession的时候,我们执行批量提交,需要手动关闭sqlSession
当使用SqlsessionTemplete的时候,如果我们没有添加事物,那样的结果就是,和普通的执行没有区别的,会一条条的执行,提交,因为sprsqlSesisonTemplete会判断是不是使用了,事物,如果没有使用事物,会自动提交当前的sqlSession会话
当开启事物的时候,才会执行批量的提交,这里有一点需要注意下,在测试的时候,需要将事物管理的方法写到service上,我用的springboot,在测试类中,直接添加事物注解,不会执行提交的,而是会直接回滚,
