【MyBatis源码分析】insert方法、update方法、delete方法处理流程（上篇）

摘要

前文分析了MyBatis将配置文件转换为Java对象的流程，本文开始分析一下insert方法、update方法、delete方法处理的流程。

原文出处： 五月的仓颉

打开一个会话Session

前文分析了MyBatis将配置文件转换为Java对象的流程，本文开始分析一下insert方法、update方法、delete方法处理的流程，至于为什么这三个方法要放在一起说，是因为：

  • 从语义的角度，insert、update、delete都是属于对数据库的行进行更新操作
  • 从实现的角度，我们熟悉的PreparedStatement里面提供了两种execute方法，一种是executeUpdate()，一种是executeQuery()，前者对应的是insert、update与delete，后者对应的是select，因此对于MyBatis来说只有update与select

示例代码为这段：

public long insertMail(Mail mail) {
    SqlSession ss = ssf.openSession();
    try {
        int rows = ss.insert(NAME_SPACE + "insertMail", mail);
        ss.commit();
        if (rows > 0) {
            return mail.getId();
        }
        return 0;
    } catch (Exception e) {
        ss.rollback();
        return 0;
    } finally {
        ss.close();
    }
}

首先关注的是第2行的代码，ssf是SqlSessionFactory，其类型是DefaultSqlSessionFactory，上文最后已经分析过了，这里通过DefaultSqlSessionFactory来打开一个Session，通过Session去进行CRUD操作。

看一下openSession()方法的实现：

public SqlSession openSession() {
     return openSessionFromDataSource(configuration.getDefaultExecutorType(), null, false);
 }

顾名思义，从DataSource中获取Session，第一个参数的值是ExecutorType.SIMPLE，继续跟代码：

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

第4行的代码，获取配置的环境信息Environment。

第5行的代码，从Environment中获取事物工厂TransactionFactory，由于<environment>中配置的是”JDBC”，因此其真实类型是JdbcTransactionFactory，上文有说过。

第6行的代码，根据Environment中的DataSource（其实际类型是PooledDataSource）、TransactionIsolationLevel、autoCommit三个参数从TransactionFactory中获取一个事物，注意第三个参数autoCommit，它是openSession()方法中传过来的，其值为false，即MyBatis默认事物是不自动提交的。

第7行的代码，代码跟一下：

public Executor newExecutor(Transaction transaction, 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;
}

这里总结一下：

  • 根据ExecutorType获取一个执行器，这里是第10行的SimpleExecutor
  • 如果满足第12行的判断开启缓存功能，则执行第13行的代码。第13行的代码使用到了装饰器模式，传入Executor，给SimpleExecutor装饰上了缓存功能
  • 第15行的代码用于设置插件

这样就获取了一个Executor。最后将Executor、Configuration、autoCommit三个变量作为参数，实例化一个SqlSession出来，其实际类型为DefaultSqlSession。

insert方法执行流程

在看了openSession()方法知道最终获得了一个DefaultSqlSession之后，看一下DefaultSqlSession的insert方法是如何实现的：

 public int insert(String statement, Object parameter) {
      return update(statement, parameter);
  }

看到虽然调用的是insert方法，但是最终统一都会去执行update方法，delete方法也是如此，这个开头已经说过了，这里证明了这一点。

接着继续看第2行的方法实现：

public int update(String statement, Object parameter) {
    try {
      dirty = true;
      MappedStatement ms = configuration.getMappedStatement(statement);
      return executor.update(ms, wrapCollection(parameter));
    } catch (Exception e) {
      throw ExceptionFactory.wrapException("Error updating database.  Cause: " + e, e);
    } finally {
      ErrorContext.instance().reset();
    }
}

第4行的代码根据statement从Configuration中获取MappedStatement，MappedStatement上文已经分析过了，存储在Configuration的mappedStatements字段中。

第5行的代码分为两部分，首先wrapCollection，顾名思义包装集合类，源码为：

private Object wrapCollection(final Object object) {
    if (object instanceof Collection) {
      StrictMap<Object> map = new StrictMap<Object>();
      map.put("collection", object);
      if (object instanceof List) {
        map.put("list", object);
      }
      return map;
    } else if (object != null && object.getClass().isArray()) {
      StrictMap<Object> map = new StrictMap<Object>();
      map.put("array", object);
      return map;
    }
    return object;
}

这里做了三层处理：

  • 如果参数是Collection（即集合）类型，放一个key为”collection”、value为参数的键值对
  • 如果参数是List类型，放一个key为”list”、value为参数的键值对
  • 如果参数是数组类型，放一个key为”array”、value为参数的键值对

