public interface Executor {
ResultHandler NO_RESULT_HANDLER = null;
/**
* 更新方法 update , insert , delete 都执行这个方法
* @param ms sql文
* @param parameter 查询参数
* @return
* @throws SQLException
*/
int update(MappedStatement ms, Object parameter) throws SQLException;
/**
* 查询方法
* @param ms sql 文
* @param parameter 查询参数
* @param rowBounds
* @param resultHandler 结果处理器
* @param cacheKey 缓存key
* @param boundSql
* @param <E>
* @return
* @throws SQLException
*/
<E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey cacheKey, BoundSql boundSql) throws SQLException;
/**
* 查询方法
* @param ms
* @param parameter
* @param rowBounds
* @param resultHandler
* @param <E>
* @return
* @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;
/**
* 批量sql
* @return
* @throws SQLException
*/
List<BatchResult> flushStatements() throws SQLException;
/**
* 提交
* @param required
* @throws SQLException
*/
void commit(boolean required) throws SQLException;
/**
* 回调, 事务回滚
* @param required
* @throws SQLException
*/
void rollback(boolean required) throws SQLException;
/**
* 缓存key的创建
* @param ms
* @param parameterObject
* @param rowBounds
* @param boundSql
* @return
*/
CacheKey createCacheKey(MappedStatement ms, Object parameterObject, RowBounds rowBounds, BoundSql boundSql);
boolean isCached(MappedStatement ms, CacheKey key);
/**
* 清空本地缓存(当前缓存)
*/
void clearLocalCache();
/**
* 延迟加载
* @param ms
* @param resultObject
* @param property
* @param key
* @param targetType
*/
void deferLoad(MappedStatement ms, MetaObject resultObject, String property, CacheKey key, Class<?> targetType);
/**
* 获取事物
* @return
*/
Transaction getTransaction();
/**
* 关闭
* @param forceRollback
*/
void close(boolean forceRollback);
/**
* 是否关闭
* @return
*/
boolean isClosed();
void setExecutorWrapper(Executor executor);
}public abstract class BaseExecutor implements Executor {
private static final Log log = LogFactory.getLog(BaseExecutor.class);
/**
* 事物
*/
protected Transaction transaction;
/**
* 执行器
*/
protected Executor wrapper;
/**
* 线程安全的队列
*/
protected ConcurrentLinkedQueue<DeferredLoad> deferredLoads;
/**
* 本地缓存
*/
protected PerpetualCache localCache;
/**
* 缓存
*/
protected PerpetualCache localOutputParameterCache;
/**
* mybatis-config 配置信息
*/
protected Configuration configuration;
protected int queryStack;
private boolean closed;
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;
}
@Override
public Transaction getTransaction() {
if (closed) {
throw new ExecutorException("Executor was closed.");
}
return transaction;
}
/**
* 关闭
*
* @param forceRollback 是否回调
*/
@Override
public void close(boolean forceRollback) {
try {
try {
rollback(forceRollback);
} finally {
if (transaction != null) {
transaction.close();
}
}
} catch (SQLException e) {
// Ignore. There's nothing that can be done at this point.
log.warn("Unexpected exception on closing transaction. Cause: " + e);
} finally {
transaction = null;
deferredLoads = null;
localCache = null;
localOutputParameterCache = null;
closed = true;
}
}
@Override
public boolean isClosed() {
return closed;
}
@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();
// 子类来实现 doUpdate 方法
return doUpdate(ms, parameter);
}
@Override
public List<BatchResult> flushStatements() throws SQLException {
return flushStatements(false);
}
public List<BatchResult> flushStatements(boolean isRollBack) throws SQLException {
if (closed) {
throw new ExecutorException("Executor was closed.");
}
return doFlushStatements(isRollBack);
}
@Override
public <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException {
// 获取绑定的sql
BoundSql boundSql = ms.getBoundSql(parameter);
// 获取缓存 key
CacheKey key = createCacheKey(ms, parameter, rowBounds, boundSql);
// 执行查询
return query(ms, parameter, rowBounds, resultHandler, key, boundSql);
}
@SuppressWarnings("unchecked")
@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()) {
// 查询堆栈==0 和 是否需要刷新缓存
// 清理本地缓存
clearLocalCache();
}
List<E> list;
try {
// 堆栈+1,防止重新清理缓存
queryStack++;
// 通过 缓存key 在本地缓存中获取
list = resultHandler == null ? (List<E>) localCache.getObject(key) : null;
if (list != null) {
// 通过缓存 key 查到后处理 localOutputParameterCache
handleLocallyCachedOutputParameters(ms, key, parameter, boundSql);
} else {
// 没有查询到从数据库查询
list = queryFromDatabase(ms, parameter, rowBounds, resultHandler, key, boundSql);
}
} finally {
// 堆栈-1
queryStack--;
}
if (queryStack == 0) {
for (DeferredLoad deferredLoad : deferredLoads) {
deferredLoad.load();
}
// 清空线程安全队列(延迟队列)
// issue #601
deferredLoads.clear();
if (configuration.getLocalCacheScope() == LocalCacheScope.STATEMENT) {
// STATEMENT 清空本地缓存
// issue #482
clearLocalCache();
}
}
return list;
}
@Override
public <E> Cursor<E> queryCursor(MappedStatement ms, Object parameter, RowBounds rowBounds) throws SQLException {
BoundSql boundSql = ms.getBoundSql(parameter);
return doQueryCursor(ms, parameter, rowBounds, boundSql);
}
@Override
public void deferLoad(MappedStatement ms, MetaObject resultObject, String property, CacheKey key, Class<?> targetType) {
if (closed) {
throw new ExecutorException("Executor was closed.");
}
DeferredLoad deferredLoad = new DeferredLoad(resultObject, property, key, localCache, configuration, targetType);
if (deferredLoad.canLoad()) {
// 判断是否可以立即加载
// 可以直接加载
deferredLoad.load();
} else {
// 不可以 new 一个新的放进去
deferredLoads.add(new DeferredLoad(resultObject, property, key, localCache, configuration, targetType));
}
}
/**
* 缓存key 创建,生成策略
* 标签id+offset+limit+sql+参数列表+Environment
*
* @param ms
* @param parameterObject
* @param rowBounds
* @param boundSql
* @return
*/
@Override
public CacheKey createCacheKey(MappedStatement ms, Object parameterObject, RowBounds rowBounds, BoundSql boundSql) {
if (closed) {
throw new ExecutorException("Executor was closed.");
}
CacheKey cacheKey = new CacheKey();
// select|insert|update|delete 标签的id
cacheKey.update(ms.getId());
// offset
cacheKey.update(rowBounds.getOffset());
// limit
cacheKey.update(rowBounds.getLimit());
// sql
cacheKey.update(boundSql.getSql());
List<ParameterMapping> parameterMappings = boundSql.getParameterMappings();
TypeHandlerRegistry typeHandlerRegistry = ms.getConfiguration().getTypeHandlerRegistry();
// mimic DefaultParameterHandler logic
for (ParameterMapping parameterMapping : parameterMappings) {
if (parameterMapping.getMode() != ParameterMode.OUT) {
Object value;
String propertyName = parameterMapping.getProperty();
if (boundSql.hasAdditionalParameter(propertyName)) {
value = boundSql.getAdditionalParameter(propertyName);
} else if (parameterObject == null) {
value = null;
} else if (typeHandlerRegistry.hasTypeHandler(parameterObject.getClass())) {
value = parameterObject;
} else {
MetaObject metaObject = configuration.newMetaObject(parameterObject);
value = metaObject.getValue(propertyName);
}
cacheKey.update(value);
}
}
if (configuration.getEnvironment() != null) {
// issue #176
cacheKey.update(configuration.getEnvironment().getId());
}
return cacheKey;
}
@Override
public boolean isCached(MappedStatement ms, CacheKey key) {
return localCache.getObject(key) != null;
}
@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();
}
}
/**
* 回掉方法
*
* @param required
* @throws SQLException
*/
@Override
public void rollback(boolean required) throws SQLException {
if (!closed) {
try {
clearLocalCache();
flushStatements(true);
} finally {
if (required) {
transaction.rollback();
}
}
}
}
@Override
public void clearLocalCache() {
if (!closed) {
// 清理缓存
localCache.clear();
localOutputParameterCache.clear();
}
}
/**
* 执行更新
*
* @param ms
* @param parameter
* @return
* @throws SQLException
*/
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;
protected void closeStatement(Statement statement) {
if (statement != null) {
try {
statement.close();
} catch (SQLException e) {
// ignore
}
}
}
/**
* Apply a transaction timeout.
*
* @param statement a current statement
* @throws SQLException if a database access error occurs, this method is called on a closed <code>Statement</code>
* @see StatementUtil#applyTransactionTimeout(Statement, Integer, Integer)
* @since 3.4.0
*/
protected void applyTransactionTimeout(Statement statement) throws SQLException {
StatementUtil.applyTransactionTimeout(statement, statement.getQueryTimeout(), transaction.getTimeout());
}
/**
* 处理 localOutputParameterCache
* 存储过程
*
* @param ms
* @param key
* @param parameter
* @param boundSql
*/
private void handleLocallyCachedOutputParameters(MappedStatement ms, CacheKey key, Object parameter, BoundSql boundSql) {
if (ms.getStatementType() == StatementType.CALLABLE) {
final Object cachedParameter = localOutputParameterCache.getObject(key);
if (cachedParameter != null && parameter != null) {
final MetaObject metaCachedParameter = configuration.newMetaObject(cachedParameter);
final MetaObject metaParameter = configuration.newMetaObject(parameter);
for (ParameterMapping parameterMapping : boundSql.getParameterMappings()) {
if (parameterMapping.getMode() != ParameterMode.IN) {
final String parameterName = parameterMapping.getProperty();
final Object cachedValue = metaCachedParameter.getValue(parameterName);
metaParameter.setValue(parameterName, cachedValue);
}
}
}
}
}
/**
* 在数据库中查询
*
* @param ms
* @param parameter
* @param rowBounds
* @param resultHandler
* @param key
* @param boundSql
* @param <E>
* @return
* @throws SQLException
*/
private <E> List<E> queryFromDatabase(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey key, BoundSql boundSql) throws SQLException {
List<E> list;
localCache.putObject(key, EXECUTION_PLACEHOLDER);
try {
list = doQuery(ms, parameter, rowBounds, resultHandler, boundSql);
} finally {
localCache.removeObject(key);
}
localCache.putObject(key, list);
if (ms.getStatementType() == StatementType.CALLABLE) {
localOutputParameterCache.putObject(key, parameter);
}
return list;
}
protected Connection getConnection(Log statementLog) throws SQLException {
Connection connection = transaction.getConnection();
if (statementLog.isDebugEnabled()) {
return ConnectionLogger.newInstance(connection, statementLog, queryStack);
} else {
return connection;
}
}
@Override
public void setExecutorWrapper(Executor wrapper) {
this.wrapper = wrapper;
}
/**
* 延迟队列中数据对象
*/
private static class DeferredLoad {
private final MetaObject resultObject;
private final String property;
private final Class<?> targetType;
private final CacheKey key;
private final PerpetualCache localCache;
private final ObjectFactory objectFactory;
private final ResultExtractor resultExtractor;
// issue #781
public DeferredLoad(MetaObject resultObject,
String property,
CacheKey key,
PerpetualCache localCache,
Configuration configuration,
Class<?> targetType) {
this.resultObject = resultObject;
this.property = property;
this.key = key;
this.localCache = localCache;
this.objectFactory = configuration.getObjectFactory();
this.resultExtractor = new ResultExtractor(configuration, objectFactory);
this.targetType = targetType;
}
public boolean canLoad() {
// 判断是否可以立即加载
// 从缓存中直接获取
return localCache.getObject(key) != null && localCache.getObject(key) != EXECUTION_PLACEHOLDER;
}
public void load() {
@SuppressWarnings("unchecked")
// we suppose we get back a List
List<Object> list = (List<Object>) localCache.getObject(key);
Object value = resultExtractor.extractObjectFromList(list, targetType);
resultObject.setValue(property, value);
}
}
}