Spring 动态数据源事务处理

在一般的 Spring 应用中，如果底层数据库访问采用的是 MyBatis，那么在大多数情况下，只使用一个单独的数据源，Spring 的事务管理在大多数情况下都是有效的。然而，在一些复杂的业务场景下，如需要在某一时刻访问不同的数据库，由于 Spring 对于事务管理实现的方式，可能不能达到预期的效果。本文将简要介绍 Spring 中事务的实现方式，并对以 MyBatis 为底层数据库访问的系统为例，提供多数据源事务处理的解决方案

Spring 事务的实现原理

常见地，在 Spring 中添加事务的方式通常都是在对应的方法或类上加上 @Transactional 注解显式地将这部分处理加上事务，对于 @Transactional 注解，Spring 会在 org.springframework.transaction.annotation.AnnotationTransactionAttributeSource 定义方法拦截的匹配规则（即 AOP 部分中的 PointCut），而具体的处理逻辑（即 AOP 中的 Advice）则是在 org.springframework.transaction.interceptor.TransactionInterceptor 中定义

具体事务执行的调用链路如下

Spring 对于事务切面采取的具体行为实现如下：

public class TransactionInterceptor 
    extends TransactionAspectSupport 
    implements MethodInterceptor, Serializable {
    
    // 这里的方法定义为 MethodInterceptor，即 AOP 实际调用点
    @Override
	@Nullable
	public Object invoke(MethodInvocation invocation) throws Throwable {
		// Work out the target class: may be {@code null}.
		// The TransactionAttributeSource should be passed the target class
		// as well as the method, which may be from an interface.
		Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);

		// Adapt to TransactionAspectSupport's invokeWithinTransaction...
        // invokeWithinTransaction 为父类 TransactionAspectSupport 定义的方法
		return invokeWithinTransaction(invocation.getMethod(), targetClass, new CoroutinesInvocationCallback() {
			@Override
			@Nullable
			public Object proceedWithInvocation() throws Throwable {
				return invocation.proceed();
			}
			@Override
			public Object getTarget() {
				return invocation.getThis();
			}
			@Override
			public Object[] getArguments() {
				return invocation.getArguments();
			}
		});
	}
}

继续进入 TransactionAspectSupport 的 invokeWithinTransaction 方法：

public abstract class TransactionAspectSupport 
    implements BeanFactoryAware, InitializingBean {
    protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
			final InvocationCallback invocation) throws Throwable {
        // 省略响应式事务和编程式事务的处理逻辑

        // 当前事务管理的实际
		PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
		final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);

		if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
			// Standard transaction demarcation with getTransaction and commit/rollback calls.
            /*
            	检查在当前的执行上下文中，是否需要创建新的事务，这是因为当前执行的业务处理可能在上一个已经开始
            	的事务处理中
            */
			TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);

			Object retVal;
			try {
				// This is an around advice: Invoke the next interceptor in the chain.
				// This will normally result in a target object being invoked.
				retVal = invocation.proceedWithInvocation(); // 实际业务代码的业务处理
			}
			catch (Throwable ex) {
				// target invocation exception
				completeTransactionAfterThrowing(txInfo, ex); // 出现异常的回滚处理
				throw ex;
			}
			finally {
				cleanupTransactionInfo(txInfo);
			}

			if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
				// Set rollback-only in case of Vavr failure matching our rollback rules...
				TransactionStatus status = txInfo.getTransactionStatus();
				if (status != null && txAttr != null) {
					retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
				}
			}

            // 如果没有出现异常，则提交本次事务
			commitTransactionAfterReturning(txInfo);
			return retVal;
		}
	}
}

在获取事务信息对象时，首先需要获取到对应的事务状态对象 TransactionStatus，这个状态对象决定了 Spring 后续要对当前事务采取的何种行为，具体代码在 org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction

// 这里的 definition 是通过解析 @Transactional 注解中的属性得到的配置对象
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
    throws TransactionException {

    // Use defaults if no transaction definition given.
    TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());

    /*
    	这里获取事务相关的对象（如持有的数据库连接等），具体由子类来定义相关的实现
    */
    Object transaction = doGetTransaction();
    boolean debugEnabled = logger.isDebugEnabled();

    // 如果当前已经在一个事务中，那么需要按照定义的属性采取对应的行为
    if (isExistingTransaction(transaction)) {
        // Existing transaction found -> check propagation behavior to find out how to behave.
        return handleExistingTransaction(def, transaction, debugEnabled);
    }

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

    // No existing transaction found -> check propagation behavior to find out how to proceed.
    if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
        throw new IllegalTransactionStateException(
            "No existing transaction found for transaction marked with propagation 'mandatory'");
    }
    // 需要重新开启一个新的事务的情况，具体在 org.springframework.transaction.TransactionDefinition 有相关的定义
    else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
             def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
             def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
        SuspendedResourcesHolder suspendedResources = suspend(null);
        if (debugEnabled) {
            logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
        }
        try {
            // 开启一个新的事务
            return startTransaction(def, transaction, debugEnabled, suspendedResources);
        }
        catch (RuntimeException | Error ex) {
            resume(null, suspendedResources);
            throw ex;
        }
    }
    else {
        // Create "empty" transaction: no actual transaction, but potentially synchronization.
        if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
            logger.warn("Custom isolation level specified but no actual transaction initiated; " +
                        "isolation level will effectively be ignored: " + def);
        }
        boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
        return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
    }
}

在 AbstractPlatformTransactionManager 中已经定义了事务处理的大体框架，而实际的事务实现则交由具体的子类实现，在一般情况下，由 org.springframework.jdbc.datasource.DataSourceTransactionManager 采取具体的实现

主要关注的点在于对于事务信息对象的创建，事务的开启、提交回滚操作，具体对应的代码如下：

事务信息对象的创建代码：

protected Object doGetTransaction() {
    /*
    	简单地理解，DataSourceTransactionObject 就是一个持有数据库连接的资源对象
    */
    DataSourceTransactionObject txObject = new DataSourceTransactionObject();
    txObject.setSavepointAllowed(isNestedTransactionAllowed());
    /*
    	TransactionSynchronizationManager 是用于管理在事务执行过程相关的信息对象的一个工具类，基本上
    	这个类持有的事务信息贯穿了整个 Spring 事务管理
    */
    ConnectionHolder conHolder =
        (ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
    txObject.setConnectionHolder(conHolder, false);
    return txObject;
}

开启事务对应的源代码：

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

    try {
        /*
        	如果当前事务对象没有持有数据库连接，则需要从对应的 DataSource 中获取对应的连接
        */
        if (!txObject.hasConnectionHolder() ||
            txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
            Connection newCon = obtainDataSource().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);
        txObject.setReadOnly(definition.isReadOnly());

        // 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).
        
        /*
        	由于当前的事务已经交由 Spring 进行管理，那么在这种情况下，原有数据库连接的自动提交
        	必须是关闭的，因为如果开启了自动提交，那么实际上就相当于每一次的 SQL 都会执行一次事务的提交，
        	这种情况下事务的管理没有意义
        */
        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);
        }

        // Bind the connection holder to the thread.
        
        /*
        	如果是新创建的事务，那么需要绑定这个数据库连接对象到这个事务中，使得后续再进来的业务处理
        	能够顺利地进入原有的事务中
        */
        if (txObject.isNewConnectionHolder()) {
            TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
        }
    }

    catch (Throwable ex) {
        if (txObject.isNewConnectionHolder()) {
            DataSourceUtils.releaseConnection(con, obtainDataSource());
            txObject.setConnectionHolder(null, false);
        }
        throw new CannotCreateTransactionException("Could not open