1、引言

spring的spring-tx模块提供了对事务管理支持，使用spring事务可以让我们从复杂的事务处理中得到解脱，无需要去处理获得连接、关闭连接、事务提交和回滚等这些操作。

spring事务有编程式事务和声明式事务两种实现方式。编程式事务是通过编写代码来管理事务的提交、回滚、以及事务的边界。这意味着开发者需要在代码中显式地调用事务的开始、提交和回滚。声明式事务是通过配置来管理事务，您可以使用注解或XML配置来定义事务的边界和属性，而无需显式编写事务管理的代码。

下面我们逐步分析spring源代码，理解spring事务的实现原理。

2、编程式事务

2.1 使用示例

1// transactionManager是某一个具体的PlatformTransactionManager实现类的对象
2private PlatformTransactionManager transactionManager;
3
4
5// 定义事务属性
6DefaultTransactionDefinition def = new DefaultTransactionDefinition();
7
8// 获取事务
9TransactionStatus status = transactionManager.getTransaction(def);
10
11try {
12    // 执行数据库操作
13    // ...
14    
15    // 提交事务
16    transactionManager.commit(status);
17} catch (Exception ex) {
18    // 回滚事务
19    transactionManager.rollback(status);
20}
21
22

在使用编程式事务处理的过程中，利用 DefaultTransactionDefinition 对象来持有事务处理属性。同时，在创建事务的过程中得到一个 TransactionStatus 对象，然后通过直接调用 transactionManager 对象的 commit() 和 rollback()方法来完成事务处理。

2.2 PlatformTransactionManager核心接口

PlatformTransactionManager是Spring事务管理的核心接口，通过 PlatformTransactionManager 接口设计了一系列与事务处理息息相关的接口方法，如 getTransaction()、commit()、rollback() 这些和事务处理相关的统一接口。对于这些接口的实现，很大一部分是由 AbstractTransactionManager 抽象类来完成的。

AbstractPlatformManager 封装了 Spring 事务处理中通用的处理部分，比如事务的创建、提交、回滚，事务状态和信息的处理，与线程的绑定等，有了这些通用处理的支持，对于具体的事务管理器而言，它们只需要处理和具体数据源相关的组件设置就可以了，比如在DataSourceTransactionManager中，就只需要配置好和DataSource事务处理相关的接口以及相关的设置。

2.3 事务的创建

PlatformTransactionManager的getTransaction()方法，封装了底层事务的创建，并生成一个 TransactionStatus对象。AbstractPlatformTransactionManager提供了创建事务的模板，这个模板会被具体的事务处理器所使用。从下面的代码中可以看到，AbstractPlatformTransactionManager会根据事务属性配置和当前进程绑定的事务信息，对事务是否需要创建，怎样创建进行一些通用的处理，然后把事务创建的底层工作交给具体的事务处理器完成，如：DataSourceTransactionManager、HibernateTransactionManager。

1public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
2        throws TransactionException {
3    TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
4    Object transaction = doGetTransaction();
5    boolean debugEnabled = logger.isDebugEnabled();
6    if (isExistingTransaction(transaction)) {
7        return handleExistingTransaction(def, transaction, debugEnabled);
8    }
9    if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
10        throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
11    }
12    if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
13        throw new IllegalTransactionStateException(
14                "No existing transaction found for transaction marked with propagation 'mandatory'");
15    }
16    else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
17            def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
18            def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
19        SuspendedResourcesHolder suspendedResources = suspend(null);
20        if (debugEnabled) {
21            logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
22        }
23        try {
24            return startTransaction(def, transaction, false, debugEnabled, suspendedResources);
25        }
26        catch (RuntimeException | Error ex) {
27            resume(null, suspendedResources);
28            throw ex;
29        }
30    }
31    else {
32        if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
33            logger.warn("Custom isolation level specified but no actual transaction initiated; " +
34                    "isolation level will effectively be ignored: " + def);
35        }
36        boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
37        return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
38    }
39}
40
41private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction, boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
42    boolean newSynchronization = this.getTransactionSynchronization() != SYNCHRONIZATION_NEVER;
43    DefaultTransactionStatus status = this.newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
44    this.doBegin(transaction, definition);
45    this.prepareSynchronization(status, definition);
46    return status;
47}
48
49

事务创建的结果是生成一个TransactionStatus对象，通过这个对象来保存事务处理需要的基本信息，TransactionStatus的创建过程如下：

