回首Spring源码之声明式事务源码级详解（下篇）

前言

在上篇中，详细分析了事务的执行过程，中间留了一些尾巴没讲：在一个内部事务中spring是如何处理外部事务的？事务同步管理器TransactionSynchronizationManager到底是何方神圣，它有哪些作用？

handleExistingTransaction

当外层有了一个事务，内层需要根据传播行为决定应该怎样处理，下面进入源码：

//判断是否是存在的事务，第二次进来事务对象上的connectionHolder有值了，表明存在事务
if (isExistingTransaction(transaction)) {
		// Existing transaction found -> check propagation behavior to find out how to behave.
		//如果是存在的事务，则处理
		return handleExistingTransaction(def, transaction, debugEnabled);
}


private TransactionStatus handleExistingTransaction(
			TransactionDefinition definition, Object transaction, boolean debugEnabled)
			throws TransactionException {
		//PROPAGATION_NEVER，抛异常
		if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
			throw new IllegalTransactionStateException(
					"Existing transaction found for transaction marked with propagation 'never'");
		}
		//PROPAGATION_NOT_SUPPORTED不支持事务，那么挂起事务，内层页不开启事务
		if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
			if (debugEnabled) {
				logger.debug("Suspending current transaction");
			}
			Object suspendedResources = suspend(transaction);
			boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
			return prepareTransactionStatus(
					definition, null, false, newSynchronization, debugEnabled, suspendedResources);
		}
		//PROPAGATION_REQUIRES_NEW开起一个新事务
		if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
			if (debugEnabled) {
				logger.debug("Suspending current transaction, creating new transaction with name [" +
						definition.getName() + "]");
			}
			//挂起外层事务
			SuspendedResourcesHolder suspendedResources = suspend(transaction);
			try {
				//REQUIRES_NEW需要开启一个新事务
				return startTransaction(definition, transaction, debugEnabled, suspendedResources);
			}
			catch (RuntimeException | Error beginEx) {
				resumeAfterBeginException(transaction, suspendedResources, beginEx);
				throw beginEx;
			}
		}
		//PROPAGATION_NESTED嵌套事务
		if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
			if (!isNestedTransactionAllowed()) {
				throw new NestedTransactionNotSupportedException(
						"Transaction manager does not allow nested transactions by default - " +
						"specify 'nestedTransactionAllowed' property with value 'true'");
			}
			if (debugEnabled) {
				logger.debug("Creating nested transaction with name [" + definition.getName() + "]");
			}
			if (useSavepointForNestedTransaction()) {
				// Create savepoint within existing Spring-managed transaction,
				// through the SavepointManager API implemented by TransactionStatus.
				// Usually uses JDBC 3.0 savepoints. Never activates Spring synchronization.
				DefaultTransactionStatus status =
						prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
				//设置保存点，嵌套事务不需要开启新的事务
				status.createAndHoldSavepoint();
				return status;
			}
			else {
				// Nested transaction through nested begin and commit/rollback calls.
				// Usually only for JTA: Spring synchronization might get activated here
				// in case of a pre-existing JTA transaction.
				return startTransaction(definition, transaction, debugEnabled, null);
			}
		}

		// Assumably PROPAGATION_SUPPORTS or PROPAGATION_REQUIRED.
		if (debugEnabled) {
			logger.debug("Participating in existing transaction");
		}
		if (isValidateExistingTransaction()) {
			if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
				Integer currentIsolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
				if (currentIsolationLevel == null || currentIsolationLevel != definition.getIsolationLevel()) {
					Constants isoConstants = DefaultTransactionDefinition.constants;
					throw new IllegalTransactionStateException("Participating transaction with definition [" +
							definition + "] specifies isolation level which is incompatible with existing transaction: " +
							(currentIsolationLevel != null ?
									isoConstants.toCode(currentIsolationLevel, DefaultTransactionDefinition.PREFIX_ISOLATION) :
									"(unknown)"));
				}
			}
			if (!definition.isReadOnly()) {
				if (TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
					throw new IllegalTransactionStateException("Participating transaction with definition [" +
							definition + "] is not marked as read-only but existing transaction is");
				}
			}
		}
		boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
		//如果传播行为是PROPAGATION_SUPPORTS or PROPAGATION_REQUIRED，不开启新事物，直接返回外层事务状态
		return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
	}

看一下如何挂起一个事务：

