MyBatis框架的使用及源碼分析(九) Executor

時間 2019-12-02

標籤 mybatis 框架使用源碼分析 executor 欄目 MyBatis 简体版

原文原文鏈接

從<MyBatis框架的使用及源碼分析(八) MapperMethod>文中咱們知道執行Mapper的每個接口方法，最後調用的是MapperMethod.execute方法。而當執行MapperMethod的execute方法的時候，根據當前MapperMethod對應的mapper配置會執行Session的insert, update, delete, select, selectList, selectMap, selectCursor, selectOne或flushStatements方法。html

咱們來看DefaultSqlSession一個具體的實現方法：java

  public <E> List<E> selectList(String statement, Object parameter, RowBounds rowBounds) {
    try {
      MappedStatement ms = configuration.getMappedStatement(statement);
      List<E> result = executor.query(ms, wrapCollection(parameter), rowBounds, Executor.NO_RESULT_HANDLER);
      return result;
    } catch (Exception e) {
      throw ExceptionFactory.wrapException("Error querying database.  Cause: " + e, e);
    } finally {
      ErrorContext.instance().reset();
    }
  }

咱們看到，真正執行的代碼主體在Executor對象裏面。sql

DefaultSqlSession類的成員executor是在構造函數裏面給他賦值的。因此咱們又要回頭去查看一下是在何時實例化了DefaultSqlSession類。數據庫

DefaultSqlSession在SqlSessionFactory的實現類DefaultSqlSessionFactory中被建立：apache

private SqlSession openSessionFromDataSource(ExecutorType execType, TransactionIsolationLevel level, boolean autoCommit) {    
  Transaction tx = null;    
  try {    
    final Environment environment = configuration.getEnvironment();    
    final TransactionFactory transactionFactory = getTransactionFactoryFromEnvironment(environment);    
    tx = transactionFactory.newTransaction(environment.getDataSource(), level, autoCommit);    
    final Executor executor = configuration.newExecutor(tx, execType);    
    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();    
  }    
}    
    
private SqlSession openSessionFromConnection(ExecutorType execType, Connection connection) {    
  try {    
    boolean autoCommit;    
    try {    
      autoCommit = connection.getAutoCommit();    
    } catch (SQLException e) {    
      // Failover to true, as most poor drivers    
      // or databases won't support transactions    
      autoCommit = true;    
    }          
    final Environment environment = configuration.getEnvironment();    
    final TransactionFactory transactionFactory = getTransactionFactoryFromEnvironment(environment);    
    final Transaction tx = transactionFactory.newTransaction(connection);    
    final Executor executor = configuration.newExecutor(tx, execType);    
    return new DefaultSqlSession(configuration, executor, autoCommit);    
  } catch (Exception e) {    
    throw ExceptionFactory.wrapException("Error opening session.  Cause: " + e, e);    
  } finally {    
    ErrorContext.instance().reset();    
  }    
}

從源碼中咱們能夠看到他是經過Configuration類的newExecutor方法來獲得的，代碼以下：緩存

public Executor newExecutor(Transaction transaction, ExecutorType 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對外統一提供了一個操做接口類Executor，提供的接口方法有update、query、flushStatements、commit、rollback等接口函數，源碼以下：安全

package org.apache.ibatis.executor;

import java.sql.SQLException;
import java.util.List;

import org.apache.ibatis.cache.CacheKey;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.reflection.MetaObject;
import org.apache.ibatis.session.ResultHandler;
import org.apache.ibatis.session.RowBounds;
import org.apache.ibatis.transaction.Transaction;

/**
 * @author Clinton Begin
 */
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;

  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();
  //關閉鏈接
  void close(boolean forceRollback);
  //Executor是否關閉
  boolean isClosed();
  
  void setExecutorWrapper(Executor executor);

}

具體實現類有抽象類BaseExecutor、實現類CachingExecutor、實現類BatchExecutor、實現類ReuseExecutor和實現類SimpleExecutor。session

具體選用哪一個子類SimpleExecutor、ReuseExecutor和BatchExecutor實現，能夠在Mybatis的配置文件中進行配，配置以下：mybatis

<settings>  
    <setting name="defaultExecutorType" value="REUSE"/> <!--SIMPLE、REUSE、BATCH-->  
</settings>

配置以後在Configuration類中的newExecutor（）函數會選擇具體使用的子類。app

抽象類BaseExecutor的實現源碼及解析以下以下：

package org.apache.ibatis.executor;

import static org.apache.ibatis.executor.ExecutionPlaceholder.EXECUTION_PLACEHOLDER;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.ibatis.cache.CacheKey;
import org.apache.ibatis.cache.impl.PerpetualCache;
import org.apache.ibatis.logging.Log;
import org.apache.ibatis.logging.LogFactory;
import org.apache.ibatis.logging.jdbc.ConnectionLogger;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.mapping.ParameterMapping;
import org.apache.ibatis.mapping.ParameterMode;
import org.apache.ibatis.mapping.StatementType;
import org.apache.ibatis.reflection.MetaObject;
import org.apache.ibatis.reflection.factory.ObjectFactory;
import org.apache.ibatis.session.Configuration;
import org.apache.ibatis.session.LocalCacheScope;
import org.apache.ibatis.session.ResultHandler;
import org.apache.ibatis.session.RowBounds;
import org.apache.ibatis.transaction.Transaction;
import org.apache.ibatis.type.TypeHandlerRegistry;