1protected DefaultTransactionStatus newTransactionStatus(TransactionDefinition definition, @Nullable Object transaction, boolean newTransaction, boolean newSynchronization, boolean debug, @Nullable Object suspendedResources) {
2    boolean actualNewSynchronization = newSynchronization && !TransactionSynchronizationManager.isSynchronizationActive();
3    return new DefaultTransactionStatus(transaction, newTransaction, actualNewSynchronization, definition.isReadOnly(), debug, suspendedResources);
4}
5
6

以上是创建一个全新事务的过程，还有另一种情况是：在创建当前事务时，线程中已经有事务存在了。这种情况会涉及事务传播行为的处理。spring中七种事务传播行为如下：

如果检测到已存在事务，handleExistingTransaction()方法将根据不同的事务传播行为类型执行相应逻辑。

PROPAGATION_NEVER

即当前方法需要在非事务的环境下执行，如果有事务存在，那么抛出异常。

1if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
2    throw new IllegalTransactionStateException(
3            "Existing transaction found for transaction marked with propagation 'never'");
4}
5
6

PROPAGATION_NOT_SUPPORTED

与前者的区别在于，如果有事务存在，那么将事务挂起，而不是抛出异常。

1if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
2    Object suspendedResources = suspend(transaction);
3    boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
4    return prepareTransactionStatus(
5        definition, null, false, newSynchronization, debugEnabled, suspendedResources);
6}
7
8

PROPAGATION_REQUIRES_NEW

新建事务，如果当前存在事务，把当前事务挂起。

1if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
2    SuspendedResourcesHolder suspendedResources = suspend(transaction);
3    boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
4    DefaultTransactionStatus status = newTransactionStatus(
5            definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
6    doBegin(transaction, definition);
7    prepareSynchronization(status, definition);
8    return status;
9}
10
11

PROPAGATION_NESTED

开始一个 "嵌套的" 事务, 它是已经存在事务的一个真正的子事务. 嵌套事务开始执行时, 它将取得一个 savepoint. 如果这个嵌套事务失败, 我们将回滚到此 savepoint. 嵌套事务是外部事务的一部分, 只有外部事务结束后它才会被提交。

1if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
2    if (useSavepointForNestedTransaction()) {
3        DefaultTransactionStatus status = newTransactionStatus(
4                definition, transaction, false, false, true, debugEnabled, null);
5        this.transactionExecutionListeners.forEach(listener -> listener.beforeBegin(status));
6        try {
7            status.createAndHoldSavepoint();
8        }
9        catch (RuntimeException | Error ex) {
10            this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, ex));
11            throw ex;
12        }
13        this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, null));
14        return status;
15    }
16    else {
17        return startTransaction(definition, transaction, true, debugEnabled, null);
18    }
19}
20
21

2.4 事务挂起

事务挂起在AbstractTransactionManager.suspend()中处理，该方法内部将调用具体事务管理器的doSuspend()方法。以DataSourceTransactionManager为例，将ConnectionHolder设为null，因为一个ConnectionHolder对象就代表了一个数据库连接，将ConnectionHolder设为null就意味着我们下次要使用连接时，将重新从连接池获取。

1protected Object doSuspend(Object transaction) {
2    DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
3    txObject.setConnectionHolder(null);
4    return TransactionSynchronizationManager.unbindResource(obtainDataSource());
5}
6
7

unbindResource()方法最终会调用TransactionSynchronizationManager.doUnbindResource()方法，该方法将移除当前线程与事务对象的绑定。

1private static Object doUnbindResource(Object actualKey) {
2    Map<Object, Object> map = resources.get();
3    if (map == null) {
4        return null;
5    }
6    Object value = map.remove(actualKey);
7    if (map.isEmpty()) {
8        resources.remove();
9    }
10    if (value instanceof ResourceHolder resourceHolder && resourceHolder.isVoid()) {
11        value = null;
12    }
13    return value;
14}
15
16

而被挂起的事务的各种状态最终会保存在TransactionStatus对象中。

2.5 事务提交&回滚

主要是对jdbc的封装、源码逻辑较清晰，不展开细说。

3、声明式事务

其底层建立在 AOP 的基础之上，对方法前后进行拦截，然后在目标方法开始之前创建或者加入一个事务，在执行完目标方法之后根据执行情况提交或者回滚事务。通过声明式事物，无需在业务逻辑代码中掺杂事务管理的代码，只需在配置文件中做相关的事务规则声明（或通过等价的基于标注的方式），便可以将事务规则应用到业务逻辑中。

3.1 使用示例

配置：

