一切福田，不離方寸，從心而覓，感無不通。

SpringAOP学习--Spring事务简介及原理

文章目录

事务简介

前篇介绍了SpringAOP，Spring事务是SpringAOP一个典型的应用。

事务即数据库事务，指同一批次对数据的读写要么全成功，要么全失败，用以保证数据的一致性，是关系统数据库核心功能。编程中通过设置事务手动提交，然后根据情况选择提交事务或者回滚事务。

数据库中事务使用:
BEGIN;#开始事务
update table_name set name=’XXX’ where id=’XXX’;#执行数据库操作
COMMIT; #提交
ROLLBACK;#回滚

Java jdbc使用事务使用如下

1
2
3
4
5
6
7
8
Connection connection = DriverManager.getConnection(url, username, password);
connection.setAutoCommit(false);
boolean error = doSomething…;
if (error){
connection.commit();
} else {
connection.rollback();
}

 

编程式事务与声明式事务

Spring事务是对jdbc事务的封装，上述jdbc事务可以看出，使用事务时除了doSmoething部分，其他都是固定的。
Spring将事务公共部分封装，封装了jdbc细节，提供编程方式或AOP方式使用事务，即编程式事务和声明式事务，其中声明式事务是本片的重点，编程式事务不做过多探究

1. 编程式事务：Spring提供模板代码用于包裹用户业务代码，根据用户代码执行结果进行事务提交或回滚。
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
protected void doInTransactionWithoutResult(TransactionStatus status) {
//doSomething…
}
});
或者可以直接使用Spring事务管理器获取事务，然后在代码中根据需要手动提交或者回滚。

2. 声明式事务：通过配置或者注解声明使用事务，用来告诉Spring哪里需要使用事务， Spring会对其生成aop切面，之后执行对应方法时Spring通过切面配置及用户代码执行结果进行事务提交或回滚。
@Transactiona
public void doSomething(){}

事务原理

声明式事务基于aop，使用声明式事务时，需要引入事务切面配置的Advisor。
老版本基于xml配置阶段需要配置Advisor和切点等，Spring3.1版本提供了EnableTransactionManagement注解，用于开启Spring事务。
使用这个注解会导入ProxyTransactionManagementConfiguration配置类。

我们来看下这个类都包含了什么

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
@Configuration(proxyBeanMethods = false)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {
 
@Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(
TransactionAttributeSource transactionAttributeSource, TransactionInterceptor transactionInterceptor) {
 
BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
advisor.setTransactionAttributeSource(transactionAttributeSource);
advisor.setAdvice(transactionInterceptor);
if (this.enableTx != null) {
advisor.setOrder(this.enableTx.<Integer>getNumber("order"));
}
return advisor;
}
 
@Bean
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public TransactionAttributeSource transactionAttributeSource() {
return new AnnotationTransactionAttributeSource();
}
 
@Bean
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public TransactionInterceptor transactionInterceptor(TransactionAttributeSource transactionAttributeSource) {
TransactionInterceptor interceptor = new TransactionInterceptor();
interceptor.setTransactionAttributeSource(transactionAttributeSource);
if (this.txManager != null) {
interceptor.setTransactionManager(this.txManager);
}
return interceptor;
}
 
}

 

可以看到其中定义了三个bean
BeanFactoryTransactionAttributeSourceAdvisor
TransactionAttributeSource -> AnnotationTransactionAttributeSource
TransactionInterceptor

事务配置

回忆下SpringAOP内容：
《SpringAOP学习–SpringAOP简介及原理》

使用直接定义Advisor方式定义AOP，需要定义
Advisor、Advice、PointCut

上面的三个bean与定义AOP需要的bean如何对应呢？
BeanFactoryTransactionAttributeSourceAdvisor == Advisor
TransactionInterceptor = Advice
TransactionAttributeSource ≈ PointCut

