mybatis-executor解析

概述

在这里插入图片描述

执行器包主要包含了 Executor、ParameterHandler、ResultSetHandler、StatementHandler。

这些都是sql执行中非常重要的一环，本篇从Executor开始。

Executor：执行器，主要职责是在sql执行过程中添加缓存和事务的功能。与jdbc相关的操作会继续委托给StatementHandler。

Executor

java">public interface Executor {

  ResultHandler NO_RESULT_HANDLER = null;

  // 更新
  int update(MappedStatement ms, Object parameter) throws SQLException;

  // 查询 resultHandler + cacheKey + boundSql
  <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey cacheKey, BoundSql boundSql) throws SQLException;

  // 查询 resultHandler
  <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;

  // 刷入批处理
  List<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提供的方法分类

sql操作
缓存
事务

可得出Executor的基本职责。

Executor的实现类：

在这里插入图片描述

BaseExecutor：基础执行器，实现Executor方法，添加通用缓存、事务处理，模板方法模式。
CachingExecutor：缓存执行器，BaseExecutor内部包含的是一级缓存，CachingExecutor是二级缓存。
ReuseExecutor：可重用执行器，会将Statement缓存。
SimpleExecutor：基础执行器，值也包含默认数据库执行。
BatchExecutor：批处理执行器，可进行批量执行sql（仅限Update）。

BaseExecutor

java">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;
  /**
   * 配置
   */
  protected Configuration configuration;

  /**
   * 查询的深度
   */
  protected int queryStack;
  /**
   * 是否关闭
   */
  private boolean closed;
}

update

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

query

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

java">@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) {
      // 存储过程时，缓存outPut类型参数
      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();
    // 缓存是Statement级别，则清楚
    if (configuration.getLocalCacheScope() == LocalCacheScope.STATEMENT) {
      // issue #482
      clearLocalCache();
    }
  }
  // 返回实体
  return list;
}

commit

java">@Override
public void commit(boolean required) throws SQLException {
  if (closed) {
    throw new ExecutorException("Cannot commit, transaction is already closed");
  }
  // 清除缓存
  clearLocalCache();
  // 刷新statement
  flushStatements();
  if (required) {
    transaction.commit();
  }
}

rollback

java">@Override
public void rollback(boolean required) throws SQLException {
  if (!closed) {
    try {
      clearLocalCache();
      flushStatements(true);
    } finally {
      if (required) {
        transaction.rollback();
      }
    }
  }
}

SimpleExecutor

简单执行器，最常用的执行器。

java">@Override
public int doUpdate(MappedStatement ms, Object parameter) throws SQLException {
  Statement stmt = null;
  try {
    Configuration configuration = ms.getConfiguration();
    // 创建 statementHandler
    StatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null);
    // 创建statement实例
    stmt = prepareStatement(handler, ms.getStatementLog());
    // 执行
    return handler.update(stmt);
  } finally {
    closeStatement(stmt);
  }
}

这里步骤很简单，创建StatementHandler和Statement实例，执行sql交给StatementHandler。

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

    /*
     * 创建一个statement处理器 用于操作jdbc statement
     */
    StatementHandler handler = configuration.newStatementHandler(wrapper, ms, parameter, rowBounds, resultHandler, boundSql);

    /*
     * 根据statementType创建对应的statement实例
     * simple
     * prepare
     * callable
     */
    stmt = prepareStatement(handler, ms.getStatementLog());

    /*
     * 执行sql并转换返回值
     * 这里handler类型是RoutingStatementHandler
     * 根据org.apache.ibatis.mapping.MappedStatement.statementType
     * 路由到真正的StatementHandler
     */
    return handler.query(stmt, resultHandler);
  } finally {
    closeStatement(stmt);
  }
}

query和update逻辑相同。

创建Statement实例

java">private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {
  Statement stmt;
  // 获取数据源连接，这里会对Connection创建代理类，添加了日志打印的功能
  Connection connection = getConnection(statementLog);
  //获取一个Statement实例
  stmt = handler.prepare(connection, transaction.getTimeout());

  //往statement里设置参数
  handler.parameterize(stmt);
  return stmt;
}

ReuseExecutor

可复用Statement实例执行器。SimpleExecutor中Statement是一次性的，用完马上关闭。

java">private final Map<String, Statement> statementMap = new HashMap<>();

存储Statement缓存，key是动态sql。其他与SimpleExecutor完全相同。

BatchExecutor

批量执行器，可批量执行sql。仅限update（jdbc仅支持批量update）

实在用的少，略过。

CachingExecutor

缓存执行器，是mybatis二级缓存机制的实现类。

java">public class CachingExecutor implements Executor {

  private final Executor delegate;
  /**
   * 缓存跨SqlSession，所以需要事务控制
   */
  private final TransactionalCacheManager tcm = new TransactionalCacheManager();
}

这里也是典型的装饰模式，执行器的具体实现都委托给Delegate执行。自己在之上添加缓存功能。

这里主要看下query方法，看下缓存的处理

java">@Override
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) {
        // 二级缓存为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);
}