Mybatis源码阅读 之 玩转Executor

承接上篇博客, 本文探究MyBatis中的Executor, 如下图: 是Executor体系图

本片博客的目的就是探究如上图中从顶级接口Executor中拓展出来的各个子执行器的功能,以及进一步了解Mybatis的一级缓存和二级缓存

预览:

• BaseExecutor :实现了Executor的全部方法,包括对缓存,事务,连接提供了一系列的模板方法, 这写模板方法中留出来了四个抽象的方法等待子类去实现如下

protected abstract int doUpdate(MappedStatement ms, Object parameter)
throws SQLException;

protected abstract List<BatchResult> doFlushStatements(boolean isRollback)
throws SQLException;

protected abstract <E> List<E> doQuery(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, BoundSql boundSql)
throws SQLException;

protected abstract <E> Cursor<E> doQueryCursor(MappedStatement ms, Object parameter, RowBounds rowBounds, BoundSql boundSql)
throws SQLException;

• SimpleExecutor: 特点是每次执行完毕后都会将创建出来的statement关闭掉,他也是默认的执行器类型
• ReuseExecutor: 在它在本地维护了一个容器,用来存放针对每条sql创建出来的statement,下次执行相同的sql时,会先检查容器中是否存在相同的sql,如果存在就使用现成的,不再重复获取
• BatchExecutor: 特点是进行批量修改,她会将修改操作记录在本地,等待程序触发提交事务,或者是触发下一次查询时,批量执行修改

创建执行器

当我们通过

SqlSessionFactory

openSessionFromDataBase()

newExecutor()

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;
}

通过这个函数,可以找到上面列举出来的所有的 执行器, MyBatis默认创建的执行器的类型的是SimpleExecutor,而且MyBatis默认开启着对mapper的缓存(这其实就是Mybatis的二级缓存,但是,不论是注解版,还是xml版,都需要添加额外的配置才能使添加这个额外配置的mapper享受二级缓存,二级缓存被这个CachingExecutor维护着)

BaseExecutor 的模板方法

在BaseExecutor的模本方法之前,其实省略了很多步骤,我们上一篇博文中有详细的叙述,感兴趣可以去看看,下面我就简述一下: 程序员使用获取到了mapper的代理对象,调用对象的

findAll()

DefaultSqlSession

CachingExecutor

query()

CachingExecutor

如下是

BaseExecutor

query()

@Override
public <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey key, BoundSql boundSql) throws SQLException {
ErrorContext.instance().resource(ms.getResource()).activity("executing a query").object(ms.getId());
if (closed) {
throw new ExecutorException("Executor was closed.");
}
if (queryStack == 0 && ms.isFlushCacheRequired()) {
clearLocalCache();
}
List<E> list;
try {
queryStack++;
list = resultHandler == null ? (List<E>) localCache.getObject(key) : null;
if (list != null) {
handleLocallyCachedOutputParameters(ms, key, parameter, boundSql);
} else {
list = queryFromDatabase(ms, parameter, rowBounds, resultHandler, key, boundSql);
}
} finally {
queryStack--;
}
if (queryStack == 0) {
for (DeferredLoad deferredLoad : deferredLoads) {
deferredLoad.load();
}
// issue #601
deferredLoads.clear();
if (configuration.getLocalCacheScope() == LocalCacheScope.STATEMENT) {
// issue #482
clearLocalCache();
}
}
return list;
}

BaseExecutor维护的一级缓存

从上面的代码中,其实我们就跟传说中的Mybatis的一级缓存无限接近了,上面代码中的逻辑很清楚,就是先检查是否存在一级缓存,如果存在的话,就不再去创建statement查询数据库了

那问题来了,什么是这个一级缓存呢? 一级缓存就是上面代码中的

localCache

再详细一点就看下面这张图:

嗯! 原来传说中的一级缓存叫localCache,它的封装类叫

PerpetualCache

一级缓存什么时候被填充的值呢?填充值的操作在一个叫做

queryFromDataBase()

其中的key=

1814536652:3224182340:com.changwu.dao.IUserDao.findAll:0:2147483647:select * from user:mysql

