源碼解析MyBatis Sharding-Jdbc SQL語句執行流程詳解(文末有流程圖)

時間 2019-11-30

標籤源碼解析 mybatis sharding jdbc sql 語句執行流程詳解流程圖欄目 MyBatis 简体版

原文原文鏈接

本文將詳細介紹Mybatis SQL語句執行的全流程，本文與上篇具備必定的關聯性，建議先閱讀該系列中的前面3篇文章，重點掌握Mybatis Mapper類的初始化過程，由於在Mybatis中，Mapper是執行SQL語句的入口，相似下面這段代碼：java

@Service
public UserService implements IUserService {
 	@Autowired
    private UserMapper userMapper;
    public User findUser(Integer id) {
        return userMapper.find(id);
    }
}
複製代碼

開始進入本文的主題，以源碼爲手段，分析Mybatis執行SQL語句的流行，而且使用了數據庫分庫分表中間件sharding-jdbc,其版本爲sharding-jdbc1.4.1。spring

爲了方便你們對本文的源碼分析，先給出Mybatis層面核心類的方法調用序列圖。sql

一、SQL執行序列圖

二、源碼解析SQL執行流程

接下來從從源碼的角度對其進行剖析。數據庫

舒適提示：在本文的末尾，還會給出一張詳細的Mybatis Shardingjdbc語句執行流程圖。（請勿錯過哦）。緩存

2.1 MapperProxy#invoker

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    if (Object.class.equals(method.getDeclaringClass())) {
      try {
        return method.invoke(this, args);
      } catch (Throwable t) {
        throw ExceptionUtil.unwrapThrowable(t);
      }
    }
    final MapperMethod mapperMethod = cachedMapperMethod(method);   // @1
    return mapperMethod.execute(sqlSession, args);                                     // @2
  }
複製代碼

代碼@1：建立並緩存MapperMethod對象。mybatis

代碼@2：調用MapperMethod對象的execute方法，即mapperInterface中定義的每個方法最終會對應一個MapperMethod。併發

2.2 MapperMethod#execute

public Object execute(SqlSession sqlSession, Object[] args) {
    Object result;
    if (SqlCommandType.INSERT == command.getType()) { 
      Object param = method.convertArgsToSqlCommandParam(args);
      result = rowCountResult(sqlSession.insert(command.getName(), param));
    } else if (SqlCommandType.UPDATE == command.getType()) {
      Object param = method.convertArgsToSqlCommandParam(args);
      result = rowCountResult(sqlSession.update(command.getName(), param));
    } else if (SqlCommandType.DELETE == command.getType()) {
      Object param = method.convertArgsToSqlCommandParam(args);
      result = rowCountResult(sqlSession.delete(command.getName(), param));
    } else if (SqlCommandType.SELECT == command.getType()) {
      if (method.returnsVoid() && method.hasResultHandler()) {
        executeWithResultHandler(sqlSession, args);
        result = null;
      } else if (method.returnsMany()) {
        result = executeForMany(sqlSession, args);
      } else if (method.returnsMap()) {
        result = executeForMap(sqlSession, args);
      } else {
        Object param = method.convertArgsToSqlCommandParam(args);
        result = sqlSession.selectOne(command.getName(), param);
      }
    } else {
      throw new BindingException("Unknown execution method for: " + command.getName());
    }
    if (result == null && method.getReturnType().isPrimitive() && !method.returnsVoid()) {
      throw new BindingException("Mapper method '" + command.getName() 
          + " attempted to return null from a method with a primitive return type (" + method.getReturnType() + ").");
    }
    return result;
  }
複製代碼

該方法主要是根據SQL類型，insert、update、select等操做，執行對應的邏輯，本文咱們以查詢語句，進行跟蹤，進入executeForMany(sqlSession, args)方法。app

2.3 MapperMethod#executeForMany

private <E> Object executeForMany(SqlSession sqlSession, Object[] args) {
    List<E> result;
    Object param = method.convertArgsToSqlCommandParam(args);
    if (method.hasRowBounds()) {
      RowBounds rowBounds = method.extractRowBounds(args);
      result = sqlSession.<E>selectList(command.getName(), param, rowBounds);
    } else {
      result = sqlSession.<E>selectList(command.getName(), param);
    }
    // issue #510 Collections & arrays support
    if (!method.getReturnType().isAssignableFrom(result.getClass())) {
      if (method.getReturnType().isArray()) {
        return convertToArray(result);
      } else {
        return convertToDeclaredCollection(sqlSession.getConfiguration(), result);
      }
    }
    return result;
  }
複製代碼

該方法也比較簡單，最終經過SqlSession調用selectList方法。框架

2.4 DefaultSqlSession#selectList

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