将集合进行包装之后，就可以执行Executor的update方法了，Executor上面说了，是使用装饰器模式将SimpleExecutor加上了缓存功能的CacheExecutor，它的update方法实现为：

public int update(MappedStatement ms, Object parameterObject) throws SQLException {
     flushCacheIfRequired(ms);
     return delegate.update(ms, parameterObject);
}

第2行的代码是判断是否要求清缓存的，这里首先我们的示例配置文件mail.xml中没有配置<cache>，其次<insert>、<delete>、<update>、<select>中没有配置flushCache=”true”属性，因此这一句代码不会执行任何操作。

第3行的代码delegate就是SimpleExecutor本身，因为是装饰器模式，因此会持有接口的引用，deletegate其类型就是Executor。继续跟代码，看一下update方法：

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

前面的没什么好看的，继续跟第7行的代码：

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

第4行的代码获取MappedStatement中的Configuration对象。

第5行的代码获取Statement处理器StatementHandler接口实现类，Statement是Java原生的为JDBC设计的声明，StatementHandler接口实现类的真实类型为RoutingStatementHandler。

第6行和第7行的代码后文逐步分析，因为里面一点一点封装了我们平时写JDBC时的一些基本步骤，比如获取Connection，构建PreparedStatement、对execute后的结果进行处理等，先看一下prepareStatement的源码：

private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {
    Statement stmt;
    Connection connection = getConnection(statementLog);
    stmt = handler.prepare(connection, transaction.getTimeout());
    handler.parameterize(stmt);
    return stmt;
}

后面逐步分析。

获取Connection

第一步，看下获取Connection的步骤。看一下上面getConnection方法如何实现：

protected Connection getConnection(Log statementLog) throws SQLException {
    Connection connection = transaction.getConnection();
    if (statementLog.isDebugEnabled()) {
      return ConnectionLogger.newInstance(connection, statementLog, queryStack);
    } else {
      return connection;
    }
}

Connection从Transaction中获取，配置的是JDBC，这里代码进入JdbcTransaction的getConnection()：

protected Connection getConnection(Log statementLog) throws SQLException {
    Connection connection = transaction.getConnection();
    if (statementLog.isDebugEnabled()) {
      return ConnectionLogger.newInstance(connection, statementLog, queryStack);
    } else {
      return connection;
    }
}

先看一下第3行~第7行的代码，判断的意思是是否开启Statement的表达式，如果开启，那么第4行会给生成的Connection加上一个代理，代理的内容是在调用prepareStatement、prepareCall等方法前或者方法后打印日志，具体可见ConnectionLogger、PreparedStatementLogger、ResultSetLogger与StatementLogger的invoke方法。

接着继续跟第2行的代码：

public Connection getConnection() throws SQLException {
     if (connection == null) {
       openConnection();
     }
     return connection;
}

跟一下第3行的代码：

protected void openConnection() throws SQLException {
    if (log.isDebugEnabled()) {
      log.debug("Opening JDBC Connection");
    }
    connection = dataSource.getConnection();
    if (level != null) {
      connection.setTransactionIsolation(level.getLevel());
    }
    setDesiredAutoCommit(autoCommmit);
}

第6行~第8行的代码用于设置事物隔离级别，第9行的代码用于设置是否自动提交事物。下面跟一下第5行的代码getConnection()方法：

public Connection getConnection() throws SQLException {
     return popConnection(dataSource.getUsername(), dataSource.getPassword()).getProxyConnection();
}

这里简单提一下，在方法名中如果看到了”pop”、”push”字样，一定要把该方法使用的数据结构和栈联想起来，栈（stack）是一个先进先出数据结构，”pop”、”push”是栈特有的操作，前者将栈顶的数据推送出栈让调用者获取到，后者将数据压入栈顶。

后面的getProxyConnection()方法就是将获取到的Connection返回而已，没什么特殊的操作，这里跟一下popConnection方法实现，它位于PooledDataSource类中，这是由<dataSource>标签中的type属性决定的：

