SpringAOP 源码之事务

发表于 更新于 分类于

xml 配置

1
2
3
4
5
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
	<constructor-arg ref="dataSource"/>
</bean>
<!-- 启用声明式事务 -->
<tx:annotation-driven />

xml 解析

TxNamespaceHandler

定位: org.springframework.transaction.config.TxNamespaceHandler

1
2
3
4
5
6
7
8
9
10
11
12
13
// 如果事务注解没有配置事务管理器引用, 使用 Spring 容器中的 transactionManager 事务管理器
static String getTransactionManagerName(Element element) {
	return (element.hasAttribute(TRANSACTION_MANAGER_ATTRIBUTE) ?
			element.getAttribute(TRANSACTION_MANAGER_ATTRIBUTE) : DEFAULT_TRANSACTION_MANAGER_BEAN_NAME);
}

@Override
public void init() {
	// 注册不同的 xml 文件标签解析器
	registerBeanDefinitionParser("advice", new TxAdviceBeanDefinitionParser());
	registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenBeanDefinitionParser());
	registerBeanDefinitionParser("jta-transaction-manager", new JtaTransactionManagerBeanDefinitionParser());
}

AnnotationDrivenBeanDefinitionParser

定位: org.springframework.transaction.config.AnnotationDrivenBeanDefinitionParser

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 解析 annotation-driven 标签
public BeanDefinition parse(Element element, ParserContext parserContext) {
    // 注册事务事件监听工厂
	registerTransactionalEventListenerFactory(parserContext);
	String mode = element.getAttribute("mode");
	if ("aspectj".equals(mode)) {
		// mode="aspectj"
		registerTransactionAspect(element, parserContext);
		if (ClassUtils.isPresent("javax.transaction.Transactional", getClass().getClassLoader())) {
			registerJtaTransactionAspect(element, parserContext);
		}
	} else {
		// mode="proxy"  // 默认配置
		AopAutoProxyConfigurer.configureAutoProxyCreator(element, parserContext);
	}
	return null;
}

AopAutoProxyConfigurer

定位: org.springframework.transaction.config.AnnotationDrivenBeanDefinitionParser.AopAutoProxyConfigurer

作用: 依据 AnnotationTransactionAttributeSource 和 TransactionInterceptor 元信息构造出 BeanFactoryTransactionAttributeSourceAdvisor 的元信息

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
private static class AopAutoProxyConfigurer {
	public static void configureAutoProxyCreator(Element element, ParserContext parserContext) {
        // aop 注册
		AopNamespaceUtils.registerAutoProxyCreatorIfNecessary(parserContext, element);

		String txAdvisorBeanName = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME;
		if (!parserContext.getRegistry().containsBeanDefinition(txAdvisorBeanName)) {
			Object eleSource = parserContext.extractSource(element);

			// Create the TransactionAttributeSource definition.
			// 创建 TransactionAttributeSource 的 bean
			RootBeanDefinition sourceDef = new RootBeanDefinition(
					"org.springframework.transaction.annotation.AnnotationTransactionAttributeSource");
			sourceDef.setSource(eleSource);
			sourceDef.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
			// 注册 bean，并使用Spring中的定义规则生成 beanName
			String sourceName = parserContext.getReaderContext().registerWithGeneratedName(sourceDef);

			// Create the TransactionInterceptor definition.
			// 创建 TransactionInterceptor 的 bean
			RootBeanDefinition interceptorDef = new RootBeanDefinition(TransactionInterceptor.class);
			interceptorDef.setSource(eleSource);
			interceptorDef.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
            // 设置事务管理器名称
			registerTransactionManager(element, interceptorDef);
			interceptorDef.getPropertyValues().add("transactionAttributeSource", new RuntimeBeanReference(sourceName));
			String interceptorName = parserContext.getReaderContext().registerWithGeneratedName(interceptorDef);

			// Create the TransactionAttributeSourceAdvisor definition.
			// 创建 TransactionAttributeSourceAdvisor 的 bean
			RootBeanDefinition advisorDef = new RootBeanDefinition(BeanFactoryTransactionAttributeSourceAdvisor.class);
			advisorDef.setSource(eleSource);
			advisorDef.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
			// 将 sourceName 的 bean 注入 advisorDef 的 transactionAttributeSource 属性中
			advisorDef.getPropertyValues().add("transactionAttributeSource", new RuntimeBeanReference(sourceName));
			// 将 interceptorName 的 bean 注入 advisorDef 的 adviceBeanName 属性中
			advisorDef.getPropertyValues().add("adviceBeanName", interceptorName);
			// 如果配置了 order 属性，则加入到 bean 中
			if (element.hasAttribute("order")) {
				advisorDef.getPropertyValues().add("order", element.getAttribute("order"));
			}
			parserContext.getRegistry().registerBeanDefinition(txAdvisorBeanName, advisorDef);

			CompositeComponentDefinition compositeDef = new CompositeComponentDefinition(element.getTagName(), eleSource);
			compositeDef.addNestedComponent(new BeanComponentDefinition(sourceDef, sourceName));
			compositeDef.addNestedComponent(new BeanComponentDefinition(interceptorDef, interceptorName));
			compositeDef.addNestedComponent(new BeanComponentDefinition(advisorDef, txAdvisorBeanName));
			parserContext.registerComponent(compositeDef);
		}
	}
}

