Spring原理篇(17)--Spring事务的传播机制；该篇章是Spring原理篇的最后一章；

Posted 2021-12-08 喜欢编码的老胡

@TOC# Spring系列
记录在程序走的每一步___auth:huf

---

由浅入深:
我们再使用@Transctional注解的时候;我们非常喜欢一个动作就是调度方法 直接就xxxx(); 就调度了Bean内的方法; 这种方式 是不起效果的 请注意

如果要事务起效 请用他的代理方法进行调度 那么就会起效; 这个跟事务传播没有关系; 是跟AOP 是否使用了代理对象有关系;

---

我们前一篇文章 有大量源码. 可能看起来很枯燥. 昨天文章有大量的文字描述原理 也想睡觉; 今天我们图文并茂.先来一段文字: 我们执行对象方法的时候 对象方法如何去开启事务;

1: Spring 事务管理器 创建数据库连接;(跟上章呼应上)
2: conn.setSqlLevel 设置隔离级别;
3: conn.setAuthCommit=false; 关闭自动提交
4: conn.setThreadlocal 它是一个Map 实际上就是把数据库连接 conn放进去;
5: target.sql 对象方法执行;

	如果这里由调用了一个方法 并且方法上有 @Transactional(propagation = Propagation.REQUIRES_NEW)
	关键的来了;
	1)挂起上面一个事务;
	2) Spring 事务管理器 创建数据库连接;
	3) conn.setSqlLevel;设置隔离级别;
	4) conn.setAuthCommit = false;关闭自动提交
	5) conn.setThreadlocal 它是一个Map 实际上就是把数据库连接 conn放进去;
	target.sql
	....又遇到一个事务...又在执行一遍; 
	conn.commit; 或者是 rollback 
	恢复 挂起的事务 ---->把挂起事务放进去Threadlocal

N:conn.commit; 或者是 rollback

原理部分就讲清楚 讲明白了; 下面上源码部分;

---

因为我们使用了@Trancastional 注解 所以它最终会进来 TransactionInterceptor 也就是上一篇文章 说的Advice 中 执行 invoke方法;

中间的proceed 就是去执行我们的被代理类的代理方法; 在这过程当中; 我们一步一步分析:在Invoke方法中 return invokeWithinTransaction 该方法点进去 我们一路看下去

@Nullable
	protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
			final InvocationCallback invocation) throws Throwable {
		"如果事务属性为null，则该方法是非事务性的。"
		TransactionAttributeSource tas = getTransactionAttributeSource();
		"获取propagation 等配置属性 "
		final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
		"获取事务管理器,事务管理器 就是去创建连接  Commit rollback 等一系列操作用的"
		final TransactionManager tm = determineTransactionManager(txAttr);
		"是否是 ReactiveTransactionManager 管理器;目前笔者没有用过这个管理器; "
		"我们在config配置的 是PlatformTransactionManager"
		if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {
			"该段代码忽略"
			"该段代码最后肯定是return的"
			return result;
		}
		"事务管理器 : 可以在Config里面进行配置; 只需要传入dataSource() 就可以"
		PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
		final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);

		if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
			"该方法内部有一个开启了事务的核心方法, 下一自然段我们重点看这个方法"
			TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);

			Object retVal;
			try {
				"执行被代理对象的代理方法;"
				retVal = invocation.proceedWithInvocation();
			}
			catch (Throwable ex) {
				"如果报错 进入该方法进行回滚 "
				completeTransactionAfterThrowing(txInfo, ex);
				"抛出异常"
				throw ex;
			}
			finally {
				cleanupTransactionInfo(txInfo);
			}
												"设置回滚规则,自行配置规则,这里不会诉说"
			if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
				"仅当Vavr故障与回滚规则匹配时，才进行et回滚"
				TransactionStatus status = txInfo.getTransactionStatus();
				if (status != null && txAttr != null) {
					retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
				}
			}
			"提交事务"
			commitTransactionAfterReturning(txInfo);
			return retVal;
		}
		"这下面还会return 一种情况"
	}

我们先来看 事务传播机制的规则:

Propagation.MANDATORY   "强制性"
Propagation.NESTED "嵌套"
Propagation.NEVER "从不"
Propagation.NOT_SUPPORTED "不支持"
Propagation.REQUIRED "默认事务"
Propagation.REQUIRES_NEW "从新再开启一个事务" 
Propagation.SUPPORTS "支持" 

在说核心方法之前 我们先知道 在传播机制规则中 我们可以配置这7种规则; 这七种规则 我并没有详细说明或者是直接说明它干什么的. 跟着源码 就知道了

