Spring事务源码解析（三）

小鱼儿 2024-04-18 20:39 90阅读 0赞

在之前的文章[Spring事务源码解析（二）获取增强][Spring]中，我们分析了Spring事务的实现是基于AOP实现的，还分析了增强`BeanFactoryTransactionAttributeSourceAdvisor`中的相关内容。而这个增强中包含一个拦截器`TransactionInterceptor`，代理的实现就是基于这个拦截器  
现在来看一下这个拦截器的代码

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 </pre></td> 
 <td><pre>	public Object invoke(final MethodInvocation invocation) throws Throwable {
 		Class&lt;?&gt; targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null); 		// 往下看 		return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed); 	} protected Object invokeWithinTransaction(Method method, @Nullable Class&lt;?&gt; targetClass, 			final InvocationCallback invocation) throws Throwable {
 		 		TransactionAttributeSource tas = getTransactionAttributeSource(); 		 // 获取事务属性 		final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null); 		final PlatformTransactionManager tm = determineTransactionManager(txAttr); 		//方法标识，例如usersService.testTransaction 		final String joinpointIdentification = methodIdentification(method, targetClass, txAttr); 		//声明式事务 		if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
 			// 1.创建TransactionInfo 			TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification); 			Object retVal = null; 			try {
 				// 2. 调用执行链 				retVal = invocation.proceedWithInvocation(); 			} 			catch (Throwable ex) {
 				// 3. 异常回滚 				completeTransactionAfterThrowing(txInfo, ex); 				throw ex; 			} 			finally {
 			 	//4. 清除事务信息 				cleanupTransactionInfo(txInfo); 			} 		 	//5 提交事务 			commitTransactionAfterReturning(txInfo); 			return retVal; 		} 		else {
 		 //编程式事务处理 			final ThrowableHolder throwableHolder = new ThrowableHolder(); 			try {
 				Object result = ((CallbackPreferringPlatformTransactionManager) tm).execute(txAttr, status -&gt; {
 					TransactionInfo txInfo = prepareTransactionInfo(tm, txAttr, joinpointIdentification, status); 					try {
 						return invocation.proceedWithInvocation(); 					} 					catch (Throwable ex) {
 						if (txAttr.rollbackOn(ex)) {
 							if (ex instanceof RuntimeException) {
 								throw (RuntimeException) ex; 							} 							else {
 								throw new ThrowableHolderException(ex); 							} 						} 						else {
 							throwableHolder.throwable = ex; 							return null; 						} 					} 					finally {
 						cleanupTransactionInfo(txInfo); 					} 				}); 				if (throwableHolder.throwable != null) {
 					throw throwableHolder.throwable; 				} 				return result; 			} 			catch (ThrowableHolderException ex) {
 				throw ex.getCause(); 			} 			catch (TransactionSystemException ex2) {
 				if (throwableHolder.throwable != null) {
 					logger.error("Application exception overridden by commit exception", throwableHolder.throwable); 					ex2.initApplicationException(throwableHolder.throwable); 				} 				throw ex2; 			} 			catch (Throwable ex2) {
 				if (throwableHolder.throwable != null) {
 					logger.error("Application exception overridden by commit exception", throwableHolder.throwable); 				} 				throw ex2; 			} 		} 	} </pre></td> 
 </tr> 
 </tbody> 
</table>

由于咱们经常使用的就是声明式事务，所以接下来的解析也是基于声明式事务来的

##### 创建事务 #####

###### 1. 获取事务 ######

整个获取事务的过程还是包含多个步骤的，其中doBegin比较值得聊一聊

至于记录事务状态这里先不说，先看一下已存在的事务是如何处理的

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 </pre></td> 
 <td><pre>private TransactionStatus handleExistingTransaction( 			TransactionDefinition definition, Object transaction, boolean debugEnabled) 			throws TransactionException {
 		//判断传播行为是否需要存在事务 		if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
 			throw new IllegalTransactionStateException( 					"Existing transaction found for transaction marked with propagation 'never'"); 		} 		//不开启事务 		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); 		} 		//总是新建事务 		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 {
 				boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER); 				DefaultTransactionStatus status = newTransactionStatus( 						definition, transaction, true, newSynchronization, debugEnabled, suspendedResources); 				doBegin(transaction, definition); 				prepareSynchronization(status, definition); 				return status; 			} 			catch (RuntimeException | Error beginEx) {
 				resumeAfterBeginException(transaction, suspendedResources, beginEx); 				throw beginEx; 			} 		} 		//嵌套事务 		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. 				boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER); 				DefaultTransactionStatus status = newTransactionStatus( 						definition, transaction, true, newSynchronization, debugEnabled, null); 				doBegin(transaction, definition); 				prepareSynchronization(status, definition); 				return status; 			} 		} 		// 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); } </pre></td> 
 </tr> 
 </tbody> 