代碼@1：根據資源名稱獲取對應的MappedStatement對象，此時的statement爲資源名稱，例如com.demo.UserMapper.findUser。至於MappedStatement對象的生成在上一節初始化時已詳細介紹過，此處再也不重複介紹。異步

代碼@2：調用Executor的query方法。這裏說明一下，其實一開始會進入到CachingExecutor#query方法，因爲CachingExecutor的Executor delegate屬性默認是SimpleExecutor，故最終仍是會進入到SimpleExecutor#query中。

接下來咱們進入到SimpleExecutor的父類BaseExecutor的query方法中。

2.5 BaseExecutor#query

public <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey key, BoundSql boundSql) throws SQLException {   // @1
    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;                                            // @2
      if (list != null) {
        handleLocallyCachedOutputParameters(ms, key, parameter, boundSql);
      } else {
        list = queryFromDatabase(ms, parameter, rowBounds, resultHandler, key, boundSql);                   // @3
      }
    } finally {
      queryStack--;
    }
    if (queryStack == 0) {
      for (DeferredLoad deferredLoad : deferredLoads) {
        deferredLoad.load();
      }
      deferredLoads.clear(); // issue #601
      if (configuration.getLocalCacheScope() == LocalCacheScope.STATEMENT) {                         // @4
        clearLocalCache(); // issue #482
      }
    }
    return list;
  }
複製代碼

代碼@1：首先介紹一下該方法的入參，這些類都是Mybatis的重要類：

MappedStatement ms 映射語句，一個MappedStatemnet對象表明一個Mapper中的一個方法，是映射的最基本對象。
Object parameter SQL語句的參數列表。
RowBounds rowBounds 行邊界對象，其實就是分頁參數limit與size。
ResultHandler resultHandler 結果處理Handler。
CacheKey key Mybatis緩存Key
BoundSql boundSql SQL與參數綁定信息，從該對象能夠獲取在映射文件中的SQL語句。

代碼@2：首先從緩存中獲取，Mybatis支持一級緩存（SqlSession）與二級緩存（多個SqlSession共享）。

代碼@3：從數據庫查詢結果，而後進入到doQuery方法，執行真正的查詢動做。

代碼@4：若是一級緩存是語句級別的，則語句執行完畢後，刪除緩存。

2.6 SimpleExecutor#doQuery

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);   // @1
      stmt = prepareStatement(handler, ms.getStatementLog());                                                                                                                   // @2
      return handler.<E>query(stmt, resultHandler);                                                                                                                                        // @3
    } finally {
      closeStatement(stmt);
    }
  }
複製代碼

代碼@1：建立StatementHandler，這裏會加入Mybatis的插件擴展機制(將在下篇詳細介紹)，如圖所示：

代碼@2：建立Statement對象，注意，這裏就是JDBC協議的java.sql.Statement對象了。

代碼@3：使用Statment對象執行SQL語句。

接下來詳細介紹Statement對象的建立過程與執行過程，即分佈詳細跟蹤代碼@2與代碼@3。

三、Statement對象建立流程

3.1 java.sql.Connection對象建立

3.1.1 SimpleExecutor#prepareStatement

private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException {
    Statement stmt;
    Connection connection = getConnection(statementLog);  // @1
    stmt = handler.prepare(connection);                                  // @2
    handler.parameterize(stmt);                                               // @3
    return stmt;
}
複製代碼

建立Statement對象，分紅三步：代碼@1：建立java.sql.Connection對象。

代碼@2：使用Connection對象建立Statment對象。

代碼@3：對Statement進行額外處理，特別是PrepareStatement的參數設置(ParameterHandler)。

3.1.2 SimpleExecutor#getConnection

getConnection方法，根據上面流程圖所示，先是進入到org.mybatis.spring.transaction.SpringManagedTransaction，再經過spring-jdbc框架，利用DataSourceUtils獲取鏈接，其代碼以下：

public static Connection doGetConnection(DataSource dataSource) throws SQLException {  
		Assert.notNull(dataSource, "No DataSource specified");
		ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource); 
		if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
			conHolder.requested();
			if (!conHolder.hasConnection()) {
				conHolder.setConnection(dataSource.getConnection());
			}
			return conHolder.getConnection();
		}
		// Else we either got no holder or an empty thread-bound holder here.

		logger.debug("Fetching JDBC Connection from DataSource");
		Connection con = dataSource.getConnection();      // @1

        // 這裏省略與事務處理相關的代碼
		return con;
	}
複製代碼

代碼@1：經過DataSource獲取connection，那此處的DataSource是「誰」呢？看一下咱們工程的配置：

故最終dataSouce.getConnection獲取的鏈接，是從SpringShardingDataSource中獲取鏈接。

com.dangdang.ddframe.rdb.sharding.jdbc.ShardingDataSource#getConnection
public ShardingConnection getConnection() throws SQLException {
        MetricsContext.init(shardingProperties);
        return new ShardingConnection(shardingContext);
}
複製代碼