private PooledConnection popConnection(String username, String password) throws SQLException {
    boolean countedWait = false;
    PooledConnection conn = null;
    long t = System.currentTimeMillis();
    int localBadConnectionCount = 0;

    while (conn == null) {
      synchronized (state) {
        if (!state.idleConnections.isEmpty()) {
          // Pool has available connection
          conn = state.idleConnections.remove(0);
          if (log.isDebugEnabled()) {
            log.debug("Checked out connection " + conn.getRealHashCode() + " from pool.");
          }
        } else {
          // Pool does not have available connection
          if (state.activeConnections.size() < poolMaximumActiveConnections) {
            // Can create new connection
            conn = new PooledConnection(dataSource.getConnection(), this);
            if (log.isDebugEnabled()) {
              log.debug("Created connection " + conn.getRealHashCode() + ".");
            }
          } else {
            // Cannot create new connection
            PooledConnection oldestActiveConnection = state.activeConnections.get(0);
            long longestCheckoutTime = oldestActiveConnection.getCheckoutTime();
            if (longestCheckoutTime > poolMaximumCheckoutTime) {
              // Can claim overdue connection
              state.claimedOverdueConnectionCount++;
              state.accumulatedCheckoutTimeOfOverdueConnections += longestCheckoutTime;
              state.accumulatedCheckoutTime += longestCheckoutTime;
              state.activeConnections.remove(oldestActiveConnection);
              if (!oldestActiveConnection.getRealConnection().getAutoCommit()) {
                try {
                  oldestActiveConnection.getRealConnection().rollback();
                } catch (SQLException e) {
                  log.debug("Bad connection. Could not roll back");
                }  
              }
              conn = new PooledConnection(oldestActiveConnection.getRealConnection(), this);
              oldestActiveConnection.invalidate();
              if (log.isDebugEnabled()) {
                log.debug("Claimed overdue connection " + conn.getRealHashCode() + ".");
              }
            } else {
              // Must wait
              try {
                if (!countedWait) {
                  state.hadToWaitCount++;
                  countedWait = true;
                }
                if (log.isDebugEnabled()) {
                  log.debug("Waiting as long as " + poolTimeToWait + " milliseconds for connection.");
                }
                long wt = System.currentTimeMillis();
                state.wait(poolTimeToWait);
                state.accumulatedWaitTime += System.currentTimeMillis() - wt;
              } catch (InterruptedException e) {
                break;
              }
            }
          }
        }
        if (conn != null) {
          if (conn.isValid()) {
            if (!conn.getRealConnection().getAutoCommit()) {
              conn.getRealConnection().rollback();
            }
            conn.setConnectionTypeCode(assembleConnectionTypeCode(dataSource.getUrl(), username, password));
            conn.setCheckoutTimestamp(System.currentTimeMillis());
            conn.setLastUsedTimestamp(System.currentTimeMillis());
            state.activeConnections.add(conn);
            state.requestCount++;
            state.accumulatedRequestTime += System.currentTimeMillis() - t;
          } else {
            if (log.isDebugEnabled()) {
              log.debug("A bad connection (" + conn.getRealHashCode() + ") was returned from the pool, getting another connection.");
            }
            state.badConnectionCount++;
            localBadConnectionCount++;
            conn = null;
            if (localBadConnectionCount > (poolMaximumIdleConnections + 3)) {
              if (log.isDebugEnabled()) {
                log.debug("PooledDataSource: Could not get a good connection to the database.");
              }
              throw new SQLException("PooledDataSource: Could not get a good connection to the database.");
            }
          }
        }
      }

    }

    if (conn == null) {
      if (log.isDebugEnabled()) {
        log.debug("PooledDataSource: Unknown severe error condition.  The connection pool returned a null connection.");
      }
      throw new SQLException("PooledDataSource: Unknown severe error condition.  The connection pool returned a null connection.");
    }

    return conn;
}

这段方法很长，分解一下。

首先是第9行~第15行的判断，假使空闲的Connection列表不是空的，Connection就是空闲Connection列表的第一个Connection，且移除空闲Connection列表的第一个Connection，这也符合PooledDataSource的定义，有一个Connection池，对Connection进行复用而不是每次都new出来，这就是典型的栈的操作。但是这里有一点我认为MyBatis写得不是很好，List的实际类型是ArrayList，每次的移除操作是remove(0)，ArrayList处理remove效率并不高尤其还是remove(0)的操作，因此这里替换成LinkedList会更好一些。

接着先看第23行~第63行的判断，它的判断逻辑是假如当前在使用的Connection数量大于或等于最大可用的Connection数量，那么获取当前正在使用的Connection列表中的第一个Connection做一个判断：

  • 如果当前Connection执行时间已经超过了指定的Connection最大超时时间，那么原Connection如果不是自动Commit的，数据回滚，新建一个Connection，原Connection失效
  • 如果当前Connection执行时间没有超过指定的Connection最大超时时间，那么使用wait方法等待

最后回到第17行~第23行的判断，即当前在使用的Connection数量小于最大可用的Connection数量，那么此时直接new一个PooledConnection出来，看一下PooledDataSource的getConnection()方法实现：

public Connection getConnection() throws SQLException {
     return doGetConnection(username, password);
}

继续跟代码doGetConnection方法：