/**
 * @author Clinton Begin
 */
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的配置信息
  protected Configuration configuration;
  //查詢堆棧
  protected int queryStack = 0;
  private boolean closed;

  protected BaseExecutor(Configuration configuration, Transaction transaction) {
    this.transaction = transaction;
    this.deferredLoads = new ConcurrentLinkedQueue<DeferredLoad>();
    this.localCache = new PerpetualCache("LocalCache");
    this.localOutputParameterCache = new PerpetualCache("LocalOutputParameterCache");
    this.closed = false;
    this.configuration = configuration;
    this.wrapper = this;
  }

  public Transaction getTransaction() {
    if (closed) throw new ExecutorException("Executor was closed.");
    return transaction;
  }

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

  public boolean isClosed() {
    return closed;
  }

   //SqlSession的update/insert/delete會調用此方法
  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);
  }

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

  //SqlSession.selectList會調用此方法 
  public <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException {
    BoundSql boundSql = ms.getBoundSql(parameter);//獲得綁定sql  
    CacheKey key = createCacheKey(ms, parameter, rowBounds, boundSql);//建立緩存key
    return query(ms, parameter, rowBounds, resultHandler, key, boundSql);//查詢  
 }

  @SuppressWarnings("unchecked")
  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.");
    //先清局部緩存，再查詢，但僅僅查詢堆棧爲0才清，爲了處理遞歸調用 
    if (queryStack == 0 && ms.isFlushCacheRequired()) {
      clearLocalCache();
    }
    List<E> list;
    try {
      queryStack++;//加一，這樣遞歸調用到上面的時候就不會再清局部緩存了 
      //根據cachekey從localCache去查 
      list = resultHandler == null ? (List<E>) localCache.getObject(key) : null;
      if (list != null) {
        //若是查到localCache緩存，處理localOutputParameterCache  
        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();
      }
       //清空延遲加載隊列
      deferredLoads.clear(); // issue #601
      if (configuration.getLocalCacheScope() == LocalCacheScope.STATEMENT) {
        //若是是statement，清本地緩存  
        clearLocalCache(); // issue #482
      }
    }
    return list;
  }

  //延遲加載 
  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 {
        deferredLoads.add(new DeferredLoad(resultObject, property, key, localCache, configuration, targetType));
    }
  }

  //建立緩存key 
  public CacheKey createCacheKey(MappedStatement ms, Object parameterObject, RowBounds rowBounds, BoundSql boundSql) {
    if (closed) throw new ExecutorException("Executor was closed.");
    CacheKey cacheKey = new CacheKey();
    //MyBatis 對於其 Key 的生成採起規則爲：[mappedStementId + offset + limit + SQL + queryParams + environment]生成一個哈希碼
    cacheKey.update(ms.getId());
    cacheKey.update(rowBounds.getOffset());
    cacheKey.update(rowBounds.getLimit());
    cacheKey.update(boundSql.getSql());
    List<ParameterMapping> parameterMappings = boundSql.getParameterMappings();
    TypeHandlerRegistry typeHandlerRegistry = ms.getConfiguration().getTypeHandlerRegistry();
    for (int i = 0; i < parameterMappings.size(); i++) { // mimic DefaultParameterHandler logic
      ParameterMapping parameterMapping = parameterMappings.get(i);
      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);
      }
    }
    return cacheKey;
  }    

  public boolean isCached(MappedStatement ms, CacheKey key) {
    return localCache.getObject(key) != null;
  }

  public void commit(boolean required) throws SQLException {
    if (closed) throw new ExecutorException("Cannot commit, transaction is already closed");
    clearLocalCache();
    flushStatements();
    if (required) {
      transaction.commit();
    }
  }

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

  //清空本地緩存，一個map結構  
  public void clearLocalCache() {
    if (!closed) {
      localCache.clear();
      localOutputParameterCache.clear();
    }
  }

  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 void closeStatement(Statement statement) {
    if (statement != null) {
      try {
        statement.close();
      } catch (SQLException e) {
        // ignore
      }
    }
  }

  //處理存儲過程的out參數
  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);
          }
        }
      }
    }
  }

  //從數據庫中查 
  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) {//若是是存儲過程，OUT參數也加入緩存
      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;
    }
  }
  
  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;

    public DeferredLoad(MetaObject resultObject,
                        String property,
                        CacheKey key,
                        PerpetualCache localCache,
                        Configuration configuration,
                        Class<?> targetType) { // issue #781
      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);
    }

  }

}