返回的結果以下：

備註：這裏只是返回了一個ShardingConnection對象，該對象包含了分庫分表上下文，但此時並無執行具體的分庫操做（切換數據源）。

Connection的獲取流程清楚後，咱們繼續來看一下Statemnet對象的建立。

3.2 java.sql.Statement對象建立

stmt = prepareStatement(handler, ms.getStatementLog());            
複製代碼

上面語句的調用鏈：RoutingStatementHandler -》BaseStatementHandler

3.2.1 BaseStatementHandler#prepare

public Statement prepare(Connection connection) throws SQLException {
    ErrorContext.instance().sql(boundSql.getSql());
    Statement statement = null;
    try {
      statement = instantiateStatement(connection);    // @1
      setStatementTimeout(statement);                         // @2
      setFetchSize(statement);                                      // @3
      return statement;
    } catch (SQLException e) {
      closeStatement(statement);
      throw e;
    } catch (Exception e) {
      closeStatement(statement);
      throw new ExecutorException("Error preparing statement. Cause: " + e, e);
    }
  }
複製代碼

代碼@1：根據Connection對象（本文中是ShardingConnection)來建立Statement對象，其默認實現類：PreparedStatementHandler#instantiateStatement方法。

代碼@2：爲Statement設置超時時間。

代碼@3：設置fetchSize。

3.2.2 PreparedStatementHandler#instantiateStatement

protected Statement instantiateStatement(Connection connection) throws SQLException {
    String sql = boundSql.getSql();
    if (mappedStatement.getKeyGenerator() instanceof Jdbc3KeyGenerator) {
      String[] keyColumnNames = mappedStatement.getKeyColumns();
      if (keyColumnNames == null) {
        return connection.prepareStatement(sql, PreparedStatement.RETURN_GENERATED_KEYS);
      } else {
        return connection.prepareStatement(sql, keyColumnNames);
      }
    } else if (mappedStatement.getResultSetType() != null) {
      return connection.prepareStatement(sql, mappedStatement.getResultSetType().getValue(), ResultSet.CONCUR_READ_ONLY);
    } else {
      return connection.prepareStatement(sql);
    }
  }
複製代碼

其實Statement對象的建立，就比較簡單了，既然Connection是ShardingConnection，那就看一下其對應的prepareStatement方法便可。

3.2.2 ShardingConnection#prepareStatement

public PreparedStatement prepareStatement(final String sql) throws SQLException {   // sql，爲配置在mybatis xml文件中的sql語句
        return new ShardingPreparedStatement(this, sql);
}
ShardingPreparedStatement(final ShardingConnection shardingConnection, 
            final String sql, final int resultSetType, final int resultSetConcurrency, final int resultSetHoldability) {
        super(shardingConnection, resultSetType, resultSetConcurrency, resultSetHoldability);
        preparedSQLRouter = shardingConnection.getShardingContext().getSqlRouteEngine().prepareSQL(sql);
}
複製代碼

在構建ShardingPreparedStatement對象的時候，會根據SQL語句建立解析SQL路由的解析器對象，但此時並不會執行相關的路由計算，PreparedStatement對象建立完成後，就開始進入SQL執行流程中。

四、SQL執行流程

接下來咱們繼續看SimpleExecutor#doQuery方法的第3步，執行SQL語句：

handler.<E>query(stmt, resultHandler)。
複製代碼

首先會進入RoutingStatementHandler這個類中，進行Mybatis層面的路由（主要是根據Statement類型）

而後進入到PreparedStatementHandler#query中。

4.1 PreparedStatementHandler#query

public <E> List<E> query(Statement statement, ResultHandler resultHandler) throws SQLException {
    PreparedStatement ps = (PreparedStatement) statement;
    ps.execute();  // @1
    return resultSetHandler.<E> handleResultSets(ps);  // @2
}
複製代碼

代碼@1：調用PreparedStatement的execute方法，因爲本例是使用了Sharding-jdbc分庫分表，此時調用的具體實現爲：ShardingPreparedStatement。

代碼@2：處理結果。

咱們接下來分別來跟進execute與結果處理方法。

4.2 ShardingPreparedStatement#execute

public boolean execute() throws SQLException {
    try {
        return new PreparedStatementExecutor(getShardingConnection().getShardingContext().getExecutorEngine(), routeSQL()).execute(); // @1
    } finally {
        clearRouteContext();
    }
}
複製代碼

這裏奧妙無窮，其關鍵點以下： 1）創造PreparedStatementExecutor對象，其兩個核心參數：

ExecutorEngine executorEngine：shardingjdbc執行引擎。
Collection< PreparedStatementExecutorWrapper> preparedStatemenWrappers 一個集合，每個集合是PreparedStatement的包裝類，這個集合如何而來？

2）preparedStatemenWrappers是經過routeSQL方法產生的。

3）最終調用PreparedStatementExecutor方法的execute來執行。

接下來分別看一下routeSQL與execute方法。

4.3 ShardingPreparedStatement#routeSQL

private List<PreparedStatementExecutorWrapper> routeSQL() throws SQLException {
        List<PreparedStatementExecutorWrapper> result = new ArrayList<>();
        SQLRouteResult sqlRouteResult = preparedSQLRouter.route(getParameters());   // @1
        MergeContext mergeContext = sqlRouteResult.getMergeContext();                      
        setMergeContext(mergeContext);
        setGeneratedKeyContext(sqlRouteResult.getGeneratedKeyContext());
        for (SQLExecutionUnit each : sqlRouteResult.getExecutionUnits()) {                      // @2 
            PreparedStatement preparedStatement = (PreparedStatement) getStatement(getShardingConnection().getConnection(each.getDataSource(), sqlRouteResult.getSqlStatementType()), each.getSql());     // @3
            replayMethodsInvocation(preparedStatement);
            getParameters().replayMethodsInvocation(preparedStatement);
            result.add(wrap(preparedStatement, each));
        }
        return result;
}
複製代碼

代碼@1：根據SQL參數進行路由計算，本文暫不關注其具體實現細節，這些將在具體分析Sharding-jdbc時具體詳解，在這裏就直觀看一下其結果：

代碼@二、@3：對分庫分表的結果進行遍歷，而後使用底層Datasource來建立Connection，建立PreparedStatement 對象。

routeSQL就暫時講到這，從這裏咱們得知，會在這裏根據路由結果，使用底層的具體數據源建立對應的Connection與PreparedStatement 對象。

4.4 PreparedStatementExecutor#execute

public boolean execute() {
    Context context = MetricsContext.start("ShardingPreparedStatement-execute");
    eventPostman.postExecutionEvents();
    final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
    final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
    try {
        if (1 == preparedStatementExecutorWrappers.size()) {     // @1
            PreparedStatementExecutorWrapper preparedStatementExecutorWrapper = preparedStatementExecutorWrappers.iterator().next();
            return executeInternal(preparedStatementExecutorWrapper, isExceptionThrown, dataMap);
        }
        List<Boolean> result = executorEngine.execute(preparedStatementExecutorWrappers, new ExecuteUnit<PreparedStatementExecutorWrapper, Boolean>() {    // @2
        
            @Override
            public Boolean execute(final PreparedStatementExecutorWrapper input) throws Exception {
                synchronized (input.getPreparedStatement().getConnection()) {
                    return executeInternal(input, isExceptionThrown, dataMap);
                }
            }
        });
        return (null == result || result.isEmpty()) ? false : result.get(0);
    } finally {
        MetricsContext.stop(context);
    }
 }
複製代碼

代碼@1：若是計算出來的路由信息爲1個，則同步執行。

代碼@2：若是計算出來的路由信息有多個，則使用線程池異步執行。

那還有一個問題，經過PreparedStatement#execute方法執行後，如何返回結果呢？特別是異步執行的。

在上文其實已經談到：

4.4 DefaultResultSetHandler#handleResultSets

public List<Object> handleResultSets(Statement stmt) throws SQLException {
    ErrorContext.instance().activity("handling results").object(mappedStatement.getId());
    
    final List<Object> multipleResults = new ArrayList<Object>();

    int resultSetCount = 0;
    ResultSetWrapper rsw = getFirstResultSet(stmt);         // @1
    //省略部分代碼，完整代碼能夠查看DefaultResultSetHandler方法。
    return collapseSingleResultList(multipleResults);
  }

private ResultSetWrapper getFirstResultSet(Statement stmt) throws SQLException {
    ResultSet rs = stmt.getResultSet();              // @2
    while (rs == null) {
      // move forward to get the first resultset in case the driver
      // doesn't return the resultset as the first result (HSQLDB 2.1)
      if (stmt.getMoreResults()) {
        rs = stmt.getResultSet();
      } else {
        if (stmt.getUpdateCount() == -1) {
          // no more results. Must be no resultset
          break;
        }
      }
    }
    return rs != null ? new ResultSetWrapper(rs, configuration) : null;
  }
複製代碼

咱們看一下其關鍵代碼以下：代碼@1：調用Statement#getResultSet()方法，若是使用shardingJdbc，則會調用ShardingStatement#getResultSet()，並會處理分庫分表結果集的合併，在這裏就不詳細進行介紹，該部分會在shardingjdbc專欄詳細分析。

代碼@2：jdbc statement中獲取結果集的通用寫法，這裏也不過多的介紹。

mybatis shardingjdbc SQL執行流程就介紹到這裏了，爲了方便你們對上述流程的理解，最後給出SQL執行的流程圖：