MyBatis框架的使用及源码分析(十) CacheExecutor，SimpleExecutor，BatchExecutor ，ReuseExecutor

时间：2020-12-23 06:44:14

Executor分成两大类，一类是CacheExecutor，另一类是普通Executor。

普通类又分为：

ExecutorType.SIMPLE: 这个执行器类型不做特殊的事情。它为每个语句的执行创建一个新的预处理语句。
ExecutorType.REUSE: 这个执行器类型会复用预处理语句。
ExecutorType.BATCH: 这个执行器会批量执行所有更新语句,如果 SELECT 在它们中间执行还会标定它们是 必须的,来保证一个简单并易于理解的行为。

分别对应SimpleExecutor，ReuseExecutor，BatchExecutor，他们都继承于BaseExecutor，BatchExecutor专门用于执行批量sql操作，ReuseExecutor会重用statement执行sql操作，SimpleExecutor只是简单执行sql没有什么特别的。

CacheExecutor有一个重要属性delegate，它保存的是某类普通的Executor，值在构照时传入。执行数据库update操作时，它直接调用delegate的update方法，执行query方法时先尝试从cache中取值，取不到再调用delegate的查询方法，并将查询结果存入cache中。代码如下：

package org.apache.ibatis.executor;

import java.sql.SQLException;

import java.util.List;

import org.apache.ibatis.cache.Cache;

import org.apache.ibatis.cache.CacheKey;

import org.apache.ibatis.cache.TransactionalCacheManager;

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.session.ResultHandler;

import org.apache.ibatis.session.RowBounds;

import org.apache.ibatis.transaction.Transaction;

/**

 * @author Clinton Begin

 * @author Eduardo Macarron

 */

public class CachingExecutor implements Executor {

  private Executor delegate;

  private TransactionalCacheManager tcm = new TransactionalCacheManager();

  public CachingExecutor(Executor delegate) {

    this.delegate = delegate;

    delegate.setExecutorWrapper(this);

  }

  public Transaction getTransaction() {

    return delegate.getTransaction();

  }

  public void close(boolean forceRollback) {

    try {

      //issues #499, #524 and #573

      if (forceRollback) {

        tcm.rollback();

      } else {

        tcm.commit();

      }

    } finally {

      delegate.close(forceRollback);

    }

  }

  //是否关闭了executor 

  public boolean isClosed() {

    return delegate.isClosed();

  }

  public int update(MappedStatement ms, Object parameterObject) throws SQLException {

    flushCacheIfRequired(ms);//是否需要更缓存 

    return delegate.update(ms, parameterObject);//更新数据 

  }

  public <E> List<E> query(MappedStatement ms, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException {

    BoundSql boundSql = ms.getBoundSql(parameterObject);

    CacheKey key = createCacheKey(ms, parameterObject, rowBounds, boundSql);

    return query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);//获取记录

  }

  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, parameterObject, boundSql);

        @SuppressWarnings("unchecked")

        List<E> list = (List<E>) tcm.getObject(cache, key);//从缓存中获取数据 
        //如果为空，则查询数据库，并把结果保存到缓存中

        if (list == null) {

          list = delegate.<E> query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);

          tcm.putObject(cache, key, list); // issue #578. Query must be not synchronized to prevent deadlocks

        }

        return list;

      }

    }

    return delegate.<E> query(ms, parameterObject, rowBounds, resultHandler, key, boundSql);

  }

  public List<BatchResult> flushStatements() throws SQLException {

    return delegate.flushStatements();

  }

  public void commit(boolean required) throws SQLException {

    delegate.commit(required);

    tcm.commit();

  }

  public void rollback(boolean required) throws SQLException {

    try {

      delegate.rollback(required);

    } finally {

      if (required) {

        tcm.rollback();

      }

    }

  }

  private void ensureNoOutParams(MappedStatement ms, Object parameter, BoundSql boundSql) {

    if (ms.getStatementType() == StatementType.CALLABLE) {

      for (ParameterMapping parameterMapping : boundSql.getParameterMappings()) {

        if (parameterMapping.getMode() != ParameterMode.IN) {

          throw new ExecutorException("Caching stored procedures with OUT params is not supported.  Please configure useCache=false in " + ms.getId() + " statement.");

        }

      }

    }

  }

  public CacheKey createCacheKey(MappedStatement ms, Object parameterObject, RowBounds rowBounds, BoundSql boundSql) {

    return delegate.createCacheKey(ms, parameterObject, rowBounds, boundSql);

  }

  public boolean isCached(MappedStatement ms, CacheKey key) {

    return delegate.isCached(ms, key);

  }

  public void deferLoad(MappedStatement ms, MetaObject resultObject, String property, CacheKey key, Class<?> targetType) {

    delegate.deferLoad(ms, resultObject, property, key, targetType);

  }

  public void clearLocalCache() {

    delegate.clearLocalCache();

  }

  private void flushCacheIfRequired(MappedStatement ms) {

    Cache cache = ms.getCache();

    if (cache != null && ms.isFlushCacheRequired()) {

      tcm.clear(cache);

    }

  }

  @Override

  public void setExecutorWrapper(Executor executor) {

    throw new UnsupportedOperationException("This method should not be called");

  }

}

