MyBatis源码阅读(八) --- Executor执行器

Mybatis源码阅读系列文章：

MyBatis源码阅读(一) --- 源码阅读环境搭建

MyBatis源码阅读(二) --- 执行流程分析

MyBatis源码阅读(三) --- 配置信息的解析以及SqlSessionFactory构建过程

MyBatis源码阅读(四) --- SqlSession的创建过程

MyBatis源码阅读(五) ---Mapper接口的获取过程

MyBatis源码阅读(六) ---mapper方法具体执行流程分析

MyBatis源码阅读(七) --- 查询结果集封装流程

MyBatis源码阅读(八) --- Executor执行器

MyBatis源码阅读(九) --- 插件原理

MyBatis源码阅读(十) --- 一级缓存、二级缓存工作原理

MyBatis源码阅读(十一) --- MyBatis事务管理机制

MyBatis源码阅读(十二) --- Spring加载MyBatis过程

一、概述

二、Executor执行器的创建过程

三、SimpleExecutor

四、ReuseExecutor

五、BatchExecutor

六、CachingExecutor

七、总结

一、概述

Executor 是一个接口，包含更新，查询，事务等一系列方法。在前面分析SqlSession创建过程的时候，我们知道每个SqlSession对象都会有一个Executor对象，SqlSession的操作都会交由Executor执行器执行。

我们先看看Executor类的继承图：

Executor接口有两个实现，

一个是BaseExecutor抽象类，BaseExecutor又有四个子类：

SimpleExecutor：简单类型的执行器，也是默认的执行器，每次执行update或者select操作，都会创建一个Statement对象，执行结束后关闭Statement对象；
ReuseExecutor：可重用的执行器，重用的是Statement对象，第一次执行一条sql，会将这条sql的Statement对象缓存在key-value结构的map缓存中。下一次执行，就可以从缓存中取出Statement对象，减少了重复编译的次数，从而提高了性能。每个SqlSession对象都有一个Executor对象，因此这个缓存是SqlSession级别的，所以当SqlSession销毁时，缓存也会销毁；
BatchExecutor：批量执行器，默认情况是每次执行一条sql，MyBatis都会发送一条sql。而批量执行器的操作是，每次执行一条sql，不会立马发送到数据库，而是批量一次性发送多条sql；
ClosedExecutor: ResultLoaderMap的内部类，用来进行处理懒加载相关功能；

另外一个是CachingExecutor实现类，在缓存的时候用到，使用到了装饰者模式对executor进行二次包装，动态增强了executor的功能；

Executor 接口采用了模版方法的设计模式，定义了一些模版方法，交给子类去实现。定义的方法如下：