TransactionAttributeSource为什么约等于PointCut呢？
PointCut实际在BeanFactoryTransactionAttributeSourceAdvisor定义了，为TransactionAttributeSourcePointcut，
而为TransactionAttributeSourcePointcut包含了TransactionAttributeSource

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class BeanFactoryTransactionAttributeSourceAdvisor extends AbstractBeanFactoryPointcutAdvisor {
...
private final TransactionAttributeSourcePointcut pointcut = new TransactionAttributeSourcePointcut() {
@Override
@Nullable
protected TransactionAttributeSource getTransactionAttributeSource() {
return transactionAttributeSource;
}
};
...
}
 
abstract class TransactionAttributeSourcePointcut extends StaticMethodMatcherPointcut implements Serializable {
...
@Override
public boolean matches(Method method, Class<?> targetClass) {
TransactionAttributeSource tas = getTransactionAttributeSource();
return (tas == null || tas.getTransactionAttribute(method, targetClass) != null);
}
...
}

 

BeanFactoryTransactionAttributeSourceAdvisor作为Advisor包含了事务Advise拦截器TransactionInterceptor与PointCut切点收集过滤器TransactionAttributeSourcePointcut。
满足AOP切面配置所需的必要条件，可以生成及执行AOP过程。具体的事务处理逻辑即在拦截器TransactionInterceptor中。
至此事务AOP配置已经确定好了

事务拦截器

下面来看看拦截器TransactionInterceptor如何工作的，TransactionInterceptor作为拦截器，提供invoke方法供动态代理对象调用（如CglibAopProxy），invoke方法内调用其父类方法TransactionAspectSupport .invokeWithinTransaction

TransactionInterceptor.invoke -> TransactionInterceptor.invokeWithinTransaction

1.获取当前bean对应的事务的属性配置TransactionAttribute，TransactionAttributeSource会在代理阶段提前收集到所有AOP代理bean的TransactionAttribute，这里只是取缓存

2.根据TransactionAttribute获取一个事务管理器，没有特殊配置时会根据类型(TransactionManager.class)获取。默认为DataSourceTransactionManager

3.非CallbackPreferringPlatformTransactionManager事务管理器
执行createTransactionIfNecessary创建(或获取)事务TransactionInfo

4.执行MethodInvocation.proceed，调用代理对象的invoke执行目标方法，记录目标方法的返回值，并对其进行try->catch->finally，根据结果处理事务

5.异常时 catch中调用completeTransactionAfterThrowing处理事务回滚(也有可能提交)，并再次抛出异常，切面结束

6.未异常时 调用commitTransactionAfterReturning处理事务提交

7.返回目标方法返回值

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
@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 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创建事务，使用PlatformTransactionManager.getTransaction事务管理器获取事务
PlatformTransactionManager.getTransaction是一个接口方法，由AbstractPlatformTransactionManager实现。
根据当前是否存在事务以及配置的事务传播行来决定创建还是使用一个已有的事务，期间还可能抛出异常或者挂起事务。

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
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();
 
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.
if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
throw new IllegalTransactionStateException(
"No existing transaction found for transaction marked with propagation 'mandatory'");
}
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;
}
}
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);
}
}

 

在这里插入图片描述

事务执行

原对象方法通过反射执行后，发生异常执行异常处理，否则执行事务提交。

在这里插入图片描述

这里提交回滚都不是绝对的，不发生异常也可能回滚，发生异常也可能提交。

首先commit方法也存在回滚的可能，正常执行自然是提交。如果应用设置回滚标记LocalRollbackOnly，那么执行提交方法也会回滚。

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
@Override
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);
}

 

