mybatis源码阅读(五) ---执行器Executor

叁歲伎倆 2023-07-02 06:23 11阅读 0赞

转载自  [mybatis源码阅读(五) ---执行器Executor][mybatis_ ---_Executor]

## **1. Executor接口设计与类结构图** ##

public interface Executor {
    
      ResultHandler NO_RESULT_HANDLER = null;
    
      // 执行update，delete，insert三种类型的sql语句
      int update(MappedStatement ms, Object parameter) throws SQLException;
    
      // 执行select类型的SQL语句，返回值分为结果对象列表和游标对象
      <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;
    
      // 批量执行SQL语句
      List<BatchResult> flushStatements() throws SQLException;
    
      // 提交事务
      void commit(boolean required) throws SQLException;
    
      // 事务回滚
      void rollback(boolean required) throws SQLException;
    
      // 创建缓存中用到的CacheKey对象
      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();
    
      // 关闭Executor对象
      void close(boolean forceRollback);
    
      // 检测Executor对象是否关闭
      boolean isClosed();
    
      void setExecutorWrapper(Executor executor);
    
    }

![format_png][]

**简单执行器SimpleExecutor**：每执行一次update或select，就开启一个Statement对象，用完立刻关闭Statement对象。（可以是Statement或PrepareStatement对象）

**重用执行器ReuseExecutor**：执行update或select，以sql作为key查找Statement对象，存在就使用，不存在就创建，用完后，不关闭Statement对象，而是放置于Map<String, Statement>内，供下一次使用。（可以是Statement或PrepareStatement对象）

**批量执行器BatchExecutor**：执行update（没有select，JDBC批处理不支持select），将所有sql都添加到批处理中（addBatch()），等待统一执行（executeBatch()），它缓存了多个Statement对象，每个Statement对象都是addBatch()完毕后，等待逐一执行executeBatch()批处理的；BatchExecutor相当于维护了多个桶，每个桶里都装了很多属于自己的SQL，就像苹果蓝里装了很多苹果，番茄蓝里装了很多番茄，最后，再统一倒进仓库。（可以是Statement或PrepareStatement对象）

**缓存执行器CachingExecutor**：装饰设计模式典范，先从缓存中获取查询结果，存在就返回，不存在，再委托给Executor delegate去数据库取，delegate可以是上面任一的SimpleExecutor、ReuseExecutor、BatchExecutor。

**无用执行器ClosedExecutor**：毫无用处，读者可自行查看其源码，仅作为一种标识，和Serializable标记接口作用相当。

**作用范围：以上这五个执行器的作用范围，都严格限制在SqlSession生命周期范围内。**

## **2. 基类BaseExecutor源码解析** ##

它是一个实现了Executor接口的抽象类，实现了接口中的大部分方法，其中就是使用了模板模式，它主要提供了缓存和事物管理的基本功能，不同的实现类，只要实现4个基本方法来完成数据库的相关操作，这4个抽象方法：doUpdate()、doQuery()、doFlushStatement()、doQueryCursor。

源码片段

protected Transaction transaction;// 实现事务的回滚和提交，关闭操作
    protected Executor wrapper; // 其中封装的Executor对象
    
    // 延迟加载队列
    protected ConcurrentLinkedQueue<DeferredLoad> deferredLoads;
    // 下面两个属性是一级缓存用到的对象
    protected PerpetualCache localCache;
    protected PerpetualCache localOutputParameterCache;
    
    protected Configuration configuration;
    
    // 嵌套查询层级
    protected int queryStack;
    private boolean closed;
    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;

### **2.1 SimpleExecutor** ###

@Override
    public int doUpdate(MappedStatement ms, Object parameter) throws SQLException {
      Statement stmt = null;
      try {
        Configuration configuration = ms.getConfiguration();
        StatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null);
        stmt = prepareStatement(handler, ms.getStatementLog());
        return handler.update(stmt);
      } finally {
        closeStatement(stmt);
      }
    }
    
    @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.<E>query(stmt, resultHandler);
      } finally {
        closeStatement(stmt);//关闭statement对象
      }
    }

源码很简单，从configuration对象中去材料，交给handler去处理，处理完后，statement对象马上关闭。

### **2.2 ReuseExecutor** ###

执行器提供了Statement的重用功能，代码片段如下：

// 缓存使用过的Statement对象，key是SQL语句，value是SQL对应的Statement对象
    private final Map<String, Statement> statementMap = new HashMap<String, Statement>();
    /**
     * 准备获取Statement对象
     * @param handler
     * @param statementLog
     * @return
     * @throws SQLException
     */
    private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {
      Statement stmt;
      BoundSql boundSql = handler.getBoundSql();
      String sql = boundSql.getSql();
      if (hasStatementFor(sql)) {// 检测是否缓存了相同模式的SQL语句所对应的Statement对象
        stmt = getStatement(sql);// 从缓存中获取statement对象
        applyTransactionTimeout(stmt);// 修改超时时间
      } else {
        Connection connection = getConnection(statementLog);
        stmt = handler.prepare(connection, transaction.getTimeout());
        putStatement(sql, stmt);
      }
      handler.parameterize(stmt);
      return stmt;
    }