TransactionAttributeSourceAdvisor

定位: org.springframework.transaction.interceptor.BeanFactoryTransactionAttributeSourceAdvisor

将 pointcut 和 advice 封装成 Advisor 对象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class TransactionAttributeSourceAdvisor extends AbstractPointcutAdvisor {
	@Nullable
	private TransactionInterceptor transactionInterceptor;
	// pointcut 属性赋值
	private final TransactionAttributeSourcePointcut pointcut = new TransactionAttributeSourcePointcut() {
		@Override
		@Nullable
		protected TransactionAttributeSource getTransactionAttributeSource() {
			return (transactionInterceptor != null ? transactionInterceptor.getTransactionAttributeSource() : null);
		}
	};
    @Override
	public Advice getAdvice() {
		Assert.state(this.transactionInterceptor != null, "No TransactionInterceptor set");
		return this.transactionInterceptor;
	}
    @Override
	public Pointcut getPointcut() {
		return this.pointcut;
	}
}

pointcut 匹配规则

定位:org.springframework.transaction.interceptor.TransactionAttributeSourcePointcut.TransactionAttributeSourceClassFilter#matches

类匹配

1
2
3
4
5
6
7
8
9
10
11
12
private class TransactionAttributeSourceClassFilter implements ClassFilter {
	@Override
	public boolean matches(Class<?> clazz) {
		if (TransactionalProxy.class.isAssignableFrom(clazz) ||
				TransactionManager.class.isAssignableFrom(clazz) ||
				PersistenceExceptionTranslator.class.isAssignableFrom(clazz)) {
			return false;
		}
		TransactionAttributeSource tas = getTransactionAttributeSource();
		return (tas == null || tas.isCandidateClass(clazz));
	}
}

定位: org.springframework.transaction.interceptor.TransactionAttributeSourcePointcut#matches

方法匹配

1
2
3
4
5
6
7
8
9
10
@Override
public boolean matches(Method method, Class<?> targetClass) {
	/**
	 * 获取我们@EnableTransactionManagement注解为我们容器中导入的ProxyTransactionManagementConfiguration
	 * 配置类中的TransactionAttributeSource对象
	 */
	TransactionAttributeSource tas = getTransactionAttributeSource();
	// 若事务属性原为null或者解析出来的事务注解属性不为空，表示方法匹配
	return (tas == null || tas.getTransactionAttribute(method, targetClass) != null);
}

advice 通知

TransactionInterceptor

此类实现了 MethodInterceptor 接口，继承了 TransactionAspectSupport 类