private Connection doGetConnection(String username, String password) throws SQLException {
    Properties props = new Properties();
    if (driverProperties != null) {
      props.putAll(driverProperties);
    }
    if (username != null) {
      props.setProperty("user", username);
    }
    if (password != null) {
      props.setProperty("password", password);
    }
    return doGetConnection(props);
}

这里就是先设置一下配置的属性，继续跟第12行的方法实现：

private Connection doGetConnection(Properties properties) throws SQLException {
     initializeDriver();
     Connection connection = DriverManager.getConnection(url, properties);
     configureConnection(connection);
     return connection;
}

到了这里就是我们比较熟悉的代码了。

第2行的代码意思是MyBatis维护了一个Driver池registeredDrivers，如果我们的Driver不在Driver池里面，那么会尝试使用Class.forName方法初始化一下，成功的话加入Driver池中。

第3行的代码不说了，使用DriverManager的getConnection方法获取Connection，第4行的代码配置一下Connection，主要就是设置一下自动提交属性与事物隔离级别。

最后将生成的Connection返回出去，完成生成Connection的流程。

为Connection生成代理

上面解析了生成Connection的流程，代码到这里还没完还有一步，看一下PooledConnection的构造方法：

public PooledConnection(Connection connection, PooledDataSource dataSource) {
    this.hashCode = connection.hashCode();
    this.realConnection = connection;
    this.dataSource = dataSource;
    this.createdTimestamp = System.currentTimeMillis();
    this.lastUsedTimestamp = System.currentTimeMillis();
    this.valid = true;
    this.proxyConnection = (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), IFACES, this);
}

这里第8行的代码会为生成的Connection创建一个代理，PooledConnection本身就实现了InvocationHandler接口，看一下代理内容是什么，invoke方法的实现：

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    String methodName = method.getName();
    if (CLOSE.hashCode() == methodName.hashCode() && CLOSE.equals(methodName)) {
      dataSource.pushConnection(this);
      return null;
    } else {
      try {
        if (!Object.class.equals(method.getDeclaringClass())) {
          // issue #579 toString() should never fail
          // throw an SQLException instead of a Runtime
          checkConnection();
        }
        return method.invoke(realConnection, args);
      } catch (Throwable t) {
        throw ExceptionUtil.unwrapThrowable(t);
      }
    }
}

这一步操作主要是为了处理close方法的，看一下pushConnection方法的实现：

protected void pushConnection(PooledConnection conn) throws SQLException {

    synchronized (state) {
      state.activeConnections.remove(conn);
      if (conn.isValid()) {
        if (state.idleConnections.size() < poolMaximumIdleConnections && conn.getConnectionTypeCode() == expectedConnectionTypeCode) {
          state.accumulatedCheckoutTime += conn.getCheckoutTime();
          if (!conn.getRealConnection().getAutoCommit()) {
            conn.getRealConnection().rollback();
          }
          PooledConnection newConn = new PooledConnection(conn.getRealConnection(), this);
          state.idleConnections.add(newConn);
          newConn.setCreatedTimestamp(conn.getCreatedTimestamp());
          newConn.setLastUsedTimestamp(conn.getLastUsedTimestamp());
          conn.invalidate();
          if (log.isDebugEnabled()) {
            log.debug("Returned connection " + newConn.getRealHashCode() + " to pool.");
          }
          state.notifyAll();
        } else {
          state.accumulatedCheckoutTime += conn.getCheckoutTime();
          if (!conn.getRealConnection().getAutoCommit()) {
            conn.getRealConnection().rollback();
          }
          conn.getRealConnection().close();
          if (log.isDebugEnabled()) {
            log.debug("Closed connection " + conn.getRealHashCode() + ".");
          }
          conn.invalidate();
        }
      } else {
        if (log.isDebugEnabled()) {
          log.debug("A bad connection (" + conn.getRealHashCode() + ") attempted to return to the pool, discarding connection.");
        }
        state.badConnectionCount++;
      }
    }
}

代码的逻辑简单来说就是当调用close方法的时候，如果当前空闲Connection列表中的Connection数量小于指定空闲Connection列表中的数量（第二个判断connectionTypeCode的值为275950209，不知道是干什么的），那么会为原Connection生成一个PooledConnection并加入空闲Connection列表中。

如果不满足上面的条件，那么就直接调用Connection的close()方法并且让原Connection失效。

文章标签：
原创文章：转载请注明原创作者！
本文标题：【MyBatis源码分析】insert方法、update方法、delete方法处理流程（上篇）
永久链接：https://itwgy.com/html/code/543.html
版权信息
IT家园
IT家园

网友最新评论 (0)

发表我的评论
取消评论
表情