</table>

可以看到记录事务状态是最后一步了

###### 构建事务信息 ######

##### 2. 事务回滚 #####

可以知道事务回滚是进入这个方法

看看回滚的逻辑

而真的的回滚逻辑则是利用JDBC的回滚逻辑实现的

##### 3. 事务提交 #####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 </pre></td> 
 <td><pre>protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) {
 		if (txInfo != null &amp;&amp; txInfo.getTransactionStatus() != null) {
 			if (logger.isTraceEnabled()) {
 				logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "]"); 			} 			txInfo.getTransactionManager().commit(txInfo.getTransactionStatus()); 		} 	} public final void commit(TransactionStatus status) throws TransactionException {
 //如果事务已经完成 		if (status.isCompleted()) {
 			throw new IllegalTransactionStateException( 					"Transaction is already completed - do not call commit or rollback more than once per transaction"); 		} 		DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status; //如果事务设置了回滚标记则直接回滚 		if (defStatus.isLocalRollbackOnly()) {
 			if (defStatus.isDebug()) {
 				logger.debug("Transactional code has requested rollback"); 			} 			processRollback(defStatus, false); 			return; 		} 		if (!shouldCommitOnGlobalRollbackOnly() &amp;&amp; defStatus.isGlobalRollbackOnly()) {
 			if (defStatus.isDebug()) {
 				logger.debug("Global transaction is marked as rollback-only but transactional code requested commit"); 			} 			processRollback(defStatus, true); 			return; 		} //提交事务 		processCommit(defStatus); 	} private void processCommit(DefaultTransactionStatus status) throws TransactionException {
 		try {
 			boolean beforeCompletionInvoked = false; 			try {
 				boolean unexpectedRollback = false; //扩展点的相关调用 				prepareForCommit(status); 				triggerBeforeCommit(status); 				triggerBeforeCompletion(status); 				beforeCompletionInvoked = true; 				if (status.hasSavepoint()) {
 					if (status.isDebug()) {
 						logger.debug("Releasing transaction savepoint"); 					} 					unexpectedRollback = status.isGlobalRollbackOnly(); //清除保存点信息 					status.releaseHeldSavepoint(); 				} 				else if (status.isNewTransaction()) {
 					if (status.isDebug()) {
 						logger.debug("Initiating transaction commit"); 					} 					unexpectedRollback = status.isGlobalRollbackOnly(); //提交事务，同样这里也是调用的JDBC的方法实现的 					doCommit(status); 				} 				else if (isFailEarlyOnGlobalRollbackOnly()) {
 					unexpectedRollback = status.isGlobalRollbackOnly(); 				} 				// Throw UnexpectedRollbackException if we have a global rollback-only 				// marker but still didn't get a corresponding exception from commit. 				if (unexpectedRollback) {
 					throw new UnexpectedRollbackException( 							"Transaction silently rolled back because it has been marked as rollback-only"); 				} 			} 			catch (UnexpectedRollbackException ex) {
 				// can only be caused by doCommit 				triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK); 				throw ex; 			} 			catch (TransactionException ex) {
 				// can only be caused by doCommit 				if (isRollbackOnCommitFailure()) {
 					doRollbackOnCommitException(status, ex); 				} 				else {
 					triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN); 				} 				throw ex; 			} 			catch (RuntimeException | Error ex) {
 				if (!beforeCompletionInvoked) {
 					triggerBeforeCompletion(status); 				} 				doRollbackOnCommitException(status, ex); 				throw ex; 			} 			// Trigger afterCommit callbacks, with an exception thrown there 			// propagated to callers but the transaction still considered as committed. 			try {
 				triggerAfterCommit(status); 			} 			finally {
 				triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED); 			} 		} 		finally {
 			cleanupAfterCompletion(status); 		} 	} </pre></td> 
 </tr> 
 </tbody> 
</table>

![640?wx\_fmt=png][640_wx_fmt_png]

[Spring]: https://mp.weixin.qq.com/s?__biz=MzU5MDgzOTYzMw==&mid=2247484623&idx=1&sn=8d0399c8117dbc8287dd3f34c4048f03&scene=21#wechat_redirect
[640_wx_fmt_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/10aaee69838949cba94789df7c6b08f7.png