定位: org.springframework.transaction.interceptor.TransactionInterceptor#invoke

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@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属性
	Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);

	// Adapt to TransactionAspectSupport's invokeWithinTransaction...
	/** 
	 * 以事务的方式调用目标方法
	 * 在这埋了一个钩子函数 用来回调目标方法的
	 */
	return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
}

TransactionAspectSupport

invokeWithinTransaction

定位: org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction

事务开启, 回滚和提交

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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@Nullable
protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
		final InvocationCallback invocation) throws Throwable {
	// If the transaction attribute is null, the method is non-transactional.
	// 获取我们的事务属性源对象
	TransactionAttributeSource tas = getTransactionAttributeSource();
	// 通过事务属性源对象获取到当前方法的事务属性信息
	final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
	// 获取我们配置的事务管理器对象
	final TransactionManager tm = determineTransactionManager(txAttr);

	if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {
		ReactiveTransactionSupport txSupport = this.transactionSupportCache.computeIfAbsent(method, key -> {
			if (KotlinDetector.isKotlinType(method.getDeclaringClass()) && KotlinDelegate.isSuspend(method)) {
				throw new TransactionUsageException(
						"Unsupported annotated transaction on suspending function detected: " + method +
						". Use TransactionalOperator.transactional extensions instead.");
			}
			ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(method.getReturnType());
			if (adapter == null) {
				throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " +
						method.getReturnType());
			}
			return new ReactiveTransactionSupport(adapter);
		});
		return txSupport.invokeWithinTransaction(
				method, targetClass, invocation, txAttr, (ReactiveTransactionManager) tm);
	}

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

	else {
		// 编程式事务处理
		Object result;
		final ThrowableHolder throwableHolder = new ThrowableHolder();

		// It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
		try {
			result = ((CallbackPreferringPlatformTransactionManager) ptm).execute(txAttr, status -> {
				TransactionInfo txInfo = prepareTransactionInfo(ptm, txAttr, joinpointIdentification, status);
				try {
					Object retVal = invocation.proceedWithInvocation();
					if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
						// Set rollback-only in case of Vavr failure matching our rollback rules...
						retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
					}
					return retVal;
				}
				catch (Throwable ex) {
					if (txAttr.rollbackOn(ex)) {
						// A RuntimeException: will lead to a rollback.
						if (ex instanceof RuntimeException) {
							throw (RuntimeException) ex;
						}
						else {
							throw new ThrowableHolderException(ex);
						}
					}
					else {
						// A normal return value: will lead to a commit.
						throwableHolder.throwable = ex;
						return null;
					}
				}
				finally {
					cleanupTransactionInfo(txInfo);
				}
			});
		}
		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;
		}

		// Check result state: It might indicate a Throwable to rethrow.
		if (throwableHolder.throwable != null) {
			throw throwableHolder.throwable;
		}
		return result;
	}
}

createTransactionIfNecessary

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
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
		@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {

	// If no name specified, apply method identification as transaction name.
	// 如果没有名称指定则使用方法唯一标识，并使用 DelegatingTransactionAttribute 封装 txAttr
	if (txAttr != null && txAttr.getName() == null) {
		txAttr = new DelegatingTransactionAttribute(txAttr) {
			@Override
			public String getName() {
				return joinpointIdentification;
			}
		};
	}

	TransactionStatus status = null;
	if (txAttr != null) {
		if (tm != null) {
			// 获取 TransactionStatus 事务状态信息
			status = tm.getTransaction(txAttr);
		}
		else {
			if (logger.isDebugEnabled()) {
				logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
						"] because no transaction manager has been configured");
			}
		}
	}
	// 根据指定的属性与status准备一个TransactionInfo，
	return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

prepareTransactionInfo