1<bean id="dataSource" class="com.zaxxer.hikari.HikariDataSource">
2    <property name="driverClassName" value="${jdbc.driverClassName}" />
3    <property name="url" value="${jdbc.url}" />
4    <property name="username" value="${jdbc.username}" />
5    <property name="password" value="${jdbc.password}" />
6</bean>
7<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
8    <property name="dataSource" ref="dataSource"/>
9</bean>
10<tx:annotation-driven transaction-manager="transactionManager"/>
11
12

代码：

1@Transactional
2public void addOrder() {
3    // 执行数据库操作
4}
5
6

3.2 自定义标签解析

先从配置文件开始入手，找到处理annotation-driven标签的类TxNamespaceHandler。TxNamespaceHandler实现了NamespaceHandler接口，定义了如何解析和处理自定义XML标签。

1@Override
2public void init() {
3    registerBeanDefinitionParser("advice", new TxAdviceBeanDefinitionParser());
4    registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenBeanDefinitionParser());
5    registerBeanDefinitionParser("jta-transaction-manager", new JtaTransactionManagerBeanDefinitionParser());
6}
7
8

AnnotationDrivenBeanDefinitionParser里的parse()方法，对XML标签annotation-driven进行解析。

1@Override
2public BeanDefinition parse(Element element, ParserContext parserContext) {
3    registerTransactionalEventListenerFactory(parserContext);
4    String mode = element.getAttribute("mode");
5    if ("aspectj".equals(mode)) {
6        // mode="aspectj"
7        registerTransactionAspect(element, parserContext);
8        if (ClassUtils.isPresent("jakarta.transaction.Transactional", getClass().getClassLoader())) {
9            registerJtaTransactionAspect(element, parserContext);
10        }
11    }
12    else {
13        // mode="proxy"
14        AopAutoProxyConfigurer.configureAutoProxyCreator(element, parserContext);
15    }
16    return null;
17}
18
19

以默认mode配置为例，执行configureAutoProxyCreator()方法，将在Spring容器中注册了3个bean:

BeanFactoryTransactionAttributeSourceAdvisor、TransactionInterceptor、AnnotationTransactionAttributeSource。同时会将TransactionInterceptor的BeanName传入到Advisor中，然后将AnnotationTransactionAttributeSource这个Bean注入到Advisor中。之后动态代理的时候会使用这个Advisor去寻找每个Bean是否需要动态代理。

1// Create the TransactionAttributeSourceAdvisor definition.
2RootBeanDefinition advisorDef = new RootBeanDefinition(BeanFactoryTransactionAttributeSourceAdvisor.class);
3advisorDef.setSource(eleSource);
4advisorDef.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
5advisorDef.getPropertyValues().add("transactionAttributeSource", new RuntimeBeanReference(sourceName));
6advisorDef.getPropertyValues().add("adviceBeanName", interceptorName);
7if (element.hasAttribute("order")) {
8    advisorDef.getPropertyValues().add("order", element.getAttribute("order"));
9}
10parserContext.getRegistry().registerBeanDefinition(txAdvisorBeanName, advisorDef);
11
12CompositeComponentDefinition compositeDef = new CompositeComponentDefinition(element.getTagName(), eleSource);
13compositeDef.addNestedComponent(new BeanComponentDefinition(sourceDef, sourceName));
14compositeDef.addNestedComponent(new BeanComponentDefinition(interceptorDef, interceptorName));
15compositeDef.addNestedComponent(new BeanComponentDefinition(advisorDef, txAdvisorBeanName));
16parserContext.registerComponent(compositeDef);
17
18

3.3 Advisor

回顾AOP用法，Advisor可用于定义一个切面，它包含切点（Pointcut）和通知（Advice），用于在特定的连接点上执行特定的操作。spring事务实现了一个Advisor: BeanFactoryTransactionAttributeSourceAdvisor。

1public class BeanFactoryTransactionAttributeSourceAdvisor extends AbstractBeanFactoryPointcutAdvisor {
2
3    private final TransactionAttributeSourcePointcut pointcut = new TransactionAttributeSourcePointcut();
4
5    public void setTransactionAttributeSource(TransactionAttributeSource transactionAttributeSource) {
6        this.pointcut.setTransactionAttributeSource(transactionAttributeSource);
7    }
8
9    public void setClassFilter(ClassFilter classFilter) {
10        this.pointcut.setClassFilter(classFilter);
11    }
12
13    @Override
14    public Pointcut getPointcut() {
15        return this.pointcut;
16    }
17}
18
19

BeanFactoryTransactionAttributeSourceAdvisor其实是一个PointcutAdvisor，是否匹配到切入点取决于Pointcut。Pointcut的核心在于其ClassFilter和MethodMatcher。

ClassFilter:

TransactionAttributeSourcePointcut内部私有类 TransactionAttributeSourceClassFilter，实现了Spring框架中的ClassFilter接口。在matches方法中，它首先检查传入的类clazz 否为TransactionalProxy、TransactionManager或PersistenceExceptionTranslator的子类，如果不是，则获取当前的 TransactionAttributeSource 并检查其是否允许该类作为候选类。

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

MethodMatcher:

TransactionAttributeSourcePointcut.matches:

1@Override
2public boolean matches(Method method, Class<?> targetClass) {
3    return (this.transactionAttributeSource == null ||
4            this.transactionAttributeSource.getTransactionAttribute(method, targetClass) != null);
5}
6
7

getTransactionAttribute()方法最终会调用至AbstractFallbackTransactionAttributeSource.computeTransactionAttribute()方法，该方法将先去方法上查找是否有相应的事务注解(比如@Transactional)，如果没有，那么再去类上查找。

1protected TransactionAttribute computeTransactionAttribute(Method method, @Nullable Class<?> targetClass) {
2    // Don't allow non-public methods, as configured.
3    if (allowPublicMethodsOnly() && !Modifier.isPublic(method.getModifiers())) {
4        return null;
5    }
6
7    // The method may be on an interface, but we need attributes from the target class.
8    // If the target class is null, the method will be unchanged.
9    Method specificMethod = AopUtils.getMostSpecificMethod(method, targetClass);
10
11    // First try is the method in the target class.
12    TransactionAttribute txAttr = findTransactionAttribute(specificMethod);
13    if (txAttr != null) {
14        return txAttr;
15    }
16
17    // Second try is the transaction attribute on the target class.
18    txAttr = findTransactionAttribute(specificMethod.getDeclaringClass());
19    if (txAttr != null && ClassUtils.isUserLevelMethod(method)) {
20        return txAttr;
21    }
22
23    if (specificMethod != method) {
24        // Fallback is to look at the original method.
25        txAttr = findTransactionAttribute(method);
26        if (txAttr != null) {
27            return txAttr;
28        }
29        // Last fallback is the class of the original method.
30        txAttr = findTransactionAttribute(method.getDeclaringClass());
31        if (txAttr != null && ClassUtils.isUserLevelMethod(method)) {
32            return txAttr;
33        }
34    }
35
36    return null;
37}
38
39

3.4 TransactionInterceptor

TransactionInterceptor是spring事务提供的AOP拦截器，实现了AOP Alliance的MethodInterceptor接口，是一种通知（advice）。其可以用于在方法调用前后进行事务管理。

1@Override
2@Nullable
3public Object invoke(MethodInvocation invocation) throws Throwable {
4    // Work out the target class: may be {@code null}.
5    // The TransactionAttributeSource should be passed the target class
6    // as well as the method, which may be from an interface.
7    Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
8
9    // Adapt to TransactionAspectSupport's invokeWithinTransaction...
10    return invokeWithinTransaction(invocation.getMethod(), targetClass, new CoroutinesInvocationCallback() {
11        @Override
12        @Nullable
13        public Object proceedWithInvocation() throws Throwable {
14            return invocation.proceed();
15        }
16        @Override
17        public Object getTarget() {
18            return invocation.getThis();
19        }
20        @Override
21        public Object[] getArguments() {
22            return invocation.getArguments();
23        }
24    });
25}
26
27

invokeWithinTransaction()方法会根据目标方法上的事务配置，来决定是开启新事务、加入已有事务，还是直接执行逻辑（如果没有事务）。其代码简化如下(仅保留PlatformTransactionManager部分)：

1protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, final InvocationCallback invocation) {
2    // If the transaction attribute is null, the method is non-transactional.
3    final TransactionAttribute txAttr = getTransactionAttributeSource()
4        .getTransactionAttribute(method, targetClass);
5    final PlatformTransactionManager tm = determineTransactionManager(txAttr);
6    final String joinpointIdentification = methodIdentification(method, targetClass);
7    if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
8        // Standard transaction demarcation with getTransaction and commit/rollback calls.
9        TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
10        ObjectretVal = null;
11        try {
12            // This is an around advice: Invoke the next interceptor in the chain.
13            // This will normally result in a target object being invoked.
14            retVal = invocation.proceedWithInvocation();
15        } catch (Throwableex) {
16            // target invocation exception
17            completeTransactionAfterThrowing(txInfo, ex);
18            throwex;
19        } finally {
20            cleanupTransactionInfo(txInfo);
21        }
22        commitTransactionAfterReturning(txInfo);
23        returnretVal;
24    }
25}
26
27

作者：京东零售范锡军

来源：京东云开发者社区转载请注明来源

Spring事务实现原理