public interface Executor {ResultHandler NO_RESULT_HANDLER = null;// 更新int update(MappedStatement ms, Object parameter) throws SQLException;// 先查询缓存，在查询数据库<E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey cacheKey, BoundSql boundSql) throws SQLException;// 查询<E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException;// 返回游标对象<E> Cursor<E> queryCursor(MappedStatement ms, Object parameter, RowBounds rowBounds) throws SQLException;// 释放StatementList<BatchResult> flushStatements() throws SQLException;// 事务提交void commit(boolean required) throws SQLException;// 事务回滚void rollback(boolean required) throws SQLException;// 创建缓存的键值对CacheKey createCacheKey(MappedStatement ms, Object parameterObject, RowBounds rowBounds, BoundSql boundSql);// 缓存是否存在boolean isCached(MappedStatement ms, CacheKey key);// 清除一级缓存void clearLocalCache();void deferLoad(MappedStatement ms, MetaObject resultObject, String property, CacheKey key, Class<?> targetType);// 获取事务对象Transaction getTransaction();void close(boolean forceRollback);boolean isClosed();void setExecutorWrapper(Executor executor);}

二、Executor执行器的创建过程

在前面分析SqlSession的获取过程的时候，我们当时暂且先跳过了executor执行器的创建过程这一部分的分析。很显然，executor的创建就是在获取SqlSession的时候。

//org.apache.ibatis.session.defaults.DefaultSqlSessionFactory#openSessionFromDataSource
private SqlSession openSessionFromDataSource(ExecutorType execType, TransactionIsolationLevel level, boolean autoCommit) {//事务对象Transaction tx = null;try {//从configuration对象中获取到我们之前解析的environment环境信息final Environment environment = configuration.getEnvironment();//事务工厂，这里是JbdcTransactionFactory工厂类final TransactionFactory transactionFactory = getTransactionFactoryFromEnvironment(environment);//通过事务工厂创建JbdcTransaction事务，传入数据源等信息tx = transactionFactory.newTransaction(environment.getDataSource(), level, autoCommit);//创建Executor执行器final Executor executor = configuration.newExecutor(tx, execType);//创建DefaultSqlSession会话，传入Configuration、Executor对象return new DefaultSqlSession(configuration, executor, autoCommit);} catch (Exception e) {closeTransaction(tx); // may have fetched a connection so lets call close()throw ExceptionFactory.wrapException("Error opening session.  Cause: " + e, e);} finally {ErrorContext.instance().reset();}
}

final Executor executor = configuration.newExecutor(tx, execType);

通过configuration对象创建executor执行器，传入前面创建好的事务tx，以及指定的执行器类型。

继续追踪一下newExecutor()的源码：

public Executor newExecutor(Transaction transaction, ExecutorType executorType) {executorType = executorType == null ? defaultExecutorType : executorType;//没有配置的话，默认创建SimpleExecutorexecutorType = executorType == null ? ExecutorType.SIMPLE : executorType;Executor executor;//第一步：根据ExecutorType来创建不同类型的执行器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);}//第二步：如果开启了一级缓存，使用装饰者模式对executor二次包装成CachingExecutorif (cacheEnabled) {executor = new CachingExecutor(executor);}//这里使用了责任链设计模式，在插件篇幅里面会详细介绍//第三步：加载插件executor = (Executor) interceptorChain.pluginAll(executor);return executor;
}

我们可以看到，newExecutor()方法根据ExecutorType来创建不同类型的执行器，默认创建的是SimpleExecutor简单类型执行器。

接下来我们看一下各种执行器的构造方法：

public SimpleExecutor(Configuration configuration, Transaction transaction) {super(configuration, transaction);
}
public ReuseExecutor(Configuration configuration, Transaction transaction) {super(configuration, transaction);
}
public BatchExecutor(Configuration configuration, Transaction transaction) {super(configuration, transaction);
}

我们看到了super(xxx)方法，也就是说三种类型的执行器其实都是调用的父类BaseExecutor的构造方法来创建：

protected BaseExecutor(Configuration configuration, Transaction transaction) {//事务this.transaction = transaction;//延迟加载this.deferredLoads = new ConcurrentLinkedQueue<>();//一级缓存this.localCache = new PerpetualCache("LocalCache");this.localOutputParameterCache = new PerpetualCache("LocalOutputParameterCache");this.closed = false;//全局配置对象this.configuration = configuration;this.wrapper = this;
}

final Executor executor = configuration.newExecutor(tx, execType);
return new DefaultSqlSession(configuration, executor, autoCommit);

到这里，executor创建就完成了，同时每个SqlSession也就有了自己唯一的Executor对象，如上代码executor作为DefaultSqlSession的成员属性。至于executor是在哪里调用的，相信看过前几篇文章的小伙伴都应该很清楚了，就是在SqlSession执行query方法的时候，具体是交给executor去执行查询的，具体过程可以参照前面的文章，这里不过多赘述了。

三、SimpleExecutor

SimpleExecutor 是默认的执行器，也是最简单的执行器。我们来看看里面几个关键方法：

public class SimpleExecutor extends BaseExecutor {public SimpleExecutor(Configuration configuration, Transaction transaction) {//默认调用父类BaseExecutor的构造方法super(configuration, transaction);}@Overridepublic int doUpdate(MappedStatement ms, Object parameter) throws SQLException {Statement stmt = null;try {//第一步：从MappedStatement中获取到configuration全局配置对象Configuration configuration = ms.getConfiguration();//第二步:通过configuration创建StatementHandlerStatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null);//第三步：创建Statement对象stmt = prepareStatement(handler, ms.getStatementLog());//第四步：执行SQL查询return handler.update(stmt);} finally {//第五步：关闭statementcloseStatement(stmt);}}@Overridepublic <E> List<E> doQuery(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, BoundSql boundSql) throws SQLException {Statement stmt = null;try {Configuration configuration = ms.getConfiguration();StatementHandler handler = configuration.newStatementHandler(wrapper, ms, parameter, rowBounds, resultHandler, boundSql);stmt = prepareStatement(handler, ms.getStatementLog());return handler.query(stmt, resultHandler);} finally {closeStatement(stmt);}}private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {Statement stmt;//获取到connection连接Connection connection = getConnection(statementLog);//创建Statement stmt = handler.prepare(connection, transaction.getTimeout());//将sql语句中的占位符替换成最终查询参数//最底层实现：typeHandler.setParameter(ps, i + 1, value, jdbcType);handler.parameterize(stmt);return stmt;}}

总结一下SimpleExecutor的操作步骤：

第一步：从MappedStatement中获取到configuration全局配置对象；
第二步:通过configuration创建StatementHandler；
第三步：创建Statement对象；
第四步：执行SQL查询；

四、ReuseExecutor

ReuseExecutor，重用类型的执行器，它的作用是重复利用statement对象，避免频繁的创建。如果在一个SqlSession中多次执行一条sql，如果每次都去生成Statement对象，会造成资源浪费。因此ReuseExecutor在SimpleExecutor的基础上，对prepareStatement()方法进行了改进，将Statement对象缓存在内存中，并且免去了关闭Statement对象这一步。

public class ReuseExecutor extends BaseExecutor {//map结构，key是执行的SQL语句，value就是对应的statement对象//可以看到，在一个会话中，相同的sql语句对应的statement可以重复利用  private final Map<String, Statement> statementMap = new HashMap<>();public ReuseExecutor(Configuration configuration, Transaction transaction) {super(configuration, transaction);}@Overridepublic int doUpdate(MappedStatement ms, Object parameter) throws SQLException {Configuration configuration = ms.getConfiguration();StatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null);Statement stmt = prepareStatement(handler, ms.getStatementLog());return handler.update(stmt);}@Overridepublic <E> List<E> doQuery(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, BoundSql boundSql) throws SQLException {//第一步：获取configuration对象Configuration configuration = ms.getConfiguration();//第二步：创建StatementHandlerStatementHandler handler = configuration.newStatementHandler(wrapper, ms, parameter, rowBounds, resultHandler, boundSql);//第三步：创建StatementStatement stmt = prepareStatement(handler, ms.getStatementLog());//第四步：执行SQL查询return handler.query(stmt, resultHandler);}@Overrideprotected <E> Cursor<E> doQueryCursor(MappedStatement ms, Object parameter, RowBounds rowBounds, BoundSql boundSql) throws SQLException {Configuration configuration = ms.getConfiguration();StatementHandler handler = configuration.newStatementHandler(wrapper, ms, parameter, rowBounds, null, boundSql);Statement stmt = prepareStatement(handler, ms.getStatementLog());return handler.queryCursor(stmt);}private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {Statement stmt;BoundSql boundSql = handler.getBoundSql();//获取到执行的SQL语句String sql = boundSql.getSql();if (hasStatementFor(sql)) {//如果缓存中存在该条sql对应的statement，则直接从缓存中取，不重新创建statementstmt = getStatement(sql);applyTransactionTimeout(stmt);} else {//如果缓存中不存在该条sql对应的statementConnection connection = getConnection(statementLog);stmt = handler.prepare(connection, transaction.getTimeout());//将创建好的statement放入缓存中，方便下次重复利用putStatement(sql, stmt);}handler.parameterize(stmt);return stmt;}//判断缓存map中是否存在sql的Statement对象private boolean hasStatementFor(String sql) {try {return statementMap.keySet().contains(sql) && !statementMap.get(sql).getConnection().isClosed();} catch (SQLException e) {return false;}}private Statement getStatement(String s) {return statementMap.get(s);}//以sql为key, Statement为value放到缓存中   private void putStatement(String sql, Statement stmt) {statementMap.put(sql, stmt);}}

从上面可以看到，ReuseExecutor内存维护了一个map结构的缓存statementMap，以sql为key, Statement为value放到缓存中，保证在同一个会话中，如果重复执行两个相同的SQL，第一次创建完的statement，可以在第二次查询的时候重复利用，节省了一些资源。注意：因为每个SqlSession都有自己唯一的对应的Executor对象，因此这个statementMap缓存是SqlSession级别的，如果SqlSession销毁了，statementMap缓存也会将销毁。

五、BatchExecutor

BatchExecutor是批处理的执行器，批量的发送sql到数据库，而不是一个个的发送。

public class BatchExecutor extends BaseExecutor {// 批量更新处理的固定返回值，不是返回受影响的行数public static final int BATCH_UPDATE_RETURN_VALUE = Integer.MIN_VALUE + 1002;// Statement集合 private final List<Statement> statementList = new ArrayList<>();//批量结果的集合private final List<BatchResult> batchResultList = new ArrayList<>();//当前Sql语句private String currentSql;//当前的MappedStatement对象private MappedStatement currentStatement;public BatchExecutor(Configuration configuration, Transaction transaction) {//默认调用父类BaseExecutor的构造方法super(configuration, transaction);}@Overridepublic int doUpdate(MappedStatement ms, Object parameterObject) throws SQLException {//第一步：获取configuration对象final Configuration configuration = ms.getConfiguration();//第二步：创建StatementHandlerfinal StatementHandler handler = configuration.newStatementHandler(this, ms, parameterObject, RowBounds.DEFAULT, null, null);//第三步：获取到BoundSql，再拿到具体的sqlfinal BoundSql boundSql = handler.getBoundSql();final String sql = boundSql.getSql();final Statement stmt;//如果当前执行的sql跟拿到的sql一致，并且MappedStatement也是同一个的话if (sql.equals(currentSql) && ms.equals(currentStatement)) {int last = statementList.size() - 1;stmt = statementList.get(last);applyTransactionTimeout(stmt);handler.parameterize(stmt);//fix Issues 322BatchResult batchResult = batchResultList.get(last);batchResult.addParameterObject(parameterObject);} else {Connection connection = getConnection(ms.getStatementLog());stmt = handler.prepare(connection, transaction.getTimeout());handler.parameterize(stmt);    //fix Issues 322//设置currentSql和currentStatement为当前sql、当前MappedStatementcurrentSql = sql;currentStatement = ms;//将statement添加到集合中statementList.add(stmt);batchResultList.add(new BatchResult(ms, sql, parameterObject));}//调用JDBC的addBatch()方法，添加到批处理中handler.batch(stmt);return BATCH_UPDATE_RETURN_VALUE;}@Overridepublic List<BatchResult> doFlushStatements(boolean isRollback) throws SQLException {try {List<BatchResult> results = new ArrayList<>();// 回滚则直接返回空集合if (isRollback) {return Collections.emptyList();}//遍历statementList集合中的Statement，一条条执行，并将结果加入到结果集合中for (int i = 0, n = statementList.size(); i < n; i++) {Statement stmt = statementList.get(i);applyTransactionTimeout(stmt);BatchResult batchResult = batchResultList.get(i);try {//执行sqlbatchResult.setUpdateCounts(stmt.executeBatch());MappedStatement ms = batchResult.getMappedStatement();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 #141for (Object parameter : parameterObjects) {keyGenerator.processAfter(this, ms, stmt, parameter);}}// 关闭statementcloseStatement(stmt);} 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.add(batchResult);}return results;} finally {for (Statement stmt : statementList) {closeStatement(stmt);}currentSql = null;statementList.clear();batchResultList.clear();}}}

BatchExecutor总结：

doUpdate()返回的值是固定的【Integer.MIN_VALUE + 1002】，不是影响的行数；
如果连续提交相同的sql，则只会执行一次；
提交sql不会立马执行，而是等到commit时候才统一执行；
底层使用的是JDBC的批处理操作，addBatch()和executeBatch()操作；

六、CachingExecutor

在前面总结executor创建的时候，我们看见过CachingExecutor，它的作用其实就是，如果项目中开启了一级缓存的话，它会将executor使用装饰者模式包装成CachingExecutor，来增强executor的功能。

if (cacheEnabled) {executor = new CachingExecutor(executor);
}

下面来看看CachingExecutor在Mybatis中是如何实现的。

public <E> List<E> query(MappedStatement ms, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException {BoundSql boundSql = ms.getBoundSql(parameterObject);//创建缓存keyCacheKey key = createCacheKey(ms, parameterObject, rowBounds, boundSql);return query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);
}//org.apache.ibatis.executor.CachingExecutor#query(org.apache.ibatis.mapping.MappedStatement, java.lang.Object, org.apache.ibatis.session.RowBounds, org.apache.ibatis.session.ResultHandler, org.apache.ibatis.cache.CacheKey, org.apache.ibatis.mapping.BoundSql)
public <E> List<E> query(MappedStatement ms, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler, CacheKey key, BoundSql boundSql)throws SQLException {//拿到二级缓存，判断二级缓存是否存在此数据Cache cache = ms.getCache();if (cache != null) {flushCacheIfRequired(ms);if (ms.isUseCache() && resultHandler == null) {ensureNoOutParams(ms, boundSql);@SuppressWarnings("unchecked")List<E> list = (List<E>) tcm.getObject(cache, key);if (list == null) {//如果二级缓存为空，再从一级缓存中取查找，还找不到，则查询数据库，然后再讲结果放到缓存中  list = delegate.query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);tcm.putObject(cache, key, list); // issue #578 and #116}return list;}}return delegate.query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);
}

七、总结

本篇文章主要总结了Mybatis里面其中一个重要的组件--Executor执行器，并介绍了它的继承体系以及分别介绍了它的常见的几个实现类，在执行查询或者缓存方面都是如何工作的，相信大家对Executor没有那么陌生了吧。

鉴于笔者水平有限，如果文章有什么错误或者需要补充的，希望小伙伴们指出来，希望这篇文章对大家有帮助。