同样发生异常时执行completeTransactionAfterThrowing，首先回匹配异常类型来判断是否回滚，如果程序抛出的异常和事务指定的异常不匹配则不会回滚而是执行提交操作

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
protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {
if (txInfo != null && txInfo.getTransactionStatus() != null) {
if (logger.isTraceEnabled()) {
logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() +
"] after exception: " + ex);
}
if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
try {
txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
}
catch (TransactionSystemException ex2) {
logger.error("Application exception overridden by rollback exception", ex);
ex2.initApplicationException(ex);
throw ex2;
}
catch (RuntimeException | Error ex2) {
logger.error("Application exception overridden by rollback exception", ex);
throw ex2;
}
}
else {
// We don't roll back on this exception.
// Will still roll back if TransactionStatus.isRollbackOnly() is true.
try {
txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
}
catch (TransactionSystemException ex2) {
logger.error("Application exception overridden by commit exception", ex);
ex2.initApplicationException(ex);
throw ex2;
}
catch (RuntimeException | Error ex2) {
logger.error("Application exception overridden by commit exception", ex);
throw ex2;
}
}
}
}

 

关于异常匹配这里有一点需要注意，这里不仅会匹配@Transactional(rollbackFor = Exception.class)中指定的异常，还会执行super.rollbackOn 即发生运行时异常及其子类会直接回滚，当我们需要发生特定异常才回滚时，应当自定义异常且不能继承RuntimeException

1
2
3
4
@Override
public boolean rollbackOn(Throwable ex) {
return (ex instanceof RuntimeException || ex instanceof Error);
}

 

事务传播性

@Transactional中有一项重要配置Propagation，即指定事务传播性
对应的值在TransactionDefinition中定义

传播性针对多事务，即已经存在事务的情况下出现了新的事务配置需要创建事务时，通过传播性设置来决定新旧事务如何处理。当前不存在事务时，传播性主要用于判断是否需要创建新事务！
各种情况下事务传播性如下：

传播性有哪些

REQUIRED:支持当前事务，如果不存在则创建新事务

REQUIRES_NEW:创建一个新事务，并挂起当前事务（如果存在）

NESTED:如果存在当前事务，则在嵌套事务中执行，否则和REQUIRED一样，嵌套事务指子事务不会影响父事务，但会被父事务影响，后面会详细说明

SUPPORTS:支持当前事务，如果不存在，则以非事务方式执行（REQUIRED不同点）

MANDATORY:支持当前事务，如果不存在异常则抛出异常

NOT_SUPPORTED:以非事务方式执行，如果存在事务，则挂起当前事务，挂起事务指暂时把数据库连接对象从线程同步器中移除，该连接不能进行提交及回滚操作。

NEVER:以非事务方式执行，如果存在事务，则抛出异常

为什么需要传播性

为了解决多事务情况下，新的事务应该如何处理，各种传播行为让事务应用多样化，增加的事务使用的灵活性，可以解决各种疑难杂症需求。
默认情况下事务传播性为REQUIRED，就是全程一把梭，出现问题就完蛋回滚，这种处理可以解决绝大部分事务需求。

如何实现

传播性针对事务创建，主要实现在TransactionAspectSupport.createTransactionIfNecessary->AbstractPlatformTransactionManager.getTransaction中。

主要逻辑为：
1.创建一个事务对象doGetTransaction
2.判断是否存在事务isExistingTransaction，存在按照已存在事务处理handleExistingTransaction
3.不存在事务则创建新事务startTransaction，REQUIRED、REQUIRES_NEW、NESTED配置才会真正创建事务

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
@Override
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();
 
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.
if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
throw new IllegalTransactionStateException(
"No existing transaction found for transaction marked with propagation 'mandatory'");
}
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;
}
}
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);
}
}

 

事务的核心是jdbc连接，Spring会把正在实行的事务jdbc连接放到事务同步器的线程变量中（同时只能存放一个jdbc连接）。
doGetTransaction会尝试取出这个连接并放入创建的事务对象。

