一、引子

  • 二、事务源码
  • 三、事务核心源码
  •  四、时序图
  • 系列目录

    spring事务详解(一)初探事务

    spring事务详解(二)简单样例

    spring事务详解(三)源码详解

    spring事务详解(四)测试验证

    spring事务详解(五)总结提高

    一、引子

    在Spring中,事务有两种实现方式:

    1. 编程式事务管理 编程式事务管理使用TransactionTemplate可实现更细粒度的事务控制。
    2. 申明式事务管理 基于Spring AOP实现。其本质是对方法前后进行拦截,然后在目标方法开始之前创建或者加入一个事务,在执行完目标方法之后根据执行情况提交或者回滚事务。

    申明式事务管理不需要入侵代码,通过@Transactional就可以进行事务操作,更快捷而且简单(尤其是配合spring boot自动配置,可以说是精简至极!),且大部分业务都可以满足,推荐使用。

    其实不管是编程式事务还是申明式事务,最终调用的底层核心代码是一致的。本章分别从编程式、申明式入手,再进入核心源码贯穿式讲解。

    二、事务源码

    2.1 编程式事务TransactionTemplate

    编程式事务,Spring已经给我们提供好了模板类TransactionTemplate,可以很方便的使用,如下图:

    TransactionTemplate全路径名是:org.springframework.transaction.support.TransactionTemplate。看包名也知道了这是spring对事务的模板类。(spring动不动就是各种Template...),看下类图先:

    一看,哟西,实现了TransactionOperations、InitializingBean这2个接口(熟悉spring源码的知道这个InitializingBean又是老套路),我们来看下接口源码如下:

    复制代码
     1 public interface TransactionOperations {
     2 
     3     /**
     4      * Execute the action specified by the given callback object within a transaction.
     5      * <p>Allows for returning a result object created within the transaction, that is,
     6      * a domain object or a collection of domain objects. A RuntimeException thrown
     7      * by the callback is treated as a fatal exception that enforces a rollback.
     8      * Such an exception gets propagated to the caller of the template.
     9      * @param action the callback object that specifies the transactional action
    10      * @return a result object returned by the callback, or {@code null} if none
    11      * @throws TransactionException in case of initialization, rollback, or system errors
    12      * @throws RuntimeException if thrown by the TransactionCallback
    13      */
    14     <T> T execute(TransactionCallback<T> action) throws TransactionException;
    15 
    16 }
    17 
    18 public interface InitializingBean {
    19 
    20     /**
    21      * Invoked by a BeanFactory after it has set all bean properties supplied
    22      * (and satisfied BeanFactoryAware and ApplicationContextAware).
    23      * <p>This method allows the bean instance to perform initialization only
    24      * possible when all bean properties have been set and to throw an
    25      * exception in the event of misconfiguration.
    26      * @throws Exception in the event of misconfiguration (such
    27      * as failure to set an essential property) or if initialization fails.
    28      */
    29     void afterPropertiesSet() throws Exception;
    30 
    31 }
    复制代码

    如上图,TransactionOperations这个接口用来执行事务的回调方法,InitializingBean这个是典型的spring bean初始化流程中(飞机票:Spring IOC(四)总结升华篇)的预留接口,专用用来在bean属性加载完毕时执行的方法。

    回到正题,TransactionTemplate的2个接口的impl方法做了什么?

    复制代码
     1     @Override
     2     public void afterPropertiesSet() {
     3         if (this.transactionManager == null) {
     4             throw new IllegalArgumentException("Property 'transactionManager' is required");
     5         }
     6     }
     7 
     8 
     9     @Override
    10     public <T> T execute(TransactionCallback<T> action) throws TransactionException {
           // 内部封装好的事务管理器 11 if (this.transactionManager instanceof CallbackPreferringPlatformTransactionManager) { 12 return ((CallbackPreferringPlatformTransactionManager) this.transactionManager).execute(this, action); 13 }// 需要手动获取事务,执行方法,提交事务的管理器 14 else {// 1.获取事务状态 15 TransactionStatus status = this.transactionManager.getTransaction(this); 16 T result; 17 try {// 2.执行业务逻辑 18 result = action.doInTransaction(status); 19 } 20 catch (RuntimeException ex) { 21 // 应用运行时异常 -> 回滚 22 rollbackOnException(status, ex); 23 throw ex; 24 } 25 catch (Error err) { 26 // Error异常 -> 回滚 27 rollbackOnException(status, err); 28 throw err; 29 } 30 catch (Throwable ex) { 31 // 未知异常 -> 回滚 32 rollbackOnException(status, ex); 33 throw new UndeclaredThrowableException(ex, "TransactionCallback threw undeclared checked exception"); 34 }// 3.事务提交 35 this.transactionManager.commit(status); 36 return result; 37 } 38 }
    复制代码

    如上图所示,实际上afterPropertiesSet只是校验了事务管理器不为空,execute()才是核心方法,execute主要步骤:

    1.getTransaction()获取事务,源码见3.3.1

    2.doInTransaction()执行业务逻辑,这里就是用户自定义的业务代码。如果是没有返回值的,就是doInTransactionWithoutResult()。

    3.commit()事务提交调用AbstractPlatformTransactionManager的commitrollbackOnException()异常回滚:调用AbstractPlatformTransactionManager的rollback()事务提交回滚,源码见3.3.3

     

    2.2 申明式事务@Transactional

    1.AOP相关概念

    申明式事务使用的是spring AOP,即面向切面编程。(什么❓你不知道什么是AOP...一句话概括就是:把业务代码中重复代码做成一个切面,提取出来,并定义哪些方法需要执行这个切面。其它的自行百度吧...)AOP核心概念如下:

    • 通知(Advice):定义了切面(各处业务代码中都需要的逻辑提炼成的一个切面)做什么what+when何时使用。例如:前置通知Before、后置通知After、返回通知After-returning、异常通知After-throwing、环绕通知Around.
    • 连接点(Joint point):程序执行过程中能够插入切面的点,一般有多个。比如调用方式时、抛出异常时。
    • 切点(Pointcut):切点定义了连接点,切点包含多个连接点,即where哪里使用通知.通常指定类+方法 或者 正则表达式来匹配 类和方法名称。
    • 切面(Aspect):切面=通知+切点,即when+where+what何时何地做什么
    • 引入(Introduction):允许我们向现有的类添加新方法或属性。
    • 织入(Weaving):织入是把切面应用到目标对象并创建新的代理对象的过程。

    2.申明式事务

    申明式事务整体调用过程,可以抽出2条线:

    1.使用代理模式,生成代理增强类。

    2.根据代理事务管理配置类,配置事务的织入,在业务方法前后进行环绕增强,增加一些事务的相关操作。例如获取事务属性、提交事务、回滚事务。

    过程如下图:

    申明式事务使用@Transactional这种注解的方式,那么我们就从springboot 容器启动时的自动配置载入(spring boot容器启动详解)开始看。在/META-INF/spring.factories中配置文件中查找,如下图:

     

    载入2个关于事务的自动配置类: 

    org.springframework.boot.autoconfigure.transaction.TransactionAutoConfiguration,
    org.springframework.boot.autoconfigure.transaction.jta.JtaAutoConfiguration,

    jta咱们就不看了,看一下TransactionAutoConfiguration这个自动配置类:

    复制代码
     1 @Configuration
     2 @ConditionalOnClass(PlatformTransactionManager.class)
     3 @AutoConfigureAfter({ JtaAutoConfiguration.class, HibernateJpaAutoConfiguration.class,
     4         DataSourceTransactionManagerAutoConfiguration.class,
     5         Neo4jDataAutoConfiguration.class })
     6 @EnableConfigurationProperties(TransactionProperties.class)
     7 public class TransactionAutoConfiguration {
     8 
     9     @Bean
    10     @ConditionalOnMissingBean
    11     public TransactionManagerCustomizers platformTransactionManagerCustomizers(
    12             ObjectProvider<List<PlatformTransactionManagerCustomizer<?>>> customizers) {
    13         return new TransactionManagerCustomizers(customizers.getIfAvailable());
    14     }
    15 
    16     @Configuration
    17     @ConditionalOnSingleCandidate(PlatformTransactionManager.class)
    18     public static class TransactionTemplateConfiguration {
    19 
    20         private final PlatformTransactionManager transactionManager;
    21 
    22         public TransactionTemplateConfiguration(
    23                 PlatformTransactionManager transactionManager) {
    24             this.transactionManager = transactionManager;
    25         }
    26 
    27         @Bean
    28         @ConditionalOnMissingBean
    29         public TransactionTemplate transactionTemplate() {
    30             return new TransactionTemplate(this.transactionManager);
    31         }
    32     }
    33 
    34     @Configuration
    35     @ConditionalOnBean(PlatformTransactionManager.class)
    36     @ConditionalOnMissingBean(AbstractTransactionManagementConfiguration.class)
    37     public static class EnableTransactionManagementConfiguration {
    38 
    39         @Configuration
    40         @EnableTransactionManagement(proxyTargetClass = false)
    41         @ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "false", matchIfMissing = false)
    42         public static class JdkDynamicAutoProxyConfiguration {
    43 
    44         }
    45 
    46         @Configuration
    47         @EnableTransactionManagement(proxyTargetClass = true)
    48         @ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "true", matchIfMissing = true)
    49         public static class CglibAutoProxyConfiguration {
    50 
    51         }
    52 
    53     }
    54 
    55 }
    复制代码

    TransactionAutoConfiguration这个类主要看:

    1.2个类注解

    @ConditionalOnClass(PlatformTransactionManager.class)即类路径下包含PlatformTransactionManager这个类时这个自动配置生效,这个类是spring事务的核心包,肯定引入了。

    @AutoConfigureAfter({ JtaAutoConfiguration.class, HibernateJpaAutoConfiguration.class, DataSourceTransactionManagerAutoConfiguration.class, Neo4jDataAutoConfiguration.class }),这个配置在括号中的4个配置类后才生效。

    2. 2个内部类

    TransactionTemplateConfiguration事务模板配置类

    @ConditionalOnSingleCandidate(PlatformTransactionManager.class)当能够唯一确定一个PlatformTransactionManager bean时才生效。

    @ConditionalOnMissingBean如果没有定义TransactionTemplate bean生成一个。

    EnableTransactionManagementConfiguration开启事务管理器配置类

    @ConditionalOnBean(PlatformTransactionManager.class)当存在PlatformTransactionManager bean时生效。

    @ConditionalOnMissingBean(AbstractTransactionManagementConfiguration.class)当没有自定义抽象事务管理器配置类时才生效。(即用户自定义抽象事务管理器配置类会优先,如果没有,就用这个默认事务管理器配置类)

    EnableTransactionManagementConfiguration支持2种代理方式:

    • 1.JdkDynamicAutoProxyConfiguration

    @EnableTransactionManagement(proxyTargetClass = false),即proxyTargetClass = false表示是JDK动态代理支持的是:面向接口代理。

    @ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "false", matchIfMissing = false),即spring.aop.proxy-target-class=false时生效,且没有这个配置不生效。

    • 2.CglibAutoProxyConfiguration

    @EnableTransactionManagement(proxyTargetClass = true),即proxyTargetClass = true标识Cglib代理支持的是子类继承代理。
    @ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "true", matchIfMissing = true),即spring.aop.proxy-target-class=true时生效,且没有这个配置默认生效。

    注意了,默认没有配置,走的Cglib代理。说明@Transactional注解支持直接加在类上。

    好吧,看了这么多配置类,终于到了@EnableTransactionManagement这个注解了。

    复制代码
     1 @Target(ElementType.TYPE)
     2 @Retention(RetentionPolicy.RUNTIME)
     3 @Documented
     4 @Import(TransactionManagementConfigurationSelector.class)
     5 public @interface EnableTransactionManagement {
     6 
     7     //proxyTargetClass = false表示是JDK动态代理支持接口代理。true表示是Cglib代理支持子类继承代理。
     8     boolean proxyTargetClass() default false;
     9 
    10     //事务通知模式(切面织入方式),默认代理模式(同一个类中方法互相调用拦截器不会生效),可以选择增强型AspectJ
    11     AdviceMode mode() default AdviceMode.PROXY;
    12 
    13     //连接点上有多个通知时,排序,默认最低。值越大优先级越低。
    14     int order() default Ordered.LOWEST_PRECEDENCE;
    15 
    16 }
    复制代码

    重点看类注解@Import(TransactionManagementConfigurationSelector.class)

    TransactionManagementConfigurationSelector类图如下:

    如上图所示,TransactionManagementConfigurationSelector继承自AdviceModeImportSelector实现了ImportSelector接口。

    复制代码
     1 public class TransactionManagementConfigurationSelector extends AdviceModeImportSelector<EnableTransactionManagement> {
     2 
     3     /**
     4      * {@inheritDoc}
     5      * @return {@link ProxyTransactionManagementConfiguration} or
     6      * {@code AspectJTransactionManagementConfiguration} for {@code PROXY} and
     7      * {@code ASPECTJ} values of {@link EnableTransactionManagement#mode()}, respectively
     8      */
     9     @Override
    10     protected String[] selectImports(AdviceMode adviceMode) {
    11         switch (adviceMode) {
    12             case PROXY:
    13                 return new String[] {AutoProxyRegistrar.class.getName(), ProxyTransactionManagementConfiguration.class.getName()};
    14             case ASPECTJ:
    15                 return new String[] {TransactionManagementConfigUtils.TRANSACTION_ASPECT_CONFIGURATION_CLASS_NAME};
    16             default:
    17                 return null;
    18         }
    19     }
    20 
    21 }
    复制代码

    如上图,最终会执行selectImports方法导入需要加载的类,我们只看proxy模式下,载入了AutoProxyRegistrar、ProxyTransactionManagementConfiguration2个类。

    • AutoProxyRegistrar:给容器中注册一个 InfrastructureAdvisorAutoProxyCreator 组件;利用后置处理器机制在对象创建以后,包装对象,返回一个代理对象(增强器),代理对象执行方法利用拦截器链进行调用;
    • ProxyTransactionManagementConfiguration:就是一个配置类,定义了事务增强器。

    AutoProxyRegistrar

    先看AutoProxyRegistrar实现了ImportBeanDefinitionRegistrar接口,复写registerBeanDefinitions方法,源码如下:

    复制代码
     1 public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
     2         boolean candidateFound = false;
     3         Set<String> annoTypes = importingClassMetadata.getAnnotationTypes();
     4         for (String annoType : annoTypes) {
     5             AnnotationAttributes candidate = AnnotationConfigUtils.attributesFor(importingClassMetadata, annoType);
     6             if (candidate == null) {
     7                 continue;
     8             }
     9             Object mode = candidate.get("mode");
    10             Object proxyTargetClass = candidate.get("proxyTargetClass");
    11             if (mode != null && proxyTargetClass != null && AdviceMode.class == mode.getClass() &&
    12                     Boolean.class == proxyTargetClass.getClass()) {
    13                 candidateFound = true;
    14                 if (mode == AdviceMode.PROXY) {//代理模式
    15                     AopConfigUtils.registerAutoProxyCreatorIfNecessary(registry);
    16                     if ((Boolean) proxyTargetClass) {//如果是CGLOB子类代理模式
    17                         AopConfigUtils.forceAutoProxyCreatorToUseClassProxying(registry);
    18                         return;
    19                     }
    20                 }
    21             }
    22         }
    23         if (!candidateFound) {
    24             String name = getClass().getSimpleName();
    25             logger.warn(String.format("%s was imported but no annotations were found " +
    26                     "having both 'mode' and 'proxyTargetClass' attributes of type " +
    27                     "AdviceMode and boolean respectively. This means that auto proxy " +
    28                     "creator registration and configuration may not have occurred as " +
    29                     "intended, and components may not be proxied as expected. Check to " +
    30                     "ensure that %s has been @Import'ed on the same class where these " +
    31                     "annotations are declared; otherwise remove the import of %s " +
    32                     "altogether.", name, name, name));
    33         }
    34     }
    复制代码

    代理模式:AopConfigUtils.registerAutoProxyCreatorIfNecessary(registry);

    最终调用的是:registerOrEscalateApcAsRequired(InfrastructureAdvisorAutoProxyCreator.class, registry, source);基础构建增强自动代理构造器

    复制代码
     1 private static BeanDefinition registerOrEscalateApcAsRequired(Class<?> cls, BeanDefinitionRegistry registry, Object source) {
     2         Assert.notNull(registry, "BeanDefinitionRegistry must not be null");
           //如果当前注册器包含internalAutoProxyCreator
    3 if (registry.containsBeanDefinition(AUTO_PROXY_CREATOR_BEAN_NAME)) {//org.springframework.aop.config.internalAutoProxyCreator内部自动代理构造器 4 BeanDefinition apcDefinition = registry.getBeanDefinition(AUTO_PROXY_CREATOR_BEAN_NAME); 5 if (!cls.getName().equals(apcDefinition.getBeanClassName())) {//如果当前类不是internalAutoProxyCreator 6 int currentPriority = findPriorityForClass(apcDefinition.getBeanClassName()); 7 int requiredPriority = findPriorityForClass(cls); 8 if (currentPriority < requiredPriority) {//如果下标大于已存在的内部自动代理构造器,index越小,优先级越高,InfrastructureAdvisorAutoProxyCreator index=0,requiredPriority最小,不进入 9 apcDefinition.setBeanClassName(cls.getName()); 10 } 11 } 12 return null;//直接返回 13 }//如果当前注册器不包含internalAutoProxyCreator,则把当前类作为根定义 14 RootBeanDefinition beanDefinition = new RootBeanDefinition(cls); 15 beanDefinition.setSource(source); 16 beanDefinition.getPropertyValues().add("order", Ordered.HIGHEST_PRECEDENCE);//优先级最高 17 beanDefinition.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); 18 registry.registerBeanDefinition(AUTO_PROXY_CREATOR_BEAN_NAME, beanDefinition); 19 return beanDefinition; 20 }
    复制代码

    如上图,APC_PRIORITY_LIST列表如下图:

    复制代码
     1 /**
     2      * Stores the auto proxy creator classes in escalation order.
     3      */
     4     private static final List<Class<?>> APC_PRIORITY_LIST = new ArrayList<Class<?>>();
     5 
     6     /**
     7      * 优先级上升list
     8      */
     9     static {
    10         APC_PRIORITY_LIST.add(InfrastructureAdvisorAutoProxyCreator.class);
    11         APC_PRIORITY_LIST.add(AspectJAwareAdvisorAutoProxyCreator.class);
    12         APC_PRIORITY_LIST.add(AnnotationAwareAspectJAutoProxyCreator.class);
    13     }
    复制代码

    如上图,由于InfrastructureAdvisorAutoProxyCreator这个类在list中第一个index=0,requiredPriority最小,不进入,所以没有重置beanClassName,啥都没做,返回null.

    那么增强代理类何时生成呢?

    InfrastructureAdvisorAutoProxyCreator类图如下:

    如上图所示,看2个核心方法:InstantiationAwareBeanPostProcessor接口的postProcessBeforeInstantiation实例化前+BeanPostProcessor接口的postProcessAfterInitialization初始化后。关于spring bean生命周期飞机票:Spring IOC(四)总结升华篇

    复制代码
     1     @Override
     2     public Object postProcessBeforeInstantiation(Class<?> beanClass, String beanName) throws BeansException {
     3         Object cacheKey = getCacheKey(beanClass, beanName);
     4 
     5         if (beanName == null || !this.targetSourcedBeans.contains(beanName)) {
     6             if (this.advisedBeans.containsKey(cacheKey)) {//如果已经存在直接返回
     7                 return null;
     8             }//是否基础构件(基础构建不需要代理):Advice、Pointcut、Advisor、AopInfrastructureBean这四类都算基础构建
     9             if (isInfrastructureClass(beanClass) || shouldSkip(beanClass, beanName)) {
    10                 this.advisedBeans.put(cacheKey, Boolean.FALSE);//添加进advisedBeans ConcurrentHashMap<k=Object,v=Boolean>标记是否需要增强实现,这里基础构建bean不需要代理,都置为false,供后面postProcessAfterInitialization实例化后使用。
    11                 return null;
    12             }
    13         }
    14 
    15         // TargetSource是spring aop预留给我们用户自定义实例化的接口,如果存在TargetSource就不会默认实例化,而是按照用户自定义的方式实例化,咱们没有定义,不进入
    18         if (beanName != null) {
    19             TargetSource targetSource = getCustomTargetSource(beanClass, beanName);
    20             if (targetSource != null) {
    21                 this.targetSourcedBeans.add(beanName);
    22                 Object[] specificInterceptors = getAdvicesAndAdvisorsForBean(beanClass, beanName, targetSource);
    23                 Object proxy = createProxy(beanClass, beanName, specificInterceptors, targetSource);
    24                 this.proxyTypes.put(cacheKey, proxy.getClass());
    25                 return proxy;
    26             }
    27         }
    28 
    29         return null;
    30     }
    复制代码

    通过追踪,由于InfrastructureAdvisorAutoProxyCreator是基础构建类,

    advisedBeans.put(cacheKey, Boolean.FALSE)

    添加进advisedBeans ConcurrentHashMap<k=Object,v=Boolean>标记是否需要增强实现,这里基础构建bean不需要代理,都置为false,供后面postProcessAfterInitialization实例化后使用。

    我们再看postProcessAfterInitialization源码如下:

    复制代码
     1     @Override
     2     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
     3         if (bean != null) {
     4             Object cacheKey = getCacheKey(bean.getClass(), beanName);
     5             if (!this.earlyProxyReferences.contains(cacheKey)) {
     6                 return wrapIfNecessary(bean, beanName, cacheKey);
     7             }
     8         }
     9         return bean;
    10     }
    11 
    12     protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
           // 如果是用户自定义获取实例,不需要增强处理,直接返回
    13 if (beanName != null && this.targetSourcedBeans.contains(beanName)) { 14 return bean; 15 }// 查询map缓存,记过false,不需要增强直接返回 16 if (Boolean.FALSE.equals(this.advisedBeans.get(cacheKey))) { 17 return bean; 18 }// 判断一遍springAOP基础构建类,标记过false,不需要增强直接返回 19 if (isInfrastructureClass(bean.getClass()) || shouldSkip(bean.getClass(), beanName)) { 20 this.advisedBeans.put(cacheKey, Boolean.FALSE); 21 return bean; 22 } 23 24 // 获取增强List<Advisor> advisors 25 Object[] specificInterceptors = getAdvicesAndAdvisorsForBean(bean.getClass(), beanName, null);
           // 如果存在增强
    26 if (specificInterceptors != DO_NOT_PROXY) { 27 this.advisedBeans.put(cacheKey, Boolean.TRUE);// 标记增强为TRUE,表示需要增强实现
             // 生成增强代理类
    28 Object proxy = createProxy( 29 bean.getClass(), beanName, specificInterceptors, new SingletonTargetSource(bean)); 30 this.proxyTypes.put(cacheKey, proxy.getClass()); 31 return proxy; 32 } 33      // 如果不存在增强,标记false,作为缓存,再次进入提高效率,第16行利用缓存先校验 34 this.advisedBeans.put(cacheKey, Boolean.FALSE); 35 return bean; 36 }
    复制代码

    下面看核心方法createProxy如下:

    复制代码
     1     protected Object createProxy(
     2             Class<?> beanClass, String beanName, Object[] specificInterceptors, TargetSource targetSource) {
     3      // 如果是ConfigurableListableBeanFactory接口(咱们DefaultListableBeanFactory就是该接口的实现类)则,暴露目标类
     4         if (this.beanFactory instanceof ConfigurableListableBeanFactory) {
             //给beanFactory->beanDefinition定义一个属性:k=AutoProxyUtils.originalTargetClass,v=需要被代理的bean class
    5 AutoProxyUtils.exposeTargetClass((ConfigurableListableBeanFactory) this.beanFactory, beanName, beanClass); 6 } 7 8 ProxyFactory proxyFactory = new ProxyFactory(); 9 proxyFactory.copyFrom(this); 10      //如果不是代理目标类 11 if (!proxyFactory.isProxyTargetClass()) {//如果beanFactory定义了代理目标类(CGLIB) 12 if (shouldProxyTargetClass(beanClass, beanName)) { 13 proxyFactory.setProxyTargetClass(true);//代理工厂设置代理目标类 14 } 15 else {//否则设置代理接口(JDK) 16 evaluateProxyInterfaces(beanClass, proxyFactory); 17 } 18 } 19      //把拦截器包装成增强(通知) 20 Advisor[] advisors = buildAdvisors(beanName, specificInterceptors); 21 proxyFactory.addAdvisors(advisors);//设置进代理工厂 22 proxyFactory.setTargetSource(targetSource); 23 customizeProxyFactory(proxyFactory);//空方法,留给子类拓展用,典型的spring的风格,喜欢处处留后路 24      //用于控制代理工厂是否还允许再次添加通知,默认为false(表示不允许) 25 proxyFactory.setFrozen(this.freezeProxy); 26 if (advisorsPreFiltered()) {//默认false,上面已经前置过滤了匹配的增强Advisor 27 proxyFactory.setPreFiltered(true); 28 } 29 //代理工厂获取代理对象的核心方法 30 return proxyFactory.getProxy(getProxyClassLoader()); 31 }
    复制代码

    最终我们生成的是CGLIB代理类.到此为止我们分析完了代理类的构造过程。

    ProxyTransactionManagementConfiguration

    下面来看ProxyTransactionManagementConfiguration

    复制代码
     1 @Configuration
     2 public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {
     3 
     4     @Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
     5     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)//定义事务增强器
     6     public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor() {
     7         BeanFactoryTransactionAttributeSourceAdvisor j = new BeanFactoryTransactionAttributeSourceAdvisor();
     8         advisor.setTransactionAttributeSource(transactionAttributeSource());
     9         advisor.setAdvice(transactionInterceptor());
    10         advisor.setOrder(this.enableTx.<Integer>getNumber("order"));
    11         return advisor;
    12     }
    13 
    14     @Bean
    15     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)//定义基于注解的事务属性资源
    16     public TransactionAttributeSource transactionAttributeSource() {
    17         return new AnnotationTransactionAttributeSource();
    18     }
    19 
    20     @Bean
    21     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)//定义事务拦截器
    22     public TransactionInterceptor transactionInterceptor() {
    23         TransactionInterceptor interceptor = new TransactionInterceptor();
    24         interceptor.setTransactionAttributeSource(transactionAttributeSource());
    25         if (this.txManager != null) {
    26             interceptor.setTransactionManager(this.txManager);
    27         }
    28         return interceptor;
    29     }
    30 
    31 }
    复制代码

    核心方法:transactionAdvisor()事务织入

    定义了一个advisor,设置事务属性、设置事务拦截器TransactionInterceptor、设置顺序。核心就是事务拦截器TransactionInterceptor

    TransactionInterceptor使用通用的spring事务基础架构实现“声明式事务”,继承自TransactionAspectSupport类(该类包含与Spring的底层事务API的集成),实现了MethodInterceptor接口。spring类图如下:

    事务拦截器的拦截功能就是依靠实现了MethodInterceptor接口,熟悉spring的同学肯定很熟悉MethodInterceptor了,这个是spring的方法拦截器,主要看invoke方法:

    复制代码
     1 @Override
     2     public Object invoke(final MethodInvocation invocation) throws Throwable {
     3         // Work out the target class: may be {@code null}.
     4         // The TransactionAttributeSource should be passed the target class
     5         // as well as the method, which may be from an interface.
     6         Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
     7 
     8         // 调用TransactionAspectSupport的 invokeWithinTransaction方法
     9         return invokeWithinTransaction(invocation.getMethod(), targetClass, new InvocationCallback() {
    10             @Override
    11             public Object proceedWithInvocation() throws Throwable {
    12                 return invocation.proceed();
    13             }
    14         });
    15     }
    复制代码

    如上图TransactionInterceptor复写MethodInterceptor接口的invoke方法,并在invoke方法中调用了父类TransactionAspectSupportinvokeWithinTransaction()方法,源码如下:

    复制代码
     1 protected Object invokeWithinTransaction(Method method, Class<?> targetClass, final InvocationCallback invocation)
     2             throws Throwable {  
     3 
     4         // 如果transaction attribute为空,该方法就是非事务(非编程式事务)
     5         final TransactionAttribute txAttr = getTransactionAttributeSource().getTransactionAttribute(method, targetClass);
     6         final PlatformTransactionManager tm = determineTransactionManager(txAttr);
     7         final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
     8      // 标准声明式事务:如果事务属性为空 或者 非回调偏向的事务管理器
     9         if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
    10             // Standard transaction demarcation with getTransaction and commit/rollback calls.
    11             TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
    12             Object retVal = null;
    13             try {
    14                 // 这里就是一个环绕增强,在这个proceed前后可以自己定义增强实现
    15                 // 方法执行
    16                 retVal = invocation.proceedWithInvocation();
    17             }
    18             catch (Throwable ex) {
    19                 // 根据事务定义的,该异常需要回滚就回滚,否则提交事务
    20                 completeTransactionAfterThrowing(txInfo, ex);
    21                 throw ex;
    22             }
    23             finally {//清空当前事务信息,重置为老的
    24                 cleanupTransactionInfo(txInfo);
    25             }//返回结果之前提交事务
    26             commitTransactionAfterReturning(txInfo);
    27             return retVal;
    28         }
    29      // 编程式事务:(回调偏向)
    30         else {
    31             final ThrowableHolder throwableHolder = new ThrowableHolder();
    32 
    33             // It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
    34             try {
    35                 Object result = ((CallbackPreferringPlatformTransactionManager) tm).execute(txAttr,
    36                         new TransactionCallback<Object>() {
    37                             @Override
    38                             public Object doInTransaction(TransactionStatus status) {
    39                                 TransactionInfo txInfo = prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
    40                                 try {
    41                                     return invocation.proceedWithInvocation();
    42                                 }
    43                                 catch (Throwable ex) {// 如果该异常需要回滚
    44                                     if (txAttr.rollbackOn(ex)) {
    45                                         // 如果是运行时异常返回
    46                                         if (ex instanceof RuntimeException) {
    47                                             throw (RuntimeException) ex;
    48                                         }// 如果是其它异常都抛ThrowableHolderException
    49                                         else {
    50                                             throw new ThrowableHolderException(ex);
    51                                         }
    52                                     }// 如果不需要回滚
    53                                     else {
    54                                         // 定义异常,最终就直接提交事务了
    55                                         throwableHolder.throwable = ex;
    56                                         return null;
    57                                     }
    58                                 }
    59                                 finally {//清空当前事务信息,重置为老的
    60                                     cleanupTransactionInfo(txInfo);
    61                                 }
    62                             }
    63                         });
    64 
    65                 // 上抛异常
    66                 if (throwableHolder.throwable != null) {
    67                     throw throwableHolder.throwable;
    68                 }
    69                 return result;
    70             }
    71             catch (ThrowableHolderException ex) {
    72                 throw ex.getCause();
    73             }
    74             catch (TransactionSystemException ex2) {
    75                 if (throwableHolder.throwable != null) {
    76                     logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
    77                     ex2.initApplicationException(throwableHolder.throwable);
    78                 }
    79                 throw ex2;
    80             }
    81             catch (Throwable ex2) {
    82                 if (throwableHolder.throwable != null) {
    83                     logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
    84                 }
    85                 throw ex2;
    86             }
    87         }
    88     }
    复制代码

     如上图,我们主要看第一个分支,申明式事务,核心流程如下:

    1.createTransactionIfNecessary():如果有必要,创建事务

    2.InvocationCallbackproceedWithInvocation():InvocationCallback是父类的内部回调接口,子类中实现该接口供父类调用,子类TransactionInterceptorinvocation.proceed()。回调方法执行

    3.异常回滚completeTransactionAfterThrowing()

    1.createTransactionIfNecessary():

    复制代码
     1 protected TransactionInfo createTransactionIfNecessary(
     2             PlatformTransactionManager tm, TransactionAttribute txAttr, final String joinpointIdentification) {
     3 
     4         // 如果还没有定义名字,把连接点的ID定义成事务的名称
     5         if (txAttr != null && txAttr.getName() == null) {
     6             txAttr = new DelegatingTransactionAttribute(txAttr) {
     7                 @Override
     8                 public String getName() {
     9                     return joinpointIdentification;
    10                 }
    11             };
    12         }
    13 
    14         TransactionStatus status = null;
    15         if (txAttr != null) {
    16             if (tm != null) {
    17                 status = tm.getTransaction(txAttr);
    18             }
    19             else {
    20                 if (logger.isDebugEnabled()) {
    21                     logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
    22                             "] because no transaction manager has been configured");
    23                 }
    24             }
    25         }
    26         return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
    27     }
    复制代码

    核心就是:

    1)getTransaction(),根据事务属性获取事务TransactionStatus,大道归一,都是调用PlatformTransactionManager.getTransaction(),源码见3.3.1。
    2)prepareTransactionInfo(),构造一个TransactionInfo事务信息对象,绑定当前线程:ThreadLocal<TransactionInfo>。

    2.invocation.proceed()回调业务方法:

    最终实现类是ReflectiveMethodInvocation,类图如下:

    如上图,ReflectiveMethodInvocation类实现了ProxyMethodInvocation接口,但是ProxyMethodInvocation继承了3层接口...ProxyMethodInvocation->MethodInvocation->Invocation->Joinpoint

    Joinpoint:连接点接口,定义了执行接口:Object proceed() throws Throwable; 执行当前连接点,并跳到拦截器链上的下一个拦截器。

    Invocation:调用接口,继承自Joinpoint,定义了获取参数接口: Object[] getArguments();是一个带参数的、可被拦截器拦截的连接点。

    MethodInvocation:方法调用接口,继承自Invocation,定义了获取方法接口:Method getMethod(); 是一个带参数的可被拦截的连接点方法。

    ProxyMethodInvocation:代理方法调用接口,继承自MethodInvocation,定义了获取代理对象接口:Object getProxy();是一个由代理类执行的方法调用连接点方法。

    ReflectiveMethodInvocation:实现了ProxyMethodInvocation接口,自然就实现了父类接口的的所有接口。获取代理类,获取方法,获取参数,用代理类执行这个方法并且自动跳到下一个连接点。

    下面看一下proceed方法源码:

    复制代码
     1 @Override
     2     public Object proceed() throws Throwable {
     3         //    启动时索引为-1,唤醒连接点,后续递增
     4         if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {
     5             return invokeJoinpoint();
     6         }
     7 
     8         Object interceptorOrInterceptionAdvice =
     9                 this.interceptorsAndDynamicMethodMatchers.get(++this.currentInterceptorIndex);
    10         if (interceptorOrInterceptionAdvice instanceof InterceptorAndDynamicMethodMatcher) {
    11             // 这里进行动态方法匹配校验,静态的方法匹配早已经校验过了(MethodMatcher接口有两种典型:动态/静态校验)
    13             InterceptorAndDynamicMethodMatcher dm =
    14                     (InterceptorAndDynamicMethodMatcher) interceptorOrInterceptionAdvice;
    15             if (dm.methodMatcher.matches(this.method, this.targetClass, this.arguments)) {
    16                 return dm.interceptor.invoke(this);
    17             }
    18             else {
    19                 // 动态匹配失败,跳过当前拦截,进入下一个(拦截器链)
    21                 return proceed();
    22             }
    23         }
    24         else {
    25             // 它是一个拦截器,所以我们只调用它:在构造这个对象之前,切入点将被静态地计算。
    27             return ((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);
    28         }
    29     }
    复制代码

    咱们这里最终调用的是((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);就是TransactionInterceptor事务拦截器回调 目标业务方法(addUserBalanceAndUser)。

    3.completeTransactionAfterThrowing()

    最终调用AbstractPlatformTransactionManager的rollback(),提交事务commitTransactionAfterReturning()最终调用AbstractPlatformTransactionManager的commit(),源码见3.3.3

    总结

    可见不管是编程式事务,还是声明式事务,最终源码都是调用事务管理器的PlatformTransactionManager接口的3个方法:

    1. getTransaction
    2. commit
    3. rollback

    下一节我们就来看看这个事务管理如何实现这3个方法。

    三、事务核心源码

    咱们看一下核心类图:

    如上提所示,PlatformTransactionManager顶级接口定义了最核心的事务管理方法,下面一层是AbstractPlatformTransactionManager抽象类,实现了PlatformTransactionManager接口的方法并定义了一些抽象方法,供子类拓展。最后下面一层是2个经典事务管理器:

    1.DataSourceTransactionmanager,即JDBC单数据库事务管理器,基于Connection实现,

    2.JtaTransactionManager,即多数据库事务管理器(又叫做分布式事务管理器),其实现了JTA规范,使用XA协议进行两阶段提交。

    我们这里只看基于JDBC connection的DataSourceTransactionmanager源码。

    PlatformTransactionManager接口:

    复制代码
    1 public interface PlatformTransactionManager {
    2     // 获取事务状态
    3     TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException;
    4   // 事务提交
    5     void commit(TransactionStatus status) throws TransactionException;
    6   // 事务回滚
    7     void rollback(TransactionStatus status) throws TransactionException;
    8 }
    复制代码

    3.1 getTransaction获取事务

    AbstractPlatformTransactionManager实现了getTransaction()方法如下:

    复制代码
     1     @Override
     2     public final TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {
     3         Object transaction = doGetTransaction();
     4 
     5         // Cache debug flag to avoid repeated checks.
     6         boolean debugEnabled = logger.isDebugEnabled();
     7 
     8         if (definition == null) {
     9             // Use defaults if no transaction definition given.
    10             definition = new DefaultTransactionDefinition();
    11         }
    12       // 如果当前已经存在事务
    13         if (isExistingTransaction(transaction)) {
    14             // 根据不同传播机制不同处理
    15             return handleExistingTransaction(definition, transaction, debugEnabled);
    16         }
    17 
    18         // 超时不能小于默认值
    19         if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
    20             throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
    21         }
    22 
    23         // 当前不存在事务,传播机制=MANDATORY(支持当前事务,没事务报错),报错
    24         if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
    25             throw new IllegalTransactionStateException(
    26                     "No existing transaction found for transaction marked with propagation 'mandatory'");
    27         }// 当前不存在事务,传播机制=REQUIRED/REQUIRED_NEW/NESTED,这三种情况,需要新开启事务,且加上事务同步
    28         else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
    29                 definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
    30                 definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
    31             SuspendedResourcesHolder suspendedResources = suspend(null);
    32             if (debugEnabled) {
    33                 logger.debug("Creating new transaction with name [" + definition.getName() + "]: " + definition);
    34             }
    35             try {// 是否需要新开启同步// 开启// 开启
    36                 boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    37                 DefaultTransactionStatus status = newTransactionStatus(
    38                         definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
    39                 doBegin(transaction, definition);// 开启新事务
    40                 prepareSynchronization(status, definition);//预备同步
    41                 return status;
    42             }
    43             catch (RuntimeException ex) {
    44                 resume(null, suspendedResources);
    45                 throw ex;
    46             }
    47             catch (Error err) {
    48                 resume(null, suspendedResources);
    49                 throw err;
    50             }
    51         }
    52         else {
    53             // 当前不存在事务当前不存在事务,且传播机制=PROPAGATION_SUPPORTS/PROPAGATION_NOT_SUPPORTED/PROPAGATION_NEVER,这三种情况,创建“空”事务:没有实际事务,但可能是同步。警告:定义了隔离级别,但并没有真实的事务初始化,隔离级别被忽略有隔离级别但是并没有定义实际的事务初始化,有隔离级别但是并没有定义实际的事务初始化,
    54             if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
    55                 logger.warn("Custom isolation level specified but no actual transaction initiated; " +
    56                         "isolation level will effectively be ignored: " + definition);
    57             }
    58             boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
    59             return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null);
    60         }
    61     }
    复制代码

     如上图,源码分成了2条处理线,

    1.当前已存在事务:isExistingTransaction()判断是否存在事务,存在事务handleExistingTransaction()根据不同传播机制不同处理

    2.当前不存在事务: 不同传播机制不同处理

    handleExistingTransaction()源码如下:

    复制代码
     1 private TransactionStatus handleExistingTransaction(
     2             TransactionDefinition definition, Object transaction, boolean debugEnabled)
     3             throws TransactionException {
     4      // 1.NERVER(不支持当前事务;如果当前事务存在,抛出异常)报错
     5         if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
     6             throw new IllegalTransactionStateException(
     7                     "Existing transaction found for transaction marked with propagation 'never'");
     8         }
     9       // 2.NOT_SUPPORTED(不支持当前事务,现有同步将被挂起)挂起当前事务
    10         if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
    11             if (debugEnabled) {
    12                 logger.debug("Suspending current transaction");
    13             }
    14             Object suspendedResources = suspend(transaction);
    15             boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
    16             return prepareTransactionStatus(
    17                     definition, null, false, newSynchronization, debugEnabled, suspendedResources);
    18         }
    19       // 3.REQUIRES_NEW挂起当前事务,创建新事务
    20         if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
    21             if (debugEnabled) {
    22                 logger.debug("Suspending current transaction, creating new transaction with name [" +
    23                         definition.getName() + "]");
    24             }// 挂起当前事务
    25             SuspendedResourcesHolder suspendedResources = suspend(transaction);
    26             try {// 创建新事务
    27                 boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    28                 DefaultTransactionStatus status = newTransactionStatus(
    29                         definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
    30                 doBegin(transaction, definition);
    31                 prepareSynchronization(status, definition);
    32                 return status;
    33             }
    34             catch (RuntimeException beginEx) {
    35                 resumeAfterBeginException(transaction, suspendedResources, beginEx);
    36                 throw beginEx;
    37             }
    38             catch (Error beginErr) {
    39                 resumeAfterBeginException(transaction, suspendedResources, beginErr);
    40                 throw beginErr;
    41             }
    42         }
    43      // 4.NESTED嵌套事务
    44         if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
    45             if (!isNestedTransactionAllowed()) {
    46                 throw new NestedTransactionNotSupportedException(
    47                         "Transaction manager does not allow nested transactions by default - " +
    48                         "specify 'nestedTransactionAllowed' property with value 'true'");
    49             }
    50             if (debugEnabled) {
    51                 logger.debug("Creating nested transaction with name [" + definition.getName() + "]");
    52             }// 是否支持保存点:非JTA事务走这个分支。AbstractPlatformTransactionManager默认是true,JtaTransactionManager复写了该方法false,DataSourceTransactionmanager没有复写,还是true,
    53             if (useSavepointForNestedTransaction()) { 
    54                 // Usually uses JDBC 3.0 savepoints. Never activates Spring synchronization.
    55                 DefaultTransactionStatus status =
    56                         prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
    57                 status.createAndHoldSavepoint();// 创建保存点
    58                 return status;
    59             }
    60             else {
    61                 // JTA事务走这个分支,创建新事务
    62                 boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    63                 DefaultTransactionStatus status = newTransactionStatus(
    64                         definition, transaction, true, newSynchronization, debugEnabled, null);
    65                 doBegin(transaction, definition);
    66                 prepareSynchronization(status, definition);
    67                 return status;
    68             }
    69         }
    70 
    71         
    72         if (debugEnabled) {
    73             logger.debug("Participating in existing transaction");
    74         }
    75         if (isValidateExistingTransaction()) {
    76             if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
    77                 Integer currentIsolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
    78                 if (currentIsolationLevel == null || currentIsolationLevel != definition.getIsolationLevel()) {
    79                     Constants isoConstants = DefaultTransactionDefinition.constants;
    80                     throw new IllegalTransactionStateException("Participating transaction with definition [" +
    81                             definition + "] specifies isolation level which is incompatible with existing transaction: " +
    82                             (currentIsolationLevel != null ?
    83                                     isoConstants.toCode(currentIsolationLevel, DefaultTransactionDefinition.PREFIX_ISOLATION) :
    84                                     "(unknown)"));
    85                 }
    86             }
    87             if (!definition.isReadOnly()) {
    88                 if (TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
    89                     throw new IllegalTransactionStateException("Participating transaction with definition [" +
    90                             definition + "] is not marked as read-only but existing transaction is");
    91                 }
    92             }
    93         }// 到这里PROPAGATION_SUPPORTS 或 PROPAGATION_REQUIRED或PROPAGATION_MANDATORY,存在事务加入事务即可,prepareTransactionStatus第三个参数就是是否需要新事务。false代表不需要新事物
    94         boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    95         return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
    96     }
    复制代码

    如上图,当前线程已存在事务情况下,新的不同隔离级别处理情况:

    1.NERVER:不支持当前事务;如果当前事务存在,抛出异常:"Existing transaction found for transaction marked with propagation 'never'"
    2.NOT_SUPPORTED:不支持当前事务,现有同步将被挂起:suspend()
    3.REQUIRES_NEW挂起当前事务,创建新事务:

      1)suspend()

      2)doBegin()
    4.NESTED嵌套事务

      1)非JTA事务:createAndHoldSavepoint()创建JDBC3.0保存点,不需要同步

      2) JTA事务:开启新事务,doBegin()+prepareSynchronization()需要同步

     这里有几个核心方法:挂起当前事务suspend()、开启新事务doBegin()。

    suspend()源码如下:

    复制代码
     1 protected final SuspendedResourcesHolder suspend(Object transaction) throws TransactionException {
     2         if (TransactionSynchronizationManager.isSynchronizationActive()) {// 1.当前存在同步,
     3             List<TransactionSynchronization> suspendedSynchronizations = doSuspendSynchronization();
     4             try {
     5                 Object suspendedResources = null;
     6                 if (transaction != null) {// 事务不为空,挂起事务
     7                     suspendedResources = doSuspend(transaction);
     8                 }// 解除绑定当前事务各种属性:名称、只读、隔离级别、是否是真实的事务.
     9                 String name = TransactionSynchronizationManager.getCurrentTransactionName();
    10                 TransactionSynchronizationManager.setCurrentTransactionName(null);
    11                 boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();
    12                 TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);
    13                 Integer isolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
    14                 TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(null);
    15                 boolean wasActive = TransactionSynchronizationManager.isActualTransactionActive();
    16                 TransactionSynchronizationManager.setActualTransactionActive(false);
    17                 return new SuspendedResourcesHolder(
    18                         suspendedResources, suspendedSynchronizations, name, readOnly, isolationLevel, wasActive);
    19             }
    20             catch (RuntimeException ex) {
    21                 // doSuspend failed - original transaction is still active...
    22                 doResumeSynchronization(suspendedSynchronizations);
    23                 throw ex;
    24             }
    25             catch (Error err) {
    26                 // doSuspend failed - original transaction is still active...
    27                 doResumeSynchronization(suspendedSynchronizations);
    28                 throw err;
    29             }
    30         }// 2.没有同步但,事务不为空,挂起事务
    31         else if (transaction != null) {
    32             // Transaction active but no synchronization active.
    33             Object suspendedResources = doSuspend(transaction);
    34             return new SuspendedResourcesHolder(suspendedResources);
    35         }// 2.没有同步但,事务为空,什么都不用做
    36         else {
    37             // Neither transaction nor synchronization active.
    38             return null;
    39         }
    40     }
    复制代码

    doSuspend(),挂起事务,AbstractPlatformTransactionManager抽象类doSuspend()会报错:不支持挂起,如果具体事务执行器支持就复写doSuspend(),DataSourceTransactionManager实现如下:

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

    挂起DataSourceTransactionManager事务的核心操作就是:

    1.把当前事务的connectionHolder数据库连接持有者清空。

    2.当前线程解绑datasource.其实就是ThreadLocal移除对应变量(TransactionSynchronizationManager类中定义的private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<Map<Object, Object>>("Transactional resources");

    TransactionSynchronizationManager事务同步管理器,该类维护了多个线程本地变量ThreadLocal,如下图:

    复制代码
     1 public abstract class TransactionSynchronizationManager {
     2 
     3     private static final Log logger = LogFactory.getLog(TransactionSynchronizationManager.class);
     4     // 事务资源:map<k,v> 两种数据对。1.会话工厂和会话k=SqlsessionFactory v=SqlSessionHolder 2.数据源和连接k=DataSource v=ConnectionHolder
     5     private static final ThreadLocal<Map<Object, Object>> resources =
     6             new NamedThreadLocal<Map<Object, Object>>("Transactional resources");
     7     // 事务同步
     8     private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =
     9             new NamedThreadLocal<Set<TransactionSynchronization>>("Transaction synchronizations");
    10   // 当前事务名称
    11     private static final ThreadLocal<String> currentTransactionName =
    12             new NamedThreadLocal<String>("Current transaction name");
    13   // 当前事务的只读属性
    14     private static final ThreadLocal<Boolean> currentTransactionReadOnly =
    15             new NamedThreadLocal<Boolean>("Current transaction read-only status");
    16   // 当前事务的隔离级别
    17     private static final ThreadLocal<Integer> currentTransactionIsolationLevel =
    18             new NamedThreadLocal<Integer>("Current transaction isolation level");
    19   // 是否存在事务
    20     private static final ThreadLocal<Boolean> actualTransactionActive =
    21             new NamedThreadLocal<Boolean>("Actual transaction active");
    22 。。。
    23 }
    复制代码

    doBegin()源码如下:

    复制代码
     1 @Override
     2     protected void doBegin(Object transaction, TransactionDefinition definition) {
     3         DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
     4         Connection con = null;
     5 
     6         try {// 如果事务还没有connection或者connection在事务同步状态,重置新的connectionHolder
     7             if (!txObject.hasConnectionHolder() ||
     8                     txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
     9                 Connection newCon = this.dataSource.getConnection();
    10                 if (logger.isDebugEnabled()) {
    11                     logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
    12                 }// 重置新的connectionHolder
    13                 txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
    14             }
    15        //设置新的连接为事务同步中
    16             txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
    17             con = txObject.getConnectionHolder().getConnection();
    18          //conn设置事务隔离级别,只读
    19             Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
    20             txObject.setPreviousIsolationLevel(previousIsolationLevel);//DataSourceTransactionObject设置事务隔离级别
    21 
    22             // 如果是自动提交切换到手动提交
    23             // so we don't want to do it unnecessarily (for example if we've explicitly
    24             // configured the connection pool to set it already).
    25             if (con.getAutoCommit()) {
    26                 txObject.setMustRestoreAutoCommit(true);
    27                 if (logger.isDebugEnabled()) {
    28                     logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
    29                 }
    30                 con.setAutoCommit(false);
    31             }
    32        // 如果只读,执行sql设置事务只读
    33             prepareTransactionalConnection(con, definition);
    34             txObject.getConnectionHolder().setTransactionActive(true);// 设置connection持有者的事务开启状态
    35 
    36             int timeout = determineTimeout(definition);
    37             if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
    38                 txObject.getConnectionHolder().setTimeoutInSeconds(timeout);// 设置超时秒数
    39             }
    40 
    41             // 绑定connection持有者到当前线程
    42             if (txObject.isNewConnectionHolder()) {
    43                 TransactionSynchronizationManager.bindResource(getDataSource(), txObject.getConnectionHolder());
    44             }
    45         }
    46 
    47         catch (Throwable ex) {
    48             if (txObject.isNewConnectionHolder()) {
    49                 DataSourceUtils.releaseConnection(con, this.dataSource);
    50                 txObject.setConnectionHolder(null, false);
    51             }
    52             throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
    53         }
    54     }
    复制代码

    如上图,开启新事务的准备工作doBegin()的核心操作就是:

    1.DataSourceTransactionObject“数据源事务对象”,设置ConnectionHolder,再给ConnectionHolder设置各种属性:自动提交、超时、事务开启、隔离级别。

    2.给当前线程绑定一个线程本地变量,key=DataSource数据源  v=ConnectionHolder数据库连接。

     

    3.2 commit提交事务

    3.2.1 资源管理类

    SqlSessionSynchronization是SqlSessionUtils的一个内部类,继承自TransactionSynchronizationAdapter抽象类,实现了事务同步接口TransactionSynchronization

    类图如下:

    TransactionSynchronization接口定义了事务操作时的对应资源的(JDBC事务那么就是SqlSessionSynchronization)管理方法:

    复制代码
     1     // 挂起事务   
    2   void suspend(); 3 // 唤醒事务   
    4
       void resume();
    5 6 void flush(); 7 8 // 提交事务前 9 void beforeCommit(boolean readOnly); 10 11 // 提交事务完成前 12 void beforeCompletion(); 13 14 // 提交事务后 15 void afterCommit(); 16 17 // 提交事务完成后 18 void afterCompletion(int status);
    复制代码

    后续很多都是使用这些接口管理事务。

     

    3.2.2 commit提交事务

    AbstractPlatformTransactionManager的commit源码如下:

    复制代码
     1 @Override
     2     public final void commit(TransactionStatus status) throws TransactionException {
     3         if (status.isCompleted()) {// 如果事务已完结,报错无法再次提交
     4             throw new IllegalTransactionStateException(
     5                     "Transaction is already completed - do not call commit or rollback more than once per transaction");
     6         }
     7 
     8         DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
     9         if (defStatus.isLocalRollbackOnly()) {// 如果事务明确标记为回滚,
    10             if (defStatus.isDebug()) {
    11                 logger.debug("Transactional code has requested rollback");
    12             }
    13             processRollback(defStatus);//执行回滚
    14             return;
    15         }//如果不需要全局回滚时提交 且 全局回滚
    16         if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
    17             if (defStatus.isDebug()) {
    18                 logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");
    19             }//执行回滚
    20             processRollback(defStatus);
    21             // 仅在最外层事务边界(新事务)或显式地请求时抛出“未期望的回滚异常”
    23             if (status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()) {
    24                 throw new UnexpectedRollbackException(
    25                         "Transaction rolled back because it has been marked as rollback-only");
    26             }
    27             return;
    28         }
    29      // 执行提交事务
    30         processCommit(defStatus);
    31     }
    复制代码

    如上图,各种判断:

    • 1.如果事务明确标记为本地回滚,-》执行回滚
    • 2.如果不需要全局回滚时提交 且 全局回滚-》执行回滚
    • 3.提交事务,核心方法processCommit()

    processCommit如下:

    复制代码
     1 private void processCommit(DefaultTransactionStatus status) throws TransactionException {
     2         try {
     3             boolean beforeCompletionInvoked = false;
     4             try {//3个前置操作
     5                 prepareForCommit(status);
     6                 triggerBeforeCommit(status);
     7                 triggerBeforeCompletion(status);
     8                 beforeCompletionInvoked = true;//3个前置操作已调用
     9                 boolean globalRollbackOnly = false;//新事务 或 全局回滚失败
    10                 if (status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()) {
    11                     globalRollbackOnly = status.isGlobalRollbackOnly();
    12                 }//1.有保存点,即嵌套事务
    13                 if (status.hasSavepoint()) {
    14                     if (status.isDebug()) {
    15                         logger.debug("Releasing transaction savepoint");
    16                     }//释放保存点
    17                     status.releaseHeldSavepoint();
    18                 }//2.新事务
    19                 else if (status.isNewTransaction()) {
    20                     if (status.isDebug()) {
    21                         logger.debug("Initiating transaction commit");
    22                     }//调用事务处理器提交事务
    23                     doCommit(status);
    24                 }
    25                 // 3.非新事务,且全局回滚失败,但是提交时没有得到异常,抛出异常
    27                 if (globalRollbackOnly) {
    28                     throw new UnexpectedRollbackException(
    29                             "Transaction silently rolled back because it has been marked as rollback-only");
    30                 }
    31             }
    32             catch (UnexpectedRollbackException ex) {
    33                 // 触发完成后事务同步,状态为回滚
    34                 triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
    35                 throw ex;
    36             }// 事务异常
    37             catch (TransactionException ex) {
    38                 // 提交失败回滚
    39                 if (isRollbackOnCommitFailure()) {
    40                     doRollbackOnCommitException(status, ex);
    41                 }// 触发完成后回调,事务同步状态为未知
    42                 else {
    43                     triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
    44                 }
    45                 throw ex;
    46             }// 运行时异常
    47             catch (RuntimeException ex) {
                // 如果3个前置步骤未完成,调用前置的最后一步操作
    48 if (!beforeCompletionInvoked) { 49 triggerBeforeCompletion(status); 50 }// 提交异常回滚 51 doRollbackOnCommitException(status, ex); 52 throw ex; 53 }// 其它异常 54 catch (Error err) {  
                // 如果3个前置步骤未完成,调用前置的最后一步操作
    55 if (!beforeCompletionInvoked) { 56 triggerBeforeCompletion(status); 57 }// 提交异常回滚 58 doRollbackOnCommitException(status, err); 59 throw err; 60 } 61 62 // Trigger afterCommit callbacks, with an exception thrown there 63 // propagated to callers but the transaction still considered as committed. 64 try { 65 triggerAfterCommit(status); 66 } 67 finally { 68 triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED); 69 } 70 71 } 72 finally { 73 cleanupAfterCompletion(status); 74 } 75 }
    复制代码

    如上图,commit事务时,有6个核心操作,分别是3个前置操作,3个后置操作,如下:

    1.prepareForCommit(status);源码是空的,没有拓展目前。

    2.triggerBeforeCommit(status); 提交前触发操作

    复制代码
    1 protected final void triggerBeforeCommit(DefaultTransactionStatus status) {
    2         if (status.isNewSynchronization()) {
    3             if (status.isDebug()) {
    4                 logger.trace("Triggering beforeCommit synchronization");
    5             }
    6             TransactionSynchronizationUtils.triggerBeforeCommit(status.isReadOnly());
    7         }
    8     }
    复制代码

    triggerBeforeCommit源码如下:

    1 public static void triggerBeforeCommit(boolean readOnly) {
    2         for (TransactionSynchronization synchronization : TransactionSynchronizationManager.getSynchronizations()) {
    3             synchronization.beforeCommit(readOnly);
    4         }
    5     }

     如上图,TransactionSynchronizationManager类定义了多个ThreadLocal(线程本地变量),其中一个用以保存当前线程的事务同步:

    private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations = new NamedThreadLocal<Set<TransactionSynchronization>>("Transaction synchronizations");

    遍历事务同步器,把每个事务同步器都执行“提交前”操作,比如咱们用的jdbc事务,那么最终就是SqlSessionUtils.beforeCommit()->this.holder.getSqlSession().commit();提交会话。(源码由于是spring管理实务,最终不会执行事务提交,例如是DefaultSqlSession:执行清除缓存、重置状态操作)

    3.triggerBeforeCompletion(status);完成前触发操作,如果是jdbc事务,那么最终就是,

    SqlSessionUtils.beforeCompletion->

    TransactionSynchronizationManager.unbindResource(sessionFactory); 解绑当前线程的会话工厂

    this.holder.getSqlSession().close();关闭会话。(源码由于是spring管理实务,最终不会执行事务close操作,例如是DefaultSqlSession,也会执行各种清除收尾操作)

    4.triggerAfterCommit(status);提交事务后触发操作。TransactionSynchronizationUtils.triggerAfterCommit();->TransactionSynchronizationUtils.invokeAfterCommit,如下:

    复制代码
    1 public static void invokeAfterCommit(List<TransactionSynchronization> synchronizations) {
    2         if (synchronizations != null) {
    3             for (TransactionSynchronization synchronization : synchronizations) {
    4                 synchronization.afterCommit();
    5             }
    6         }
    7     }
    复制代码

    好吧,一顿找,最后在TransactionSynchronizationAdapter中复写过,并且是空的....SqlSessionSynchronization继承了TransactionSynchronizationAdapter但是没有复写这个方法。

    5. triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);

    TransactionSynchronizationUtils.TransactionSynchronizationUtils.invokeAfterCompletion,如下:

    复制代码
     1 public static void invokeAfterCompletion(List<TransactionSynchronization> synchronizations, int completionStatus) {
     2         if (synchronizations != null) {
     3             for (TransactionSynchronization synchronization : synchronizations) {
     4                 try {
     5                     synchronization.afterCompletion(completionStatus);
     6                 }
     7                 catch (Throwable tsex) {
     8                     logger.error("TransactionSynchronization.afterCompletion threw exception", tsex);
     9                 }
    10             }
    11         }
    12     }
    复制代码

    afterCompletion:对于JDBC事务来说,最终:

    1)如果会话任然活着,关闭会话,

    2)重置各种属性:SQL会话同步器(SqlSessionSynchronization)的SQL会话持有者(SqlSessionHolder)的referenceCount引用计数、synchronizedWithTransaction同步事务、rollbackOnly只回滚、deadline超时时间点。

    6.cleanupAfterCompletion(status);

    1)设置事务状态为已完成。

    2)  如果是新的事务同步,解绑当前线程绑定的数据库资源,重置数据库连接

    3)如果存在挂起的事务(嵌套事务),唤醒挂起的老事务的各种资源:数据库资源、同步器。

    复制代码
     1     private void cleanupAfterCompletion(DefaultTransactionStatus status) {
     2         status.setCompleted();//设置事务状态完成
           //如果是新的同步,清空当前线程绑定的除了资源外的全部线程本地变量:包括事务同步器、事务名称、只读属性、隔离级别、真实的事务激活状态
    3 if (status.isNewSynchronization()) { 4 TransactionSynchronizationManager.clear(); 5 }//如果是新的事务同步 6 if (status.isNewTransaction()) { 7 doCleanupAfterCompletion(status.getTransaction()); 8 }//如果存在挂起的资源 9 if (status.getSuspendedResources() != null) { 10 if (status.isDebug()) { 11 logger.debug("Resuming suspended transaction after completion of inner transaction"); 12 }//唤醒挂起的事务和资源(重新绑定之前挂起的数据库资源,唤醒同步器,注册同步器到TransactionSynchronizationManager) 13 resume(status.getTransaction(), (SuspendedResourcesHolder) status.getSuspendedResources()); 14 } 15 }
    复制代码

    对于DataSourceTransactionManager,doCleanupAfterCompletion源码如下:

    复制代码
     1     protected void doCleanupAfterCompletion(Object transaction) {
     2         DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
     3 
     4         // 如果是最新的连接持有者,解绑当前线程绑定的<数据库资源,ConnectionHolder>
     5         if (txObject.isNewConnectionHolder()) {
     6             TransactionSynchronizationManager.unbindResource(this.dataSource);
     7         }
     8 
     9         // 重置数据库连接(隔离级别、只读)
    10         Connection con = txObject.getConnectionHolder().getConnection();
    11         try {
    12             if (txObject.isMustRestoreAutoCommit()) {
    13                 con.setAutoCommit(true);
    14             }
    15             DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel());
    16         }
    17         catch (Throwable ex) {
    18             logger.debug("Could not reset JDBC Connection after transaction", ex);
    19         }
    20 
    21         if (txObject.isNewConnectionHolder()) {
    22             if (logger.isDebugEnabled()) {
    23                 logger.debug("Releasing JDBC Connection [" + con + "] after transaction");
    24             }// 资源引用计数-1,关闭数据库连接
    25             DataSourceUtils.releaseConnection(con, this.dataSource);
    26         }
    27         // 重置连接持有者的全部属性
    28         txObject.getConnectionHolder().clear();
    29     }
    复制代码

     

    3.3 rollback回滚事务

     AbstractPlatformTransactionManager中rollback源码如下:

    复制代码
    1     public final void rollback(TransactionStatus status) throws TransactionException {
    2         if (status.isCompleted()) {
    3             throw new IllegalTransactionStateException(
    4                     "Transaction is already completed - do not call commit or rollback more than once per transaction");
    5         }
    6 
    7         DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
    8         processRollback(defStatus);
    9     }
    复制代码

     processRollback源码如下:

    复制代码
     1     private void processRollback(DefaultTransactionStatus status) {
     2         try {
     3             try {// 解绑当前线程绑定的会话工厂,并关闭会话
     4                 triggerBeforeCompletion(status);
     5                 if (status.hasSavepoint()) {// 1.如果有保存点,即嵌套式事务
     6                     if (status.isDebug()) {
     7                         logger.debug("Rolling back transaction to savepoint");
     8                     }//回滚到保存点
     9                     status.rollbackToHeldSavepoint();
    10                 }//2.如果就是一个简单事务
    11                 else if (status.isNewTransaction()) {
    12                     if (status.isDebug()) {
    13                         logger.debug("Initiating transaction rollback");
    14                     }//回滚核心方法
    15                     doRollback(status);
    16                 }//3.当前存在事务且没有保存点,即加入当前事务的
    17                 else if (status.hasTransaction()) {//如果已经标记为回滚 或 当加入事务失败时全局回滚(默认true)
    18                     if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
    19                         if (status.isDebug()) {//debug时会打印:加入事务失败-标记已存在事务为回滚
    20                             logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
    21                         }//设置当前connectionHolder:当加入一个已存在事务时回滚
    22                         doSetRollbackOnly(status);
    23                     }
    24                     else {
    25                         if (status.isDebug()) {
    26                             logger.debug("Participating transaction failed - letting transaction originator decide on rollback");
    27                         }
    28                     }
    29                 }
    30                 else {
    31                     logger.debug("Should roll back transaction but cannot - no transaction available");
    32                 }
    33             }
    34             catch (RuntimeException ex) {//关闭会话,重置SqlSessionHolder属性
    35                 triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
    36                 throw ex;
    37             }
    38             catch (Error err) {
    39                 triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
    40                 throw err;
    41             }
    42             triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
    43         }
    44         finally {、、解绑当前线程
    45             cleanupAfterCompletion(status);
    46         }
    47     }
    复制代码

    如上图,有几个公共方法和提交事务时一致,就不再重复。

    这里主要看doRollback,DataSourceTransactionManager的doRollback()源码如下:

    复制代码
     1 protected void doRollback(DefaultTransactionStatus status) {
     2         DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
     3         Connection con = txObject.getConnectionHolder().getConnection();
     4         if (status.isDebug()) {
     5             logger.debug("Rolling back JDBC transaction on Connection [" + con + "]");
     6         }
     7         try {
     8             con.rollback();
     9         }
    10         catch (SQLException ex) {
    11             throw new TransactionSystemException("Could not roll back JDBC transaction", ex);
    12         }
    13     }
    复制代码

    好吧,一点不复杂,就是Connection的rollback.

     四、时序图

    特地整理了时序图(简单的新事务,没有画出保存点等情况)如下:

     

    ===========参考========

    《Spring实战4》第四章 面向切面的Spring  

    SpringBoot事务注解@Transactional

    点赞(0)

    评论列表 共有 0 条评论

    暂无评论
    立即
    投稿
    发表
    评论
    返回
    顶部