createTransactionIfNecessary()方法

"再其方法内部  这里就是开启事务的核心方法了"
status = tm.getTransaction(txAttr);

getTransaction()方法;

	public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
			throws TransactionException {
		"如果没有给出事务定义，则使用默认值。"
		TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
		Object transaction = doGetTransaction();
		boolean debugEnabled = logger.isDebugEnabled();
		if (isExistingTransaction(transaction)) {
			"判断传播机制 这个方法就解释了我们 的传播机制规则; 下面会详细说"
			return handleExistingTransaction(def, transaction, debugEnabled);
		}
		if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
			throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
		}
		"注意   Propagation.MANDATORY  强制性  就会抛出异常. 它的作用是 手动的开启事务 "
		"必须要有事务.如果没有事务 就会抛出异常 . MANDATORY得到了解释 随后总结会进行述说"
		if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
			throw new IllegalTransactionStateException(
					"No existing transaction found for transaction marked with propagation 'mandatory'");
		}
		"如果 PROPAGATION_REQUIRED PROPAGATION_REQUIRES_NEW   TransactionDefinition.PROPAGATION_NESTED "
		"继续往下走."
		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;
			}
		}
		....省略部分代码
		}
	}

startTransaction

	private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
			boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {

		boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
		DefaultTransactionStatus status = newTransactionStatus(
				definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
		"开启事务 看用了什么组件;"
		doBegin(transaction, definition);
		prepareSynchronization(status, definition);
		return status;
	} 

这里 我们用了DataSource

@Override
	protected void doBegin(Object transaction, TransactionDefinition definition) {
		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
		"申明数据库连接"
		Connection con = null; 
		try {
			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());
		
			if (con.getAutoCommit()) {
				txObject.setMustRestoreAutoCommit(true);
				if (logger.isDebugEnabled()) {
					logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
				}
				"关闭事务自动提交!"
				con.setAutoCommit(false);
			}
		```省略部分代码
	}

上面那一段源码 匹配上这段文字了

主要是设置autoCommit 为false;

接下来 我们看这个方法
再我们的

private TransactionStatus handleExistingTransaction(
			TransactionDefinition definition, Object transaction, boolean debugEnabled)
			throws TransactionException {
		"如果是never 就报错 不支持当前有事务 后面会写进去总结里面"
		if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
			throw new IllegalTransactionStateException(
					"Existing transaction found for transaction marked with propagation 'never'");
		}
		"NOT_SUPPORTED 意思是不开启新的数据库连接. "
		"我们的事务 实际就是一个一个的数据库连接;我们有两个步骤 一个是先挂起 然后开启新的事务 "
		" 这个就是说  我挂起. 但是不开启新的事务;"
		  "那后面的那个数据库连接 实际上用的就是Jdbc 或者是 Mybatis里面的数据库连接了"
		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);
		}
		"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 {
				return startTransaction(definition, transaction, debugEnabled, suspendedResources);
			}
			catch (RuntimeException | Error beginEx) {
				resumeAfterBeginException(transaction, suspendedResources, beginEx);
				throw beginEx;
			}
		}
		"PROPAGATION_NESTED 不会去开启新的数据库连接 再数据库里 有一个savepoint的一个说法"
		"也就是设置了这个savepoint 我们可以回滚到 某一个点上 例如 我执行A B C 3个sql"
		"再B sql 上面我们创建了 savepoint 那么 就会回滚到B sql 这里来;"
		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);
		return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
	}

---

以上 我们就讲解了 事务的传播机制;我们挂起 以及恢复挂起 提交 以及回滚的源码; 并没有详细说道. 这里如果感兴趣的同学 可以去看一下源码;

总结

1:

Propagation.MANDATORY   "强制性"  "强制性要求开启事务 用于手动开启事务"
Propagation.NESTED "嵌套" "回滚到某一个savepoint 的那个点上 不会从新开启数据库连接 "
Propagation.NEVER "从不" "不允许有事务的存在" 
Propagation.NOT_SUPPORTED "不支持" "挂起当前事务,不创建新的事务"
Propagation.REQUIRED "默认事务"
Propagation.REQUIRES_NEW "从新再开启一个事务,与之前事务毫不相关" 
Propagation.SUPPORTS "支持,如果上一个Bean  或者方法 使用了事务 那么就延用事务. 如果没用事务 那么就不用事务" 

2:
Spring开启事务的全部流程

Spring 核心源码 在这里 就结束了;

see you