protected final SuspendedResourcesHolder suspend(@Nullable Object transaction) throws TransactionException {
		if (TransactionSynchronizationManager.isSynchronizationActive()) {
			//挂起同步对象
			List suspendedSynchronizations = doSuspendSynchronization();
			try {
				Object suspendedResources = null;
				if (transaction != null) {
					//挂起事务，事务对象释放connectionHolder
					suspendedResources = doSuspend(transaction);
				}
				//获取已存在的事物的名称
				String name = TransactionSynchronizationManager.getCurrentTransactionName();
				//清空线程变量的
				TransactionSynchronizationManager.setCurrentTransactionName(null);
				//获取是否只读事务
				boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();
				TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);
				//获取已存在事务的隔离级别
				Integer isolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
				TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(null);
				//获取激活标志
				boolean wasActive = TransactionSynchronizationManager.isActualTransactionActive();
				//清空标记Actual=外层事务
				TransactionSynchronizationManager.setActualTransactionActive(false);
				//把上诉从线程变量中获取出来的外层事务的同步对象和事务属性封装为挂起的事务资源持有者返回出去
				return new SuspendedResourcesHolder(
						suspendedResources, suspendedSynchronizations, name, readOnly, isolationLevel, wasActive);
			}
			catch (RuntimeException | Error ex) {
				// doSuspend failed - original transaction is still active...
				doResumeSynchronization(suspendedSynchronizations);
				throw ex;
			}
		}
		else if (transaction != null) {
			// Transaction active but no synchronization active.
			Object suspendedResources = doSuspend(transaction);
			return new SuspendedResourcesHolder(suspendedResources);
		}
		else {
			// Neither transaction nor synchronization active.
			return null;
		}
	}


protected Object doSuspend(Object transaction) {
		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
		txObject.setConnectionHolder(null);
		//解绑TransactionSynchronizationManager中的本地线程变量resources，并返回已和该数据源映射了的connectionHolder
		//也就是外层事务的connectionHolder，会被保存到挂起的事务信息中
		return TransactionSynchronizationManager.unbindResource(obtainDataSource());
	}

ok，以上就是挂起一个事务的全部操作了。

TransactionSynchronizationManager

TransactionSynchronizationManager是事务同步管理器，用来为每个线程管理事务资源和事务同步对象TransactionSynchronization，先看下TransactionSynchronizationManager的几ThreadLocal属性：

    //存储key为数据源，value为ConnectionHolder的本地线程map
	private static final ThreadLocal> resources =
			new NamedThreadLocal<>("Transactional resources");
	//事务同步对象，每个事务同步对象管理者整个事务的生命周期，可在相应事务执行阶段执行相应的方法，可用作用户扩展
	private static final ThreadLocal> synchronizations =
			new NamedThreadLocal<>("Transaction synchronizations");
	//当前事务名称
	private static final ThreadLocal currentTransactionName =
			new NamedThreadLocal<>("Current transaction name");
	//当前事务是否只读
	private static final ThreadLocal currentTransactionReadOnly =
			new NamedThreadLocal<>("Current transaction read-only status");
	//当前事务隔离级别
	private static final ThreadLocal currentTransactionIsolationLevel =
			new NamedThreadLocal<>("Current transaction isolation level");
	//是否活跃事务
	private static final ThreadLocal actualTransactionActive =
			new NamedThreadLocal<>("Actual transaction active");

resources这个map中存放着key为数据源，value为ConnectionHolder的本地线程变量，在后面真正执行sql的时候，spring会从这个map中获取数据库连接Connection，从而使用同一个Connection，就和声明式事务中的事务是同一个了。

当需要开启新的事务时，TransactionSynchronizationManager中的线程变量会被额外保存下来，在回到外层事务时，重新设置到TransactionSynchronizationManager中。

在spring中这里有一个扩展点，就是TransactionSynchronization事务同步对象，我们可以自行向TransactionSynchronizationManager注册TransactionSynchronization，在事务执行的不同阶段调用我们的自定义方法：

注册TransactionSynchronization：

public static void registerSynchronization(TransactionSynchronization synchronization)
			throws IllegalStateException {

		Assert.notNull(synchronization, "TransactionSynchronization must not be null");
		Set synchs = synchronizations.get();
		if (synchs == null) {
			throw new IllegalStateException("Transaction synchronization is not active");
		}
		synchs.add(synchronization);
}

主要有4个可扩展的回调方法：

    //事务提交前调用
    default void beforeCommit(boolean readOnly) {
	}

	
	default void beforeCompletion() {
	}

	
	default void afterCommit() {
	}

	
	default void afterCompletion(int status) {
	}

这些方法会在processCommit手动提交事务这个方法中被调用。例如在数据库commit后执行一些操作，比如发送消息等，如果把发送消息和数据库操作放在一个事务里，当发送消息网络延迟，会导致本地数据库事务占用资源，这样的场景就可以注册一个TransactionSynchronization，在提交后执行发送消息的操作了。

总结

spring声明式事务源码到此结束！

回首Spring源码之声明式事务源码级详解（下篇）

Java相关栏目本月热门文章