事务的准备工作, 开启事务/设置事务状态/事务信息绑定到当前线程

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
protected TransactionInfo prepareTransactionInfo(@Nullable PlatformTransactionManager tm,
		@Nullable TransactionAttribute txAttr, String joinpointIdentification,
		@Nullable TransactionStatus status) {

	// 创建事务信息
	TransactionInfo txInfo = new TransactionInfo(tm, txAttr, joinpointIdentification);
	if (txAttr != null) {
		// We need a transaction for this method...
		if (logger.isTraceEnabled()) {
			logger.trace("Getting transaction for [" + txInfo.getJoinpointIdentification() + "]");
		}
		// The transaction manager will flag an error if an incompatible tx already exists.
		// 设置新事务状态
		txInfo.newTransactionStatus(status);
	}
	else {
		// The TransactionInfo.hasTransaction() method will return false. We created it only
		// to preserve the integrity of the ThreadLocal stack maintained in this class.
		if (logger.isTraceEnabled()) {
			logger.trace("No need to create transaction for [" + joinpointIdentification +
					"]: This method is not transactional.");
		}
	}

	// We always bind the TransactionInfo to the thread, even if we didn't create
	// a new transaction here. This guarantees that the TransactionInfo stack
	// will be managed correctly even if no transaction was created by this aspect.
	// 事务信息绑定到当前线程
	txInfo.bindToThread();
	return txInfo;
}

DataSourceTransactionManager

jdbc 数据源是事务管理器，继承自 AbstractPlatformTransactionManager

doBegin

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
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");
			}
			// 把我们的数据库连接包装成一个ConnectionHolder 对象 然后设置到我们的 txObject 对象中去
			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).
		// 关闭自动提交
		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.
		// 绑定我们的数据源和连接到我们的同步管理器上，把数据源作为key,数据库连接作为value 设置到线程变量中
		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 JDBC Connection for transaction", ex);
	}
}
protected void prepareTransactionalConnection(Connection con, TransactionDefinition definition)
		throws SQLException {
	if (isEnforceReadOnly() && definition.isReadOnly()) {
		try (Statement stmt = con.createStatement()) {
            // 设置当前会话事务只读
			stmt.executeUpdate("SET TRANSACTION READ ONLY");
		}
	}
}

@Override
protected Object doGetTransaction() {
	// 创建一个数据源事务对象
	DataSourceTransactionObject txObject = new DataSourceTransactionObject();
	// 是否允许当前事务设置保持点
	txObject.setSavepointAllowed(isNestedTransactionAllowed());
	/**
	 * TransactionSynchronizationManager 事务同步管理器对象(该类中都是局部线程变量)
	 * 用来保存当前事务的信息,我们第一次从这里去线程变量中获取 事务连接持有器对象 通过数据源为 key 去获取
	 * 由于第一次进来开始事务 我们的事务同步管理器中没有被存放.所以此时获取出来的 conHolder 为 null
	 */
	ConnectionHolder conHolder =
			(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
	txObject.setConnectionHolder(conHolder, false);
	// 返回事务对象
	return txObject;
}

doCleanupAfterCompletion

此方法做清除连接相关操作，比如重置自动提交啊，只读属性啊，解绑数据源啊，释放连接啊，清除链接持有器属性

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
@Override
protected void doCleanupAfterCompletion(Object transaction) {
	DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;

	// Remove the connection holder from the thread, if exposed.
	if (txObject.isNewConnectionHolder()) {
		// 将数据库连接从当前线程中解除绑定
		TransactionSynchronizationManager.unbindResource(obtainDataSource());
	}

	// Reset connection.
	// 释放连接
	Connection con = txObject.getConnectionHolder().getConnection();
	try {
		if (txObject.isMustRestoreAutoCommit()) {
			// 恢复数据库连接的自动提交属性
			con.setAutoCommit(true);
		}
		// 重置数据库连接
		DataSourceUtils.resetConnectionAfterTransaction(
				con, txObject.getPreviousIsolationLevel(), txObject.isReadOnly());
	}
	catch (Throwable ex) {
		logger.debug("Could not reset JDBC Connection after transaction", ex);
	}

	if (txObject.isNewConnectionHolder()) {
		if (logger.isDebugEnabled()) {
			logger.debug("Releasing JDBC Connection [" + con + "] after transaction");
		}
		// 如果当前事务是独立的新创建的事务则在事务完成时释放数据库连接
		DataSourceUtils.releaseConnection(con, this.dataSource);
	}

	// 连接持有器属性清除
	txObject.getConnectionHolder().clear();
}

AbstractPlatformTransactionManager

getTransaction

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
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();

	// 判断当前线程是否存在事务，判断依据为当前线程记录的连接不为空且连接中的 transactionActive 属性不为空
	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.
	// 如果当前线程不存在事务，但是 PropagationBehavior 却被声明为 PROPAGATION_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，PROPAGATION_NESTED 都需要新建事务
	else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
			def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
			def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
		// 没有当前事务的话，REQUIRED，REQUIRES_NEW，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);
	}
}
// 开启新事务
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;
}

handleExistingTransaction

在开始事务时，如果当前线程是否存在事务，则进行嵌套事务的处理

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
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 {
			// 创建一个新的非事务状态(保存了上一个存在事务状态的属性)
			return startTransaction(definition, transaction, debugEnabled, suspendedResources);
		}
		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.
			// 有些情况是不能使用保存点操作
			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);
}

rollback

事务管理器根据事务状态来处理回滚

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
public final void rollback(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;
	processRollback(defStatus, false);
}
/**
 * unexpected 一般是 false，除非是设置 rollback-only=true，才是true，表示是全局的回滚标记。首先会进行回滚前回调，
 * 然后判断是否设置了保存点，比如 NESTED 会设置，要先回滚到保存点。如果状态是新的事务，那就进行回滚，如果不是新的，就设置一个回滚标记，
 * 内部是设置连接持有器回滚标记。然后回滚完成回调，根据事务状态信息，完成后数据清除,和线程的私有资源解绑，
 * 重置连接自动提交，隔离级别，是否只读，释放连接，恢复挂起事务等
 */
private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
	try {
		// 意外的回滚
		boolean unexpectedRollback = unexpected;

		try {
			// 回滚完成前回调
			triggerBeforeCompletion(status);
			// 有保存点回滚到保存点
			if (status.hasSavepoint()) {
				if (status.isDebug()) {
					logger.debug("Rolling back transaction to savepoint");
				}
				status.rollbackToHeldSavepoint();
			}
			// 当前状态是一个新事务
			else if (status.isNewTransaction()) {
				if (status.isDebug()) {
					logger.debug("Initiating transaction rollback");
				}
				// 进行回滚
				doRollback(status);
			}
			else {
				// Participating in larger transaction
				if (status.hasTransaction()) {
					if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
						if (status.isDebug()) {
							logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
						}
						//设置连接要回滚标记，也就是全局回滚
						doSetRollbackOnly(status);
					}
					else {
						if (status.isDebug()) {
							logger.debug("Participating transaction failed - letting transaction originator decide on rollback");
						}
					}
				}
				else {
					logger.debug("Should roll back transaction but cannot - no transaction available");
				}
				// Unexpected rollback only matters here if we're asked to fail early
				if (!isFailEarlyOnGlobalRollbackOnly()) {
					unexpectedRollback = false;
				}
			}
		}
		catch (RuntimeException | Error ex) {
			triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
			throw ex;
		}

		// 回滚完成后回调
		triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);

		// Raise UnexpectedRollbackException if we had a global rollback-only marker
		if (unexpectedRollback) {
			throw new UnexpectedRollbackException(
					"Transaction rolled back because it has been marked as rollback-only");
		}
	}
	finally {
		// 根据事务状态信息，完成后数据清除，和线程的私有资源解绑，重置连接自动提交，隔离级别，是否只读，释放连接，恢复挂起事务等
		cleanupAfterCompletion(status);
	}
}

commit

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
112
113
114
115
116
117
118
119
120
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() && 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);
			// 添加的 TransactionSynchronization 中的对应方法的调用
			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();
				// 如果是独立的事务则直接提交
				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);
	}
}