1
2
3
4
5
6
7
8
protected Object doGetTransaction() {
DataSourceTransactionObject txObject = new DataSourceTransactionObject();
txObject.setSavepointAllowed(isNestedTransactionAllowed());
ConnectionHolder conHolder =
(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
txObject.setConnectionHolder(conHolder, false);
return txObject;
}

 

isExistingTransaction会对doGetTransaction创建的事务对象判断是否存在这个连接并且激活。

1
2
3
4
protected boolean isExistingTransaction(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
}

 

startTransaction会使用数据源创建一个jdbc连接放入事务对象并标记为激活。

传播性实现的核心代码在handleExistingTransaction中，可以看到：
1.TransactionDefinition.PROPAGATION_NEVER时，NEVER即不允许存在事务，存在事务直接报错
2.TransactionDefinition.PROPAGATION_NOT_SUPPORTED，NOT_SUPPORTED即不支持事务，会挂起已存在的事务
3.TransactionDefinition.PROPAGATION_REQUIRES_NEW，REQUIRES_NEW挂起已存在的事务，新起一个事务
4.TransactionDefinition.PROPAGATION_NESTED，NESTED支持嵌套事务，对当前事务创建jdbc3.0Savepoint
5.默认为TransactionDefinition.PROPAGATION_REQUIRES，使用当前事务

还剩下两个类型去哪儿了呢？
1.TransactionDefinition.PROPAGATION_MANDATORY支持当前事务，没有事务的时候会报错，所以不在handleExistingTransaction，而在不存在事务的处理代码中。
2.TransactionDefinition.PROPAGATION_SUPPORTS支持当前事务，如果不存在，则以非事务方式执行，具体在哪儿暂时没找到。。。

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
/**
* Create a TransactionStatus for an existing transaction.
*/
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 {
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);
}

 

扩展

在TransactionDefinition中，不仅列举了所有传播性，还有些别的东西，如：
ISOLATION(隔离级别)
TIMEOUT(超时时间)
READONLY(只读)
这三个都是数据库层面的东西，即Spring事务主要提供配置，然后会传递给数据库，由数据库做具体实现处理。不过Spring事务处理过程中也会使用这些配置做一些校验和判断。

这里对超时做下说明，超时不仅是数据库执行时的超时，Spring也会自检，从事务开始处理到使用Connection发送SQL执行前如果超时，Spring会自检提示超时，也就是没到数据库就超时结束了，数据库执行期间发现超时也会报错，执行完返回后，超时时间就不起作用了，即执行完成数据库操作后，线程跑的时间长短都不会引起事务超时。

事务隔离级别

事务有四大特性：原子性、隔离性、一致性和持久性。

为什么需要隔离级别

其中隔离性指的是多个事务执行互不干扰，独立执行。两个独立事务同时操作一条数据，必然出现冲突：
脏读：读到另一个事务还没有提交的数据，此数据有可能被回滚撤销掉

不可重复度：事务内多次读取同一条数据，读到的值发生变化，因为数据被另一个事务修改并提交了，不可以重复读取。

幻读：事务内执行同一数据项读取，读到的结果不一致。

数据库针对上述问题提供不同的处理解决方式，形成了隔离级别。

隔离级别有哪些

数据库操作增删改查，其中查询不涉及多事务冲突，但涉及当前事务能不能看到其他事务对数据的修改问题。多事务对同一数据的修改操作直接引发冲突，只能排队执行。所以隔离级别主要针对数据查询，即能不能读到别的事务的修改。

Spring事务中定义的隔离级别与数据库隔离级别一致：
ISOLATION_READ_UNCOMMITTED(读未提交):一个事务可以读取另一个事务并未提交的更新结果。

ISOLATION_READ_COMMITTED(读提交):一个事务的更新操作结果只有在该事务提交之后，另一个事务才可以的读取到同一笔数据更新后的结果。

ISOLATION_REPEATABLE_READ(重复读):mysql的默认级别。整个事务过程中，对同一笔数据的读取结果是相同的，不管其他事务是否在对共享数据进行更新，也不管更新提交与否。

ISOLATION_SERIALIZABLE(串行化):所有事务操作依次顺序执行。物理上严格隔离，性能差，串行化即排队一个一个来。

事务挂起与同步

事务挂起