那statement对象是什么时候关闭的呢？当事物提交回滚或者关闭时都需要关闭这些缓存的Statement对象，在BaseExecutor.commit(),rollback(),close()方法中都会掉用doFlushStatement()方法，所以在改方法中实现关闭Statement对象是非常合适。具体如下：

@Override
    public List<BatchResult> doFlushStatements(boolean isRollback) throws SQLException {
      // 遍历map集合并关闭其中的Statement对象
      for (Statement stmt : statementMap.values()) {
        closeStatement(stmt);
      }
      // 清空缓存
      statementMap.clear();
      // 返回空集合
      return Collections.emptyList();
    }

### **2.3 BatchExecutor** ###

BatchExecutor实现了批处理多条SQL语句的功能，需要注意的是在批处理执行SQL语句时，每次向数据库发送的SQL语句条数是有上限，超过上限会抛出异常，所以批量发送SQL语句的时机是很重要的。

其中的核心字段含义如下：

//缓存多个Statement对象，每个Statement对象中都缓存了多条SQL语句
     private final List<Statement> statementList = new ArrayList<Statement>();
     //记录批处理的结果 BatchResult中通过updateCounts字段（int[]）记录每个Statement执行批处理的结果
     private final List<BatchResult> batchResultList = new ArrayList<BatchResult>();
     // 记录当前执行的SQL语句
     private String currentSql;
     // 记录当前的MappedStatement对象
     private MappedStatement currentStatement;

方法实现如下：

@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();
      // 本次执行的sql
      final String sql = boundSql.getSql();
      final Statement stmt;
      // 如果当前执行的SQL与上一次执行的SQL相同且对应的MappedStatement对象相同
      if (sql.equals(currentSql) && ms.equals(currentStatement)) {
        int last = statementList.size() - 1;
        // 已经存在Statement，取出最后一个Statement，有序
        stmt = statementList.get(last);
        applyTransactionTimeout(stmt);
       handler.parameterize(stmt);//fix Issues 322
        BatchResult batchResult = batchResultList.get(last);
        batchResult.addParameterObject(parameterObject);
      } else {
        // 尚不存在，新建Statement
        Connection connection = getConnection(ms.getStatementLog());
        stmt = handler.prepare(connection, transaction.getTimeout());
        handler.parameterize(stmt);    //fix Issues 322
        currentSql = sql;
        currentStatement = ms;
        // 放到statementList缓存
        statementList.add(stmt);
        batchResultList.add(new BatchResult(ms, sql, parameterObject));
      }
    // handler.parameterize(stmt);
        // 将sql以addBatch()的方式，添加到Statement中（该步骤由StatementHandler内部完成）
      handler.batch(stmt);
      return BATCH_UPDATE_RETURN_VALUE;
    }

需要注意的是sql.equals(currentSql)和statementList.get(last)，充分说明了其有序逻辑：AABB，将生成2个Statement对象；AABBAA，将生成3个Statement对象，而不是2个。因为，只要sql有变化，将导致生成新的Statement对象。

**缓存了这么多Statement批处理对象，何时执行它们？在doFlushStatements()方法中完成执行stmt.executeBatch()，随即关闭这些Statement对象。**

### **2.4 CachingExecutor** ###

**CachingExecutor**是一个Executor接口的装饰器，它为Executor对象增加了二级缓存的相关功能。

//委托的执行器对象，可以是SimpleExecutor、ReuseExecutor、BatchExecutor任一一个
    private final Executor delegate;
    //管理使用的二级缓存对像
    private final TransactionalCacheManager tcm = new TransactionalCacheManager();

query方法执行的查询操作步骤：

(1)获取BoundSql对象，创建查询语句对应的CacheKey对象，

(2)检测是否开启了二级缓存，如果没有，则指教调用delegate对象的query()方法查询，如果开启了，则继续后面的步骤

(3)检测查询是否包含输出类型的参数，如果是，则报错

(4)调用TransactionalCacheManager.getObject()方法查询二级缓存，如果二级缓存中查找到相应的结果，则直接返回结果。

(5)如果二级缓存没有相应的结果对象，在调用delegate对象的query()方法查询。最后将得到的结果放入

TransactionalCache.entriesToAddOnCommit集合中保存。

@Override
    public <E> List<E> query(MappedStatement ms, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException {
      // (1)
      BoundSql boundSql = ms.getBoundSql(parameterObject);
      CacheKey key = createCacheKey(ms, parameterObject, rowBounds, boundSql);
      return query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);
    }
    
    
    @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) {// (2)
        flushCacheIfRequired(ms);
        if (ms.isUseCache() && resultHandler == null) {
          // (3)
          ensureNoOutParams(ms, boundSql);
          // (4)
          @SuppressWarnings("unchecked")
          List<E> list = (List<E>) tcm.getObject(cache, key);
          if (list == null) {// (5)
    
            list = delegate.<E> query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);
            tcm.putObject(cache, key, list); // issue #578 and #116
          }
          return list;
        }
      }
      // 没有启动二级缓存，只调用底层Executor查询
      return delegate.<E> query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);
    }

[mybatis_ ---_Executor]: https://my.oschina.net/u/3737136/blog/1812422
[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9zdGF0aWMub3NjaGluYS5uZXQvdXBsb2Fkcy9zcGFjZS8yMDE4LzA1MTQvMTkwNjI4XzFINGRfMzczNzEzNi5wbmc?x-oss-process=image/format,png