其实看到这里,平时听到的为什么大家会说一级缓存是属于SqlSession的啊,诸如此类的话就是从这个看源码的过程中的出来的结果,如果你觉的印象不深刻,我就接着补刀,每次和数据库打交道都的先创建sqlSession,创建sqlSession的方法会在创建出DefaultSqlSession之前,先为它创建一个Executor,而我们说的一级缓存就是这个Executor的属性

何时清空一级缓存

清空一级缓存的方法就是

BaseExecutor

update()

@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);
}

SimpleExecutor

SimpleExecutor

XXXExecutor

SimpleExecutor

BaseExecutor

doQuery()

详细的过程在这篇博文中我就不往外贴代码了,因为我在上一篇博文中有这块源码的详细追踪

@Override
public <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);
}
}

创建StatementHandler

public StatementHandler newStatementHandler(Executor executor, MappedStatement mappedStatement, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler, BoundSql boundSql) {
StatementHandler statementHandler = new RoutingStatementHandler(executor, mappedStatement, parameterObject, rowBounds, resultHandler, boundSql);
statementHandler = (StatementHandler) interceptorChain.pluginAll(statementHandler);
return statementHandler;
}

虽然表面上看上面的代码,感觉它只会创建一个叫

RoutingStatementHandler

public RoutingStatementHandler(Executor executor, MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, BoundSql boundSql) {

switch (ms.getStatementType()) {
case STATEMENT:
// 早期的普通查询,极其容易被sql注入,不安全
delegate = new SimpleStatementHandler(executor, ms, parameter, rowBounds, resultHandler, boundSql);
break;
case PREPARED:
//  处理预编译类型的sql语句
delegate = new PreparedStatementHandler(executor, ms, parameter, rowBounds, resultHandler, boundSql);
break;
case CALLABLE:
// 处理存储过程语句
delegate = new CallableStatementHandler(executor, ms, parameter, rowBounds, resultHandler, boundSql);
break;
default:
throw new ExecutorException("Unknown statement type: " + ms.getStatementType());
}

创建PreParedStatement

点击进入上篇博文,查看如何创建PreparedStatement

执行查询

点击进入上篇博文,里面有记录如何执行查询

关闭连接

关于

SimpleExecutor

SimpleExecutor

closeStatement()

protected void closeStatement(Statement statement) {
if (statement != null) {
try {
statement.close();
} catch (SQLException e) {
// ignore
}
}
}

ReuseExecutor

这个ReuseExecutor相对于SimpleExecutor来说,不同点就是它先来的对Statement的复用,换句话说,某条Sql对应的Statement创建出来后被放在容器中保存起来,再有使用这个statement的地方就是容器中拿就行了

他是怎么实现的呢? 看看下面的代码就知道了

public class ReuseExecutor extends BaseExecutor {
private final Map<String, Statement> statementMap = new HashMap();

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

嗯! 所谓的容器,不过是一个叫statementMap的HashMap而已

下一个问题: 这个容器什么时候派上用场呢? 看看下面的代码也就知道了--

this.hasStatementFor(sql)

private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {
BoundSql boundSql = handler.getBoundSql();
String sql = boundSql.getSql();
Statement stmt;
if (this.hasStatementFor(sql)) {
stmt = this.getStatement(sql);
this.applyTransactionTimeout(stmt);
} else {
Connection connection = this.getConnection(statementLog);
stmt = handler.prepare(connection, this.transaction.getTimeout());
this.putStatement(sql, stmt);
}

handler.parameterize(stmt);
return stmt;
}

最后一点: 当MyBatis知道发生了事务的提交,回滚等操作时,

ReuseExecutor

BatchExecutor

这个执行器相对于SimpleExecutor的特点是,它的

update()

执行器提交或回滚事务时会调用 doFlushStatements，从而批量执行提交的 sql 语句并最终批量关闭 statement 对象。

CachingExecutor与二级缓存

首先来说,这个

CachingExecutor

public class CachingExecutor implements Executor {
private final Executor delegate;
private final TransactionalCacheManager tcm = new TransactionalCacheManager();

让我们回想一下他的创建时机,没错就是在每次创建一个新的SqlSession时创建出来的,源码如下,这就出现了一个惊天的大问号!!!,一级缓存和二级缓存为啥就一个属于SqlSession级别,另一个却被所有的SqlSession共享了? 这不是开玩笑呢? 我当时确实也是真的蒙,为啥他俩都是随时用随时new,包括上面代码中的

TransactionalCacheManager

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;
}

先说一下,我是看到哪行代码后意识到二级缓存是这么特殊的,如下:大家也看到了,下面代码中的

tcm.getObject(cache, key);

TransactionalCacheManager

getObject()

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会创建一个

XMLConfigBuilder

我们关注什么呢? 我们关注的是这个builder解析

<mapper>

private void parseConfiguration(XNode root) {
try {
//issue #117 read properties first
propertiesElement(root.evalNode("properties"));
...
databaseIdProviderElement(root.evalNode("databaseIdProvider"));
typeHandlerElement(root.evalNode("typeHandlers"));
mapperElement(root.evalNode("mappers"));

关注这个方法中的

configuration.addMapper(mapperInterface);

MapperAnnotationBuilder

public <T> void addMapper(Class<T> type) {
mapperRegistry.addMapper(type);
}

所以说我们就得去看看这个解析注解版本mapper的builder,到底是如何解析我提供的mapper的,源码如下:

public void parse() {
String resource = type.toString();
if (!configuration.isResourceLoaded(resource)) {
loadXmlResource();
configuration.addLoadedResource(resource);
assistant.setCurrentNamespace(type.getName());
parseCache();
parseCacheRef();
Method[] methods = type.getMethods();
for (Method method : methods) {
try {
// issue #237
if (!method.isBridge()) {
parseStatement(method);
}
} catch (IncompleteElementException e) {
configuration.addIncompleteMethod(new MethodResolver(this, method));
}
}
}

方法千千万,但是我关注的是它的

parseCache();

接下来就进入了一个高潮,相信你看到下面的代码也会激动, 为什么激动呢? 因为我们发现了Mybatis处理

@CacheNamespace

private void parseCache() {
CacheNamespace cacheDomain = type.getAnnotation(CacheNamespace.class);
if (cacheDomain != null) {
Integer size = cacheDomain.size() == 0 ? null : cacheDomain.size();
Long flushInterval = cacheDomain.flushInterval() == 0 ? null : cacheDomain.flushInterval();
Properties props = convertToProperties(cacheDomain.properties());
assistant.useNewCache(cacheDomain.implementation(), cacheDomain.eviction(), flushInterval, size, cacheDomain.readWrite(), cacheDomain.blocking(), props);
}
}

再往下跟进这个

assistant.useNewCache()

BlockingCache

创建出来的对象给谁了?

• Configuration对象自己留了一份 (放在了 caches = new StrictMap<>("Caches collection");中)
• 当前类

  MapperBuilderAssistant

  也保留一了一份
• 最主要的是

  MappedStatement

  对象中也保留了一份

  mappedStatement.cache

说了这么多了,附上一张图,用来纪念创建这个Cache的成员

小结

其实上面创建这个Cache对象才是二级缓存者, 前面说的那个

CachingExecutor

TransactionalCacheManager

我有调试他是如何从Cache中获取出缓存,事实证明,二级缓存中存放的不是对象,而是被序列化后存储的数据,需要反序列化出来

下图是Mybatis反序列化数据到新创建的对象中的截图

下图是

TransactionalCacheManager

二级缓存与一级缓存的互斥性

第一点: 通过以上代码的调用顺序也能看出,二级缓存在一级缓存之前优先被执行, 也就是说二级缓存不存在,则查询一级缓存,一级缓存再不存在,就查询DB

第二点: 就是说,对于二级缓存来说,无论我们有没有开启事务的自动提交功能,都必须手动

commit()

第三点:

CachingExecutor

@Override
public void commit(boolean required) throws SQLException {
// 代理执行器提交
delegate.commit(required);
// 事务缓存管理器提交
tcm.commit();
}

这就意味着,TransactionalCacheManager和BaseExecutor的实现类的事务都会被提交

为什么说二级缓存和以及缓存互斥呢?可以看看BaseExecutor的源码中

commit()

commit()

commit()

@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();
}
}