前面提到事务挂起，何为事务挂起？
新开启的事务不希望和已有事务互相产生影响，则需要排队执行，先执行新创建的事务，把已有事务放到一边等待。
即先把当前事务的数据库连接取出来，空出线程jdbc连接，然后把新事务的jdbc连接放进去，执行完成后，再恢复，把之前取出来的再塞回去。

1
2
3
4
5
6
7
8
9
10
11
//挂起
protected Object doSuspend(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
txObject.setConnectionHolder(null);
return TransactionSynchronizationManager.unbindResource(obtainDataSource());
}
 
//恢复
protected void doResume(@Nullable Object transaction, Object suspendedResources) {
TransactionSynchronizationManager.bindResource(obtainDataSource(), suspendedResources);
}

 

这里用了一个线程变量ThreadLocal实现，key为数据源对象，value为jdbc连接
private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<>(“Transactional resources”);

挂起的作用我认为在于已存在事务（已创建多个jdbc连接）时，新创建的事务如果需要共享当前事务jdbc连接时，可以拿到正确的jdbc连接。

事务同步

挂起是事务同步的一种应用，何为事务同步？
将当前事务保存起来，新创建事务时，可以拿到，用来判断是否存在事务以及如果需要可以拿到jdbc连接。
新事务创建并启动后，使用TransactionSynchronizationManager保存事务，主要保存了以下内容：

1
2
3
4
5
6
7
8
9
10
11
12
//jdbc连接
TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
//当前是否有活动的实际事务
TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
//事务隔离级别
TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(
definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ?
definition.getIsolationLevel() : null);
//事务是否只读
TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());
//事务名
TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());

 

事务的挂起和执行完成都会清空这些，为新事务的到来做准备。

SpringBoot事务自动装配原理

SpringBootStarter中引入了AOP依赖，自动装配的配置中包含了TransactionAutoConfiguration，所以不需要单独引入依赖
在这里插入图片描述

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
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(PlatformTransactionManager.class)
@AutoConfigureAfter({ JtaAutoConfiguration.class, HibernateJpaAutoConfiguration.class,
DataSourceTransactionManagerAutoConfiguration.class, Neo4jDataAutoConfiguration.class })
@EnableConfigurationProperties(TransactionProperties.class)
public class TransactionAutoConfiguration {
...
@Configuration(proxyBeanMethods = false)
@ConditionalOnBean(TransactionManager.class)
@ConditionalOnMissingBean(AbstractTransactionManagementConfiguration.class)
public static class EnableTransactionManagementConfiguration {
 
@Configuration(proxyBeanMethods = false)
@EnableTransactionManagement(proxyTargetClass = false)
@ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "false",
matchIfMissing = false)
public static class JdkDynamicAutoProxyConfiguration {
 
}
 
@Configuration(proxyBeanMethods = false)
@EnableTransactionManagement(proxyTargetClass = true)
@ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "true",
matchIfMissing = true)
public static class CglibAutoProxyConfiguration {
 
}
 
}
}

 

存在PlatformTransactionManager时就会进行事务的自动装配，通过@EnableTransactionManagement -> @Import(TransactionManagementConfigurationSelector.class)引入事务管理配置Bean：ProxyTransactionManagementConfiguration

1.ProxyTransactionManagementConfiguration事务管理配置类包含三个Bean：
BeanFactoryTransactionAttributeSourceAdvisor
TransactionAttributeSource -> AnnotationTransactionAttributeSource
TransactionInterceptor

关系梳理:
Advisor : Spring事务顾问，包含Advice和事务元数据
Advice : 通知拦截器(切面增强处理器)，包含增强逻辑及增强逻辑切入点
AttributeSource : 事务元数据

