Spring嵌套事物的执行过程异常(UnexpectedRollbackException)的分析，需要注意一下两点：
这里不是指传播性Propagation.NESTED嵌套事物，而是指方法间调用嵌套不同传播性事物
Spring AOP一个类方法之间的调用不会走代理类，可以通过以下两种方式可以获取到当前代理对象
@EnableAspectProxy#exposeProxy()为true
xml配置<aop:aspectj-autoproxy expose-proxy="true"/>
通过以上两种方式暴露代理对象，通过AopContext.currentProxy()可以获取到当前代理对象
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:tx="http://www.springframework.org/schema/tx"
xsi:schemaLocation="http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://www.springframework.org/schema/aop
https://www.springframework.org/schema/aop/spring-aop-2.0.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/tx
https://www.springframework.org/schema/tx/spring-tx.xsd">
<context:component-scan base-package="com.zli.tx"/>
<bean id="dataSource"
class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<property name="driverClassName" value="${jdbc.driverClassName}"/>
<property name="url" value="${jdbc.url}"/>
<property name="username" value="${jdbc.username}"/>
<property name="password" value="${jdbc.password}"/>
</bean>
<context:property-placeholder location="jdbc.properties"/>
<!-- 暴露代理对象 可以通过AopContext.currentProxy获取当前代理类-->
<aop:aspectj-autoproxy expose-proxy="true"/>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
<tx:annotation-driven transaction-manager="transactionManager"/>
</beans>
@Service
public class DemoServiceImpl implements DemoService {
@Resource
private DataSource dataSource;
@Transactional(rollbackFor = Exception.class)
@Override
public void save() throws Exception {
System.out.println("-------save before--------");
executeSQL("insert into auth_user(id, name, account, pwd) values(1, 'Jack', 'user', '123456')");
// 方法间调用不走代理对象，而且通过当前对象this调用
// 此处通过暴露后的代理类型调用
((DemoService) AopContext.currentProxy()).firstUpdate();
System.out.println("-------save after--------");
}
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
@Override
public void firstUpdate() throws Exception {
System.out.println("-------firstUpdate before--------");
try {
executeSQL("insert into auth_user(id, name, account, pwd) values(2, 'John', 'user', '123456')");
((DemoService) AopContext.currentProxy()).secondUpdate();
}catch (Exception e) {
}
System.out.println("-------firstUpdate after--------");
}
@Transactional(rollbackFor = Exception.class)
@Override
public void secondUpdate() throws Exception {
System.out.println("-------secondUpdate before--------");
executeSQL("insert into auth_user(id, name, account, pwd) values(3, 'Fake', 'user', '123456')");
System.out.println("-------secondUpdate after--------");
throw new Exception();
}
/**
* 获取数据库连接执行sql
* @param sql
*/
public void executeSQL(String sql) {
Connection connection = DataSourceUtils.getConnection(dataSource);
try {
connection.createStatement().executeUpdate(sql);
}
catch (SQLException throwables) {
throw new RuntimeException(throwables);
}
}
}
public class DataAccessObjectTests {
@Test
public void txPropagationTest() throws Exception {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath*:META-INF/tx-context.xml");
DemoService demoService = applicationContext.getBean(DemoService.class);
// 执行保存
demoService.save();
applicationContext.close();
}
}
执行后报的异常，全局回滚，接下来分析为什么会报该异常，事物如何全局回滚的
首先参考一下官方的声明式事物实现图
1.当执行save方法时，会通过Advisor对进行方法拦截（事物拦截核心类TransactionInterceptor）；动态代理（JDK、CGLIB）对象会调用TransactionInterceptor该类的invoke方法，然后执行目标对象方法
1public class TransactionInterceptor extends TransactionAspectSupport implements MethodInterceptor, Serializable {
2 ...
3
4 @Override
5 @Nullable
6 public Object invoke(MethodInvocation invocation) throws Throwable {
7 // Work out the target class: may be {@code null}.
8 // The TransactionAttributeSource should be passed the target class
9 // as well as the method, which may be from an interface.
10
11 // targetClass 可能为null，方法来自接口
12 Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
13 // 位置1
14 // Adapt to TransactionAspectSupport's invokeWithinTransaction...
15 return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
16 }
17 ...
18}
1public abstract class TransactionAspectSupport implements BeanFactoryAware, InitializingBean {
2...
3
4@Nullable
5 protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
6 final InvocationCallback invocation) throws Throwable {
7
8 // 获取事物属性资源，如AnnotationTransactionAttributeSource
9 // If the transaction attribute is null, the method is non-transactional.
10 TransactionAttributeSource tas = getTransactionAttributeSource();
11 // 获取事物定义的属性，如propagation、isolation、transactionManager等
12 final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
13 // 获取事物管理器，这里获取xml配置的DataSourceTransactionManager事物管理器
14 final TransactionManager tm = determineTransactionManager(txAttr);
15
16 // 响应式事物管理器
17 ...
18
19 //
20 PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
21 // 获取方法定义连接点，主要用于当作事物名称
22 final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
23
24 // 声明式事物
25 if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
26 // 位置2
27 // 如果需要则创建事物
28 // Standard transaction demarcation with getTransaction and commit/rollback calls.
29 TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
30
31 Object retVal;
32 try {
33 // 方法1 → 方法2 → 方法3
34 // ↓
35 // 方法1 ← 方法2 ← 方法3
36
37 // 这里是一个环绕通知，通常会调用下一个调用链，即调用目标对象方法
38 // This is an around advice: Invoke the next interceptor in the chain.
39 // This will normally result in a target object being invoked.
40 retVal = invocation.proceedWithInvocation();
41 }
42 catch (Throwable ex) {
43 // 位置3
44 // target invocation exception
45 // 异常后处理事物回滚，如果目标对象方法catch了异常，就不会触发该方法
46 completeTransactionAfterThrowing(txInfo, ex);
47 throw ex;
48 }
49 finally {
50 // 将当前线程的事物信息重置到上一次的事物信息，即oldTransactionInfo
51 cleanupTransactionInfo(txInfo);
52 }
53
54 if (vavrPresent && VavrDelegate.isVavrTry(retVal)) {
55 // Set rollback-only in case of Vavr failure matching our rollback rules...
56 TransactionStatus status = txInfo.getTransactionStatus();
57 if (status != null && txAttr != null) {
58 retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
59 }
60 }
61
62 // 位置4
63 // 提交事物
64 commitTransactionAfterReturning(txInfo);
65 return retVal;
66 }
67 else {
68 final ThrowableHolder throwableHolder = new ThrowableHolder();
69
70 // 编程式事物
71 ...
72 }
73 }
74
75 ...
76}
public abstract class TransactionAspectSupport implements BeanFactoryAware, InitializingBean {
...
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
// 如果未指定名称，则将方法标识用作事务名称。
// If no name specified, apply method identification as transaction name.
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) {// 位置2.1
// 通过事物管理器获取事物状态
status = tm.getTransaction(txAttr);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
"] because no transaction manager has been configured");
}
}
}
// 预处理事物信息，封装TransactionInfo
return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}
...
}
2.分析位置2.1，通过DataSourceTransactionManager事物管理器获取事物状态，下图为是事物管理器的抽象实现
1public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
2 ...
3
4 @Override
5 public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
6 throws TransactionException {
7
8 // Use defaults if no transaction definition given.
9 TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
10
11 // 从DataSourceTransactionManager获取一个事物对象，封装事物持有者Holder(封装连接、SavePoint等)
12 // 从TransactionSynchronizationManager#getSource获取当前线程资源，第一次进来为空
13 Object transaction = doGetTransaction();
14 boolean debugEnabled = logger.isDebugEnabled();
15
16 // 位置2.2
17 // 如果当前存在事物
18 if (isExistingTransaction(transaction)) {
19 // 处理存在的事物
20 // Existing transaction found -> check propagation behavior to find out how to behave.
21 return handleExistingTransaction(def, transaction, debugEnabled);
22 }
23
24 // Check definition settings for new transaction.
25 if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
26 throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
27 }
28
29 // 支持当前事务，假设当前没有事务，就抛出异常
30 // No existing transaction found -> check propagation behavior to find out how to proceed.
31 if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
32 throw new IllegalTransactionStateException(
33 "No existing transaction found for transaction marked with propagation 'mandatory'");
34 }
35 else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
36 def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
37 def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
38 // 位置2.3
39 // 当前不存在事物也挂起
40 SuspendedResourcesHolder suspendedResources = suspend(null);
41 if (debugEnabled) {
42 logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
43 }
44 try {
45 // 位置2.4
46 // 创建事物
47 return startTransaction(def, transaction, debugEnabled, suspendedResources);
48 }
49 catch (RuntimeException | Error ex) {
50 resume(null, suspendedResources);
51 throw ex;
52 }
53 }
54 else {
55 // Create "empty" transaction: no actual transaction, but potentially synchronization.
56 if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
57 logger.warn("Custom isolation level specified but no actual transaction initiated; " +
58 "isolation level will effectively be ignored: " + def);
59 }
60
61 // 创建空事物
62 boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
63 return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
64 }
65 }
66
67 ...
68}
1public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
2 ...
3
4 private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
5 boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
6
7 // 是否新的同步，transactionSynchronization默认是0
8 boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
9 // 创建一个默认新事物状态
10 DefaultTransactionStatus status = newTransactionStatus(
11 definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
12
13 // 位置2.4.1
14 // 开始事物
15 // 以DataSourceTransactionManager为例，将连接设置为非自动提交、只读等
16 doBegin(transaction, definition);
17
18 // 位置2.4.2
19 // 预处理需要初始化事务同步
20 prepareSynchronization(status, definition);
21 return status;
22 }
23
24 protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
25 // 如果是新事物
26 if (status.isNewSynchronization()) {
27 // 标记当前线程事物是否存活
28 TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
29 // 设置当前线程事物隔离级别
30 TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(
31 definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ?
32 definition.getIsolationLevel() : null);
33 // 设置当前线程事物只读
34 TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());
35 // 设置当前线程事物名称
36 TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());
37 // 初始化当前线程事物同步器
38 TransactionSynchronizationManager.initSynchronization();
39 }
40 }
41 ...
42}
1public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
2 ...
3
4 private TransactionStatus handleExistingTransaction(
5 TransactionDefinition definition, Object transaction, boolean debugEnabled)
6 throws TransactionException {
7
8 ...
9
10 // 位置2.2.1
11 // 新建事务，假设当前存在事务。把当前事务挂起
12 if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
13 if (debugEnabled) {
14 logger.debug("Suspending current transaction, creating new transaction with name [" +
15 definition.getName() + "]");
16 }
17 // 挂起当前线程存在的事物
18 SuspendedResourcesHolder suspendedResources = suspend(transaction);
19 try {
20 // 创建新事物
21 return startTransaction(definition, transaction, debugEnabled, suspendedResources);
22 }
23 catch (RuntimeException | Error beginEx) {
24 resumeAfterBeginException(transaction, suspendedResources, beginEx);
25 throw beginEx;
26 }
27 }
28
29 // 如果当前存在事务，则在嵌套事务内执行。如果当前没有事务，则进行与PROPAGATION_REQUIRED类似的操作
30 if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
31 if (!isNestedTransactionAllowed()) {
32 throw new NestedTransactionNotSupportedException(
33 "Transaction manager does not allow nested transactions by default - " +
34 "specify 'nestedTransactionAllowed' property with value 'true'");
35 }
36 if (debugEnabled) {
37 logger.debug("Creating nested transaction with name [" + definition.getName() + "]");
38 }
39 // 嵌套事物是否使用SavePoint
40 if (useSavepointForNestedTransaction()) {
41 // Create savepoint within existing Spring-managed transaction,
42 // through the SavepointManager API implemented by TransactionStatus.
43 // Usually uses JDBC 3.0 savepoints. Never activates Spring synchronization.
44 DefaultTransactionStatus status =
45 prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
46 // 在现有事物中创建一个SavePoint
47 status.createAndHoldSavepoint();
48 return status;
49 }
50 else {
51 // Nested transaction through nested begin and commit/rollback calls.
52 // Usually only for JTA: Spring synchronization might get activated here
53 // in case of a pre-existing JTA transaction.
54 return startTransaction(definition, transaction, debugEnabled, null);
55 }
56 }
57
58 // Assumably PROPAGATION_SUPPORTS or PROPAGATION_REQUIRED.
59 if (debugEnabled) {
60 logger.debug("Participating in existing transaction");
61 }
62 // 校验事物隔离级别、只读
63 ...
64
65 // 位置2.2.2
66 // 创建事物（同一个事物）
67 boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
68 return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
69 }
70
71 ...
72}
1public abstract class TransactionAspectSupport implements BeanFactoryAware, InitializingBean {2 ...
3
4 protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {
5 if (txInfo != null && txInfo.getTransactionStatus() != null) {
6 if (logger.isTraceEnabled()) {
7 logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() +
8 "] after exception: " + ex);
9 }
10 // 判断异常需要回滚的
11 if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
12 try {
13 // 位置3.1
14 // 回滚
15 txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
16 }
17 catch (TransactionSystemException ex2) {
18 logger.error("Application exception overridden by rollback exception", ex);
19 ex2.initApplicationException(ex);
20 throw ex2;
21 }
22 catch (RuntimeException | Error ex2) {
23 logger.error("Application exception overridden by rollback exception", ex);
24 throw ex2;
25 }
26 }
27 else {
28 // 位置3.2
29 // 忽略的异常不会回滚
30 // 但是如果事物标记为回滚，则仍为回滚
31 // We don't roll back on this exception.
32 // Will still roll back if TransactionStatus.isRollbackOnly() is true.
33 try {
34 // 提交
35 txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
36 }
37 catch (TransactionSystemException ex2) {
38 logger.error("Application exception overridden by commit exception", ex);
39 ex2.initApplicationException(ex);
40 throw ex2;
41 }
42 catch (RuntimeException | Error ex2) {
43 logger.error("Application exception overridden by commit exception", ex);
44 throw ex2;
45 }
46 }
47 }
48 }
49 ...
50}
1public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
2
3 ...
4
5 @Override
6 public final void rollback(TransactionStatus status) throws TransactionException {
7 if (status.isCompleted()) {
8 throw new IllegalTransactionStateException(
9 "Transaction is already completed - do not call commit or rollback more than once per transaction");
10 }
11
12 DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
13 // 处理回滚，非意外回滚
14 processRollback(defStatus, false);
15 }
16
17 private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
18 try {
19 boolean unexpectedRollback = unexpected;
20
21 try {
22 // 回滚前触发回调事物同步器
23 // TransactionSynchronization#beforeCompletion
24 triggerBeforeCompletion(status);
25
26 // 嵌套事物
27 if (status.hasSavepoint()) {
28 if (status.isDebug()) {
29 logger.debug("Rolling back transaction to savepoint");
30 }
31 // 回滚到SavePoint
32 status.rollbackToHeldSavepoint();
33 }
34 // 新事物
35 else if (status.isNewTransaction()) {
36 if (status.isDebug()) {
37 logger.debug("Initiating transaction rollback");
38 }
39 // 模版方法，有子类实现
40 // 如DataSourceTransactionManager#doRollback
41 doRollback(status);
42 }
43 else {
44 // 多个事物
45 // Participating in larger transaction
46 // 是否存在事物
47 if (status.hasTransaction()) {
48 // isLocalRollbackOnly是否标记回滚
49 // isGlobalRollbackOnParticipationFailure是否标记全局回滚，默认为true
50 if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
51 if (status.isDebug()) {
52 logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
53 }
54 // 位置3.1.1
55 // 设置回滚
56 doSetRollbackOnly(status);
57 }
58 else {
59 if (status.isDebug()) {
60 logger.debug("Participating transaction failed - letting transaction originator decide on rollback");
61 }
62 }
63 }
64 else {
65 logger.debug("Should roll back transaction but cannot - no transaction available");
66 }
67 // Unexpected rollback only matters here if we're asked to fail early
68 if (!isFailEarlyOnGlobalRollbackOnly()) {
69 unexpectedRollback = false;
70 }
71 }
72 }
73 catch (RuntimeException | Error ex) {
74 triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
75 throw ex;
76 }
77
78 // 回滚后触发回调事物同步器
79 triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
8081 // 位置3.1.2// 如果我们有一个仅全局回滚的标记，则引发UnexpectedRollbackException
82 // Raise UnexpectedRollbackException if we had a global rollback-only marker
83 if (unexpectedRollback) {
84 throw new UnexpectedRollbackException(
85 "Transaction rolled back because it has been marked as rollback-only");
86 }
87 }
88 finally {
89 cleanupAfterCompletion(status);
90 }
91 }
92
93 ...
94}
1public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
2
3 ...
4
5 @Override
6 public final void commit(TransactionStatus status) throws TransactionException {
7 if (status.isCompleted()) {
8 throw new IllegalTransactionStateException(
9 "Transaction is already completed - do not call commit or rollback more than once per transaction");
10 }
11
12 DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
13 // 标记回滚
14 if (defStatus.isLocalRollbackOnly()) {
15 if (defStatus.isDebug()) {
16 logger.debug("Transactional code has requested rollback");
17 }
18 processRollback(defStatus, false);
19 return;
20 }
21 // 位置4.1
22 // 全局回滚
23 // 如果是同一数据源则当前线程绑定的connection holder是一样的
24 if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
25 if (defStatus.isDebug()) {
26 logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");
27 }
28 // 意外回滚
29 processRollback(defStatus, true);
30 return;
31 }
32
33 // 提交
34 processCommit(defStatus);
35 }
36
37 ...
38}
文章转载自帽爹的技术轮子，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