SimpleExecutor通过类名可以看出，它是一个简单的执行类，并不会做一些处理就执行sql，源码及分析如下：

package org.apache.ibatis.executor;

import java.sql.Connection;

import java.sql.SQLException;

import java.sql.Statement;

import java.util.Collections;

import java.util.List;

import org.apache.ibatis.executor.statement.StatementHandler;

import org.apache.ibatis.logging.Log;

import org.apache.ibatis.mapping.BoundSql;

import org.apache.ibatis.mapping.MappedStatement;

import org.apache.ibatis.session.Configuration;

import org.apache.ibatis.session.ResultHandler;

import org.apache.ibatis.session.RowBounds;

import org.apache.ibatis.transaction.Transaction;

/**

 * @author Clinton Begin

 */

public class SimpleExecutor extends BaseExecutor {

  public SimpleExecutor(Configuration configuration, Transaction transaction) {

    super(configuration, transaction);

  }

  public int doUpdate(MappedStatement ms, Object parameter) throws SQLException {

    Statement stmt = null;

    try {

      Configuration configuration = ms.getConfiguration();//获得配置
      //获得statementHandler里面有statement，来处理 

      StatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null);

      stmt = prepareStatement(handler, ms.getStatementLog());

      return handler.update(stmt);//最终是一个statement进行处理 

    } finally {

      closeStatement(stmt);

    }

  }

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

    }

  }

  public List<BatchResult> doFlushStatements(boolean isRollback) throws SQLException {

    return Collections.emptyList();

  }

  private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {

    Statement stmt;

    Connection connection = getConnection(statementLog);

    stmt = handler.prepare(connection);

    handler.parameterize(stmt);//将Statement转为PrepareStatement 

    return stmt;

  }

}

BatchExecutor : 通过批量操作来提高性能

package org.apache.ibatis.executor;

import java.sql.BatchUpdateException;

import java.sql.Connection;

import java.sql.SQLException;

import java.sql.Statement;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

import org.apache.ibatis.executor.keygen.Jdbc3KeyGenerator;

import org.apache.ibatis.executor.keygen.KeyGenerator;

import org.apache.ibatis.executor.keygen.NoKeyGenerator;

import org.apache.ibatis.executor.statement.StatementHandler;

import org.apache.ibatis.mapping.BoundSql;

import org.apache.ibatis.mapping.MappedStatement;

import org.apache.ibatis.session.Configuration;

import org.apache.ibatis.session.ResultHandler;

import org.apache.ibatis.session.RowBounds;

import org.apache.ibatis.transaction.Transaction;

/**

 * @author Jeff Butler

 */

public class BatchExecutor extends BaseExecutor {

  public static final int BATCH_UPDATE_RETURN_VALUE = Integer.MIN_VALUE + 1002;

  private final List<Statement> statementList = new ArrayList<Statement>();//Statement list

  private final List<BatchResult> batchResultList = new ArrayList<BatchResult>();//batch list

  private String currentSql;

  private MappedStatement currentStatement;

  public BatchExecutor(Configuration configuration, Transaction transaction) {

    super(configuration, transaction);

  }

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

    final String sql = boundSql.getSql();//获得Sql语句

    final Statement stmt;
    //如果sql语句等于当前sql MappedStatement 等于当前Map碰到Statement  

    if (sql.equals(currentSql) && ms.equals(currentStatement)) {

      int last = statementList.size() - 1;

      stmt = statementList.get(last);//获得最后一个
      //有相同的MappedStatement和参数

      BatchResult batchResult = batchResultList.get(last);

      batchResult.addParameterObject(parameterObject);

    } else {  
      //如果不存在就创建一个批处理操作 

      Connection connection = getConnection(ms.getStatementLog());

      stmt = handler.prepare(connection);

      currentSql = sql;

      currentStatement = ms;

      statementList.add(stmt);//添加批量处理操作

      batchResultList.add(new BatchResult(ms, sql, parameterObject));

    }

    handler.parameterize(stmt);

    handler.batch(stmt);//最终是调用jdbc的批处理操作 

    return BATCH_UPDATE_RETURN_VALUE;

  }

  public <E> List<E> doQuery(MappedStatement ms, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler, BoundSql boundSql)

      throws SQLException {

    Statement stmt = null;

    try {

      flushStatements();

      Configuration configuration = ms.getConfiguration();

      StatementHandler handler = configuration.newStatementHandler(wrapper, ms, parameterObject, rowBounds, resultHandler, boundSql);

      Connection connection = getConnection(ms.getStatementLog());

      stmt = handler.prepare(connection);

      handler.parameterize(stmt);

      return handler.<E>query(stmt, resultHandler);

    } finally {

      closeStatement(stmt);

    }

  }

  //刷新Statement，记录执行次数

  public List<BatchResult> doFlushStatements(boolean isRollback) throws SQLException {

    try {

      List<BatchResult> results = new ArrayList<BatchResult>();

      if (isRollback) {

        return Collections.emptyList();

      } else {
        //如果进行了批量处理

        for (int i = 0, n = statementList.size(); i < n; i++) {

          Statement stmt = statementList.get(i);

          BatchResult batchResult = batchResultList.get(i);

          try {
            //记录批量处理执行操作的条数

            batchResult.setUpdateCounts(stmt.executeBatch());

            MappedStatement ms = batchResult.getMappedStatement();

            List<Object> parameterObjects = batchResult.getParameterObjects();//参数对象集合

            KeyGenerator keyGenerator = ms.getKeyGenerator();//生成key 

            if (Jdbc3KeyGenerator.class.equals(keyGenerator.getClass())) {

              Jdbc3KeyGenerator jdbc3KeyGenerator = (Jdbc3KeyGenerator) keyGenerator;

              jdbc3KeyGenerator.processBatch(ms, stmt, parameterObjects);

            } else if (!NoKeyGenerator.class.equals(keyGenerator.getClass())) { //issue #141

              for (Object parameter : parameterObjects) {

                keyGenerator.processAfter(this, ms, stmt, parameter);

              }

            }

          } catch (BatchUpdateException e) {

            StringBuffer message = new StringBuffer();

            message.append(batchResult.getMappedStatement().getId())

                .append(" (batch index #")

                .append(i + 1)

                .append(")")

                .append(" failed.");

            if (i > 0) {

              message.append(" ")

                  .append(i)

                  .append(" prior sub executor(s) completed successfully, but will be rolled back.");

            }

            throw new BatchExecutorException(message.toString(), e, results, batchResult);

          }

          results.add(batchResult);//记录操作 

        }

        return results;

      }

    } finally {

      for (Statement stmt : statementList) {

        closeStatement(stmt);

      }

      currentSql = null;

      statementList.clear();

      batchResultList.clear();

    }

  }

}

ReuseExecutor: 重复使用执行，其定义了一个Map<String, Statement>，将执行的sql作为key，将执行的Statement作为value保存，这样执行相同的sql时就可以使用已经存在的Statement，就不需要新创建了，源码及分析如下：

package org.apache.ibatis.executor;

import java.sql.Connection;

import java.sql.SQLException;

import java.sql.Statement;

import java.util.Collections;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import org.apache.ibatis.executor.statement.StatementHandler;

import org.apache.ibatis.logging.Log;

import org.apache.ibatis.mapping.BoundSql;

import org.apache.ibatis.mapping.MappedStatement;

import org.apache.ibatis.session.Configuration;

import org.apache.ibatis.session.ResultHandler;

import org.apache.ibatis.session.RowBounds;

import org.apache.ibatis.transaction.Transaction;

/**

 * @author Clinton Begin

 */

public class ReuseExecutor extends BaseExecutor {

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

  public ReuseExecutor(Configuration configuration, Transaction transaction) {

    super(configuration, transaction);

  }

  public int doUpdate(MappedStatement ms, Object parameter) throws SQLException {

    Configuration configuration = ms.getConfiguration();

    StatementHandler handler = configuration.newStatementHandler(this, ms, parameter, RowBounds.DEFAULT, null, null);

    Statement stmt = prepareStatement(handler, ms.getStatementLog());//转换为PrepareStatement 

    return handler.update(stmt);

  }

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

    Configuration configuration = ms.getConfiguration();

    StatementHandler handler = configuration.newStatementHandler(wrapper, ms, parameter, rowBounds, resultHandler, boundSql);

    Statement stmt = prepareStatement(handler, ms.getStatementLog());

    return handler.<E>query(stmt, resultHandler);

  }

  public List<BatchResult> doFlushStatements(boolean isRollback) throws SQLException {

    for (Statement stmt : statementMap.values()) {

      closeStatement(stmt);//关闭Statement

    }

    statementMap.clear();//清空sql

    return Collections.emptyList();

  }

  private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {

    Statement stmt;

    BoundSql boundSql = handler.getBoundSql();

    String sql = boundSql.getSql();//获得sql语句

    if (hasStatementFor(sql)) {  //查看是否存在Statement  

      stmt = getStatement(sql);//如果存在就获取Statement  

    } else {

      Connection connection = getConnection(statementLog);

      stmt = handler.prepare(connection);//否则通过连接创建一个Statement 

      putStatement(sql, stmt);//将sql语句及对应的Statement 保存到map中

    }

    handler.parameterize(stmt);

    return stmt;

  }

  private boolean hasStatementFor(String sql) {

    try {
      //查看map中是否含有sql语句对应的Statement  

      return statementMap.keySet().contains(sql) && !statementMap.get(sql).getConnection().isClosed();

    } catch (SQLException e) {

      return false;

    }

  }

  //获得Sql语句对应的Statement 

  private Statement getStatement(String s) {

    return statementMap.get(s);

  }

  //将sql语句及对应的Statement保存到map中

  private void putStatement(String sql, Statement stmt) {

    statementMap.put(sql, stmt);

  }

}

相关文章