提供事务支持的bean为Advisor，即BeanFactoryTransactionAttributeSourceAdvisor(实现了Advisor接口)，这个bean包含了Advice(TransactionInterceptor) 和 AttributeSource(TransactionAttributeSource)
入参需要的TransactionAttributeSource和TransactionInterceptor也在ProxyTransactionManagementConfiguration中定义了，实例化时会自动注入

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
@Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(
TransactionAttributeSource transactionAttributeSource, TransactionInterceptor transactionInterceptor) {
 
BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
advisor.setTransactionAttributeSource(transactionAttributeSource);
advisor.setAdvice(transactionInterceptor);
if (this.enableTx != null) {
advisor.setOrder(this.enableTx.<Integer>getNumber("order"));
}
return advisor;
}
@Bean
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public TransactionAttributeSource transactionAttributeSource() {
return new AnnotationTransactionAttributeSource();
}
 
@Bean
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public TransactionInterceptor transactionInterceptor(TransactionAttributeSource transactionAttributeSource) {
TransactionInterceptor interceptor = new TransactionInterceptor();
interceptor.setTransactionAttributeSource(transactionAttributeSource);
if (this.txManager != null) {
interceptor.setTransactionManager(this.txManager);
}
return interceptor;
}

 

2.Spring bean初始化需要执行事务aop代理增强时，需要获取增强的处理器（又称为拦截器、切面处理器，亦或者Advisor）
这时会找到事务的实际执行器TransactionInterceptor，并加以利用生成代理对象，默认为Cglib，使用CglibAopProxy代理

查找逻辑：
根据对应的aop BeanPostProcessor执行bean后置处理生成代理对象

生成代理对象是由aop的BeanPostProcessor即 AbstractAutoProxyCreator.wrapIfNecessary进行后置处理代理增强的。
AbstractAutoProxyCreator是抽象类，实际执行时必然有一个可实例化的子类
对应的实际后置处理器为：InfrastructureAdvisorAutoProxyCreator(不支持aspectJ)，这个类又是哪儿来的呢？
先找到所有实现了Advisor.class的bean，再根据当前bean匹配过滤，寻找确认bean的可用增强处理器，会忽略aspectJ代理的增强处理器。

重点来了：如何根据类型过滤出可用的Advisor。这块使用的是工具类AopUtils：
1.遍历所有的Advisor中找出beanClass可用的Advisor。
方法入口：findAdvisorsThatCanApply(candidateAdvisors, beanClass)
遍历执行匹配方法：canApply(candidate, clazz, hasIntroductions)

2.执行TransactionAttributeSource(AnnotationTransactionAttributeSource)的TransactionAnnotationParser(SpringTransactionAnnotationParser -> 自动装在TransactionAttributeSource时构造方法里创建的)

3.SpringTransactionAnnotationParser.parseTransactionAnnotation 获取注解了Transactional的方法，如果没有返回那么就结束了不需要被代理增强事务支持

1
2
3
4
5
6
7
8
9
10
public TransactionAttribute parseTransactionAnnotation(AnnotatedElement element) {
AnnotationAttributes attributes = AnnotatedElementUtils.findMergedAnnotationAttributes(
element, Transactional.class, false, false);
if (attributes != null) {
return parseTransactionAnnotation(attributes);
}
else {
return null;
}
}

 

3.含事务的bean执行拥有事务的方法时，会首先执行CglibAopProxy中的具体的intercept.intercept(如DynamicAdvisedInterceptor.intercept)方法，
这个方法作用是拦截请求，找到增强处理器，在执行原目的方法的前后执行增强代码，
事务这里大致是先获取事务，然后执行原目的方法，并进行try,catch,finally，一旦发生异常 catch中回滚事务，没有异常提交事务，不论是否发生异常，finally中清理事务

CglibAopProxy.proceed -> ReflectiveMethodInvocation.proceed -> TransactionInterceptor.invoke -> TransactionAspectSupport.invokeWithinTransaction -> 执行目标方法及事务回滚或提交

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
TransactionAspectSupport.invokeWithinTransaction(){
...
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);
}
...
commitTransactionAfterReturning(txInfo);
return retVal;
}

 

from:https://blog.csdn.net/u012098021/article/details/117810762

Copyright © 2024 Longsheng All rights reserved.