Spring 代理的事务实现要点

文章目录

1. Spring事务的实现要点
   1. 事务的开始
   2. 事务的初始化数据库连接
   3. 实际SQL的DAO处数据库连接获取

Spring事务的实现要点

事务的开始

org.springframework.transaction.support.TransactionTemplate#execute

@Override
	public <T> T execute(TransactionCallback<T> action) throws TransactionException {
		if (this.transactionManager instanceof CallbackPreferringPlatformTransactionManager) {
			return ((CallbackPreferringPlatformTransactionManager) this.transactionManager).execute(this, action);
		}
		else {
            //关键点1：初始化事务状态对象，其中包含初始化事务的内容
			TransactionStatus status = this.transactionManager.getTransaction(this);
			T result;
			try {
				result = action.doInTransaction(status);
			}
			catch (RuntimeException ex) {
				// Transactional code threw application exception -> rollback
				rollbackOnException(status, ex);
				throw ex;
			}
			catch (Error err) {
				// Transactional code threw error -> rollback
				rollbackOnException(status, err);
				throw err;
			}
			catch (Throwable ex) {
				// Transactional code threw unexpected exception -> rollback
				rollbackOnException(status, ex);
				throw new UndeclaredThrowableException(ex, "TransactionCallback threw undeclared checked exception");
			}
			this.transactionManager.commit(status);
			return result;
		}
	}

org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction

@Override
	public final TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {
		Object transaction = doGetTransaction();

		// Cache debug flag to avoid repeated checks.
		boolean debugEnabled = logger.isDebugEnabled();

		if (definition == null) {
			// Use defaults if no transaction definition given.
			definition = new DefaultTransactionDefinition();
		}

		if (isExistingTransaction(transaction)) {
			// Existing transaction found -> check propagation behavior to find out how to behave.
			return handleExistingTransaction(definition, transaction, debugEnabled);
		}

		// Check definition settings for new transaction.
		if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
			throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
		}

		// No existing transaction found -> check propagation behavior to find out how to proceed.
		if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
			throw new IllegalTransactionStateException(
					"No existing transaction found for transaction marked with propagation 'mandatory'");
		}
		else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
				definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
				definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
			SuspendedResourcesHolder suspendedResources = suspend(null);
			if (debugEnabled) {
				logger.debug("Creating new transaction with name [" + definition.getName() + "]: " + definition);
			}
			try {
				boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
				DefaultTransactionStatus status = newTransactionStatus(
						definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
				//关键点2：此处开始一个事务
                doBegin(transaction, definition);
				prepareSynchronization(status, definition);
				return status;
			}
			catch (RuntimeException ex) {
				resume(null, suspendedResources);
				throw ex;
			}
			catch (Error err) {
				resume(null, suspendedResources);
				throw err;
			}
		}
		else {
			// Create "empty" transaction: no actual transaction, but potentially synchronization.
			if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
				logger.warn("Custom isolation level specified but no actual transaction initiated; " +
						"isolation level will effectively be ignored: " + definition);
			}
			boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
			return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null);
		}
	}

事务的初始化数据库连接

org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin

@Override
	protected void doBegin(Object transaction, TransactionDefinition definition) {
		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
		Connection con = null;

		try {
			if (!txObject.hasConnectionHolder() ||
					txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
                    //当前事务中没有创建数据库连接则新创建一个
				Connection newCon = this.dataSource.getConnection();
				if (logger.isDebugEnabled()) {
					logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
				}
				txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
			}

            //否则取已经存在的那一个
			txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
			con = txObject.getConnectionHolder().getConnection();

			Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
			txObject.setPreviousIsolationLevel(previousIsolationLevel);

            //此处手动将自动提交设置成false用于事务，JDBC标准新连接的自动提交默认是true

			// Switch to manual commit if necessary. This is very expensive in some JDBC drivers,
			// so we don't want to do it unnecessarily (for example if we've explicitly
			// configured the connection pool to set it already).
			if (con.getAutoCommit()) {
				txObject.setMustRestoreAutoCommit(true);
				if (logger.isDebugEnabled()) {
					logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
				}
				con.setAutoCommit(false);
			}

			prepareTransactionalConnection(con, definition);
			txObject.getConnectionHolder().setTransactionActive(true);

			int timeout = determineTimeout(definition);
			if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
				txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
			}

            // 将当前的数据源对象和当前事务的conn链接等信息 和当前Thread绑定
			// Bind the connection holder to the thread.
			if (txObject.isNewConnectionHolder()) {
				TransactionSynchronizationManager.bindResource(getDataSource(), txObject.getConnectionHolder());
			}
		}

		catch (Throwable ex) {
			if (txObject.isNewConnectionHolder()) {
				DataSourceUtils.releaseConnection(con, this.dataSource);
				txObject.setConnectionHolder(null, false);
			}
			throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
		}
	}

org.springframework.transaction.support.TransactionSynchronizationManager#bindResource

/**
	 * Bind the given resource for the given key to the current thread.
	 * @param key the key to bind the value to (usually the resource factory)
	 * @param value the value to bind (usually the active resource object)
	 * @throws IllegalStateException if there is already a value bound to the thread
	 * @see ResourceTransactionManager#getResourceFactory()
	 */
	public static void bindResource(Object key, Object value) throws IllegalStateException {
		Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
		Assert.notNull(value, "Value must not be null");
		Map<Object, Object> map = resources.get();
		// set ThreadLocal Map if none found
		if (map == null) {
			map = new HashMap<Object, Object>();
			resources.set(map);
		}
		Object oldValue = map.put(actualKey, value);
		// Transparently suppress a ResourceHolder that was marked as void...
		if (oldValue instanceof ResourceHolder && ((ResourceHolder) oldValue).isVoid()) {
			oldValue = null;
		}
		if (oldValue != null) {
			throw new IllegalStateException("Already value [" + oldValue + "] for key [" +
					actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
		}
		if (logger.isTraceEnabled()) {
			logger.trace("Bound value [" + value + "] for key [" + actualKey + "] to thread [" +
					Thread.currentThread().getName() + "]");
		}
	}

实际SQL的DAO处数据库连接获取

在事务的业务逻辑中，DAO的操作中最终会走到：Executor中，此时一般默认实现是：(mybatis)
org.apache.ibatis.executor.SimpleExecutor

/**
 *    Copyright 2009-2016 the original author or authors.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package org.apache.ibatis.executor;

import org.apache.ibatis.cursor.Cursor;
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;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Collections;
import java.util.List;

/**
 * @author Clinton Begin
 */
public class SimpleExecutor extends BaseExecutor {

  public SimpleExecutor(Configuration configuration, Transaction transaction) {
    super(configuration, transaction);
  }

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

  @Override
  protected <E> Cursor<E> doQueryCursor(MappedStatement ms, Object parameter, RowBounds rowBounds, BoundSql boundSql) throws SQLException {
    Configuration configuration = ms.getConfiguration();
    StatementHandler handler = configuration.newStatementHandler(wrapper, ms, parameter, rowBounds, null, boundSql);
    Statement stmt = prepareStatement(handler, ms.getStatementLog());
    return handler.<E>queryCursor(stmt);
  }

  @Override
  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, transaction.getTimeout());
    handler.parameterize(stmt);
    return stmt;
  }

}

在org.apache.ibatis.executor.SimpleExecutor#prepareStatement方法中会去拿当前的数据库连接
org.apache.ibatis.executor.BaseExecutor#getConnection

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

使用Spring框架管理的情况下默认transaction的实现是：
org.mybatis.spring.transaction.SpringManagedTransaction，其getConn的方法如下：

/**
  * {@inheritDoc}
  */
 @Override
 public Connection getConnection() throws SQLException {
   if (this.connection == null) {
     openConnection();
   }
   return this.connection;
 }

 /**
  * Gets a connection from Spring transaction manager and discovers if this
  * {@code Transaction} should manage connection or let it to Spring.
  * <p>
  * It also reads autocommit setting because when using Spring Transaction MyBatis
  * thinks that autocommit is always false and will always call commit/rollback
  * so we need to no-op that calls.
  */
 private void openConnection() throws SQLException {
   this.connection = DataSourceUtils.getConnection(this.dataSource);
   this.autoCommit = this.connection.getAutoCommit();
   this.isConnectionTransactional = DataSourceUtils.isConnectionTransactional(this.connection, this.dataSource);

   if (LOGGER.isDebugEnabled()) {
     LOGGER.debug(
         "JDBC Connection ["
             + this.connection
             + "] will"
             + (this.isConnectionTransactional ? " " : " not ")
             + "be managed by Spring");
   }
 }

最终导向了

• org.springframework.jdbc.datasource.DataSourceUtils#doGetConnection：

  /**
     ...
   */
  public static Connection getConnection(DataSource dataSource) throws CannotGetJdbcConnectionException {
  	try {
  		return doGetConnection(dataSource);
  	}
  	catch (SQLException ex) {
  		throw new CannotGetJdbcConnectionException("Could not get JDBC Connection", ex);
  	}
  }
  
  /**
   * Actually obtain a JDBC Connection from the given DataSource.
   * Same as {@link #getConnection}, but throwing the original SQLException.
   * <p>Is aware of a corresponding Connection bound to the current thread, for example
   * when using {@link DataSourceTransactionManager}. Will bind a Connection to the thread
   * if transaction synchronization is active (e.g. if in a JTA transaction).
   * <p>Directly accessed by {@link TransactionAwareDataSourceProxy}.
   * @param dataSource the DataSource to obtain Connections from
   * @return a JDBC Connection from the given DataSource
   * @throws SQLException if thrown by JDBC methods
   * @see #doReleaseConnection
   */
  public static Connection doGetConnection(DataSource dataSource) throws SQLException {
  	Assert.notNull(dataSource, "No DataSource specified");
  
         //注意此时会首先去TransactionSynchronizationManager中拿当前线程中是否已经有针对这个datasource的事务信息，如果有
         //则会返回其中的conn，而这个conn是在org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin中创建初始化的
         //所以，针对org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource实现的业务动态数据源，在事务处理时需要在事务模板的execute方法前进行
         //determineCurrentLookupKey，而不是在dao方法之前进行，因为此时如果dao在事务之中，在executor中执行sql时候拿到的数据源仍然是
         //org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin中获取的DB链接，而这两个的时刻的db路由key可能不一致导致错误
  	ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource);
  	if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
  		conHolder.requested();
  		if (!conHolder.hasConnection()) {
  			logger.debug("Fetching resumed JDBC Connection from DataSource");
  			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();
  
  	if (TransactionSynchronizationManager.isSynchronizationActive()) {
  		logger.debug("Registering transaction synchronization for JDBC Connection");
  		// Use same Connection for further JDBC actions within the transaction.
  		// Thread-bound object will get removed by synchronization at transaction completion.
  		ConnectionHolder holderToUse = conHolder;
  		if (holderToUse == null) {
  			holderToUse = new ConnectionHolder(con);
  		}
  		else {
  			holderToUse.setConnection(con);
  		}
  		holderToUse.requested();
  		TransactionSynchronizationManager.registerSynchronization(
  				new ConnectionSynchronization(holderToUse, dataSource));
  		holderToUse.setSynchronizedWithTransaction(true);
  		if (holderToUse != conHolder) {
  			TransactionSynchronizationManager.bindResource(dataSource, holderToUse);
  		}
  	}
  
  	return con;
  }