Spring 事务

声明式事务

Propagation

  • 事务传播
  1. public enum Propagation {
  3.    /**
  4.     * 有事务则加入，没有则新建
  5.     */
  6.    REQUIRED(TransactionDefinition.PROPAGATION_REQUIRED),
  8.    /**
  9.     * 有事务就用，如果没有就不开启(继承关系)
  10.     * @see org.springframework.transaction.support.AbstractPlatformTransactionManager#setTransactionSynchronization
  11.     */
  12.    SUPPORTS(TransactionDefinition.PROPAGATION_SUPPORTS),
  14.    /**
  15.     *  必须在已有事务中
  16.     */
  17.    MANDATORY(TransactionDefinition.PROPAGATION_MANDATORY),
  19.    /**
  20.     * 不管是否已有事务，都要开启新事务，老事务挂起
  21.     * @see org.springframework.transaction.jta.JtaTransactionManager#setTransactionManager
  22.     */
  23.    REQUIRES_NEW(TransactionDefinition.PROPAGATION_REQUIRES_NEW),
  25.    /**
  26.     * 不开启事务
  27.     * @see org.springframework.transaction.jta.JtaTransactionManager#setTransactionManager
  28.     */
  29.    NOT_SUPPORTED(TransactionDefinition.PROPAGATION_NOT_SUPPORTED),
  31.    /**
  32.     * 必须在没有事务的方法中调用，否则抛出异常
  33.     */
  34.    NEVER(TransactionDefinition.PROPAGATION_NEVER),
  36.    /**
  37.     * 果已有事务，则嵌套执行，如果没有，就新建(和REQUIRED类似，和REQUIRES_NEW容易混淆)
  38.     * @see org.springframework.jdbc.datasource.DataSourceTransactionManager
  39.     */
  40.    NESTED(TransactionDefinition.PROPAGATION_NESTED);
  43.    private final int value;
  46.    Propagation(int value) {
  47.       this.value = value;
  48.    }
  50.    public int value() {
  51.       return this.value;
  52.    }
  54. }

Isolation

  • 事务级别
  1. public enum Isolation {
  3.    /**
  4.     * @see java.sql.Connection
  5.     */
  6.    DEFAULT(TransactionDefinition.ISOLATION_DEFAULT),
  8.    /**
  9.     * 读未提交
  10.     *
  11.     * @see java.sql.Connection#TRANSACTION_READ_UNCOMMITTED
  12.     */
  13.    READ_UNCOMMITTED(TransactionDefinition.ISOLATION_READ_UNCOMMITTED),
  15.    /**
  16.     * 读已提交
  17.     *
  18.     * @see java.sql.Connection#TRANSACTION_READ_COMMITTED
  19.     */
  20.    READ_COMMITTED(TransactionDefinition.ISOLATION_READ_COMMITTED),
  22.    /**
  23.     * 可重复读
  24.     *
  25.     * @see java.sql.Connection#TRANSACTION_REPEATABLE_READ
  26.     */
  27.    REPEATABLE_READ(TransactionDefinition.ISOLATION_REPEATABLE_READ),
  29.    /**
  30.     * 可串行化
  31.     *
  32.     * @see java.sql.Connection#TRANSACTION_SERIALIZABLE
  33.     */
  34.    SERIALIZABLE(TransactionDefinition.ISOLATION_SERIALIZABLE);
  37.    private final int value;
  40.    Isolation(int value) {
  41.       this.value = value;
  42.    }
  44.    public int value() {
  45.       return this.value;
  46.    }
  48. }

EnableTransactionManagement

  • 下面代码是一个注解方式的事务配置使用 EnableTransactionManagement来开启事务支持
  1. @ComponentScan(basePackages = "org.source.hot.spring.overview.ioc.tx.declarative")
  2. @EnableTransactionManagement
  3. public class TxConfig {
  5.     @Bean // 数据源
  6.     public DataSource dataSource() {
  7.         DruidDataSource dataSource = new DruidDataSource();
  8.         dataSource.setUsername("");
  9.         dataSource.setPassword("");
  10.         dataSource.setUrl("");
  11.         dataSource.setDriverClassName(com.mysql.jdbc.Driver.class.getName());
  12.         return dataSource;
  13.     }
  15.     @Bean
  16.     public JdbcTemplate jdbcTemplate(DataSource dataSource) {
  17.         return new JdbcTemplate(dataSource);
  18.     }
  20.     @Bean //事务管理器
  21.     public PlatformTransactionManager platformTransactionManager(DataSource dataSource) {
  22.         return new DataSourceTransactionManager(dataSource);
  23.     }
  25. }
  • 注解源码如下,关注于@Import(TransactionManagementConfigurationSelector.class)
  1. @Target(ElementType.TYPE)
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Documented
  4. @Import(TransactionManagementConfigurationSelector.class)
  5. public @interface EnableTransactionManagement {
  7.    boolean proxyTargetClass() default false;
  9.    AdviceMode mode() default AdviceMode.PROXY;
  11.    int order() default Ordered.LOWEST_PRECEDENCE;
  13. }
  1. public class TransactionManagementConfigurationSelector extends AdviceModeImportSelector<EnableTransactionManagement> {
  3.    @Override
  4.    protected String[] selectImports(AdviceMode adviceMode) {
  5.       // 根据切面类型进行初始化
  6.       switch (adviceMode) {
  7.          case PROXY:
  8.               // 默认值
  9.             return new String[] {AutoProxyRegistrar.class.getName(),
  10.                   ProxyTransactionManagementConfiguration.class.getName()};
  11.          case ASPECTJ:
  12.             return new String[] {determineTransactionAspectClass()};
  13.          default:
  14.             return null;
  15.       }
  16.    }
  18.    private String determineTransactionAspectClass() {
  19.       return (ClassUtils.isPresent("javax.transaction.Transactional", getClass().getClassLoader()) ?
  20.             TransactionManagementConfigUtils.JTA_TRANSACTION_ASPECT_CONFIGURATION_CLASS_NAME :
  21.             TransactionManagementConfigUtils.TRANSACTION_ASPECT_CONFIGURATION_CLASS_NAME);
  22.    }
  24. }

ProxyTransactionManagementConfiguration

  1. @Configuration(proxyBeanMethods = false)
  2. public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {
  5.    /**
  6.     * 事务切面
  7.     * @param transactionAttributeSource
  8.     * @param transactionInterceptor
  9.     * @return
  10.     */
  11.    @Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
  12.    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  13.    public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(
  14.          TransactionAttributeSource transactionAttributeSource,
  15.          TransactionInterceptor transactionInterceptor) {
  16.       // 事务切面
  17.       BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
  18.       // 事务属性
  19.       advisor.setTransactionAttributeSource(transactionAttributeSource);
  20.       advisor.setAdvice(transactionInterceptor);
  21.       if (this.enableTx != null) {
  22.          // 执行顺序
  23.          advisor.setOrder(this.enableTx.<Integer>getNumber("order"));
  24.       }
  25.       return advisor;
  26.    }
  28.    @Bean
  29.    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  30.    public TransactionAttributeSource transactionAttributeSource() {
  31.       return new AnnotationTransactionAttributeSource();
  32.    }
  34.    /***
  35.     * 事务拦截器
  36.     * @param transactionAttributeSource
  37.     * @return
  38.     */
  39.    @Bean
  40.    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  41.    public TransactionInterceptor transactionInterceptor(
  42.          TransactionAttributeSource transactionAttributeSource) {
  43.       TransactionInterceptor interceptor = new TransactionInterceptor();
  44.       interceptor.setTransactionAttributeSource(transactionAttributeSource);
  45.       if (this.txManager != null) {
  46.          // 事务管理器注入
  47.          interceptor.setTransactionManager(this.txManager);
  48.       }
  49.       return interceptor;
  50.    }
  52. }

TransactionInterceptor

image-20200729144622440

  • 实现了org.aopalliance.intercept.MethodInterceptor接口的方法
  1. @Override
  2. @Nullable
  3. public 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);
  9.    // Adapt to TransactionAspectSupport's invokeWithinTransaction...
  10.    return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
  11. }
  • 这段代码会在具有Transactional 的注解方法上生效
  1. @Service
  2. public class IssueServiceImpl {
  4.    @Autowired
  5.    private JdbcTemplate jdbcTemplate;
  7.    @Transactional()
  8.    public boolean insertIssue() throws Exception {
  9.       jdbcTemplate.execute("INSERT INTO `scrum`.`issue`() VALUES ()");
  11.       throw new Exception("a");
  12.    }
  14. }
  17. public class DeclarativeTransactionTest {
  19.     public static void main(String[] args) throws Exception {
  21.         AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(
  22.                 TxConfig.class);
  23.         IssueServiceImpl bean = applicationContext.getBean(IssueServiceImpl.class);
  24.         bean.insertIssue();
  25.         System.out.println();
  26.         applicationContext.close();
  27.     }
  30. }

image-20200729145518089

断点开始进行查阅. 再断点后执行一步会直接进入 cglib 代理对象

org.springframework.aop.framework.CglibAopProxy.DynamicAdvisedInterceptor#intercept 具体不展开，继续往下执行

image-20200729145637688

走到invoke方法了

入参对象查看

image-20200729145835608

  • 获取事务属性

    ```java @Override @Nullable public TransactionAttribute getTransactionAttribute(Method method,

    1.   @Nullable Class<?> targetClass) {

    if (method.getDeclaringClass() == Object.class) {

    1.   return null;

    }

    // First, see if we have a cached value. // 尝试缓存中获取 Object cacheKey = getCacheKey(method, targetClass); TransactionAttribute cached = this.attributeCache.get(cacheKey); if (cached != null) {

    1.   // Value will either be canonical value indicating there is no transaction attribute,
    2.   // or an actual transaction attribute.
    3.   if (cached == NULL_TRANSACTION_ATTRIBUTE) {
    4.      return null;
    5.   } else {
    6.      return cached;
    7.   }

    } else {

    1.   // We need to work it out.
    2.   // 自行构建一个事务属性
    3.   TransactionAttribute txAttr = computeTransactionAttribute(method, targetClass);
    4.   // Put it in the cache.
    5.   if (txAttr == null) {
    6.      this.attributeCache.put(cacheKey, NULL_TRANSACTION_ATTRIBUTE);
    7.   } else {
    8.      String methodIdentification = ClassUtils
    9.            .getQualifiedMethodName(method, targetClass);
    10.      if (txAttr instanceof DefaultTransactionAttribute) {
    11.         ((DefaultTransactionAttribute) txAttr).setDescriptor(methodIdentification);
    12.      }
    13.      if (logger.isTraceEnabled()) {
    14.         logger.trace("Adding transactional method '" + methodIdentification
    15.               + "' with attribute: " + txAttr);
    16.      }
    17.      this.attributeCache.put(cacheKey, txAttr);
    18.   }
    19.   return txAttr;

    } }

  1.   protected Object getCacheKey(Method method, @Nullable Class<?> targetClass) {
  2.       return new MethodClassKey(method, targetClass);
  3.   }
  2. ![image-20200729162023837](/uploads/projects/SpringBoot-source/docs/Spring/TX/images/spring/image-20200729162023837.png)
  4. - 此处方法已经获取到了这个方法就是后面的一个切面
  6. - 确定事务管理器
  8.   ```java
  9.   @Nullable
  10.   protected TransactionManager determineTransactionManager(
  11.         @Nullable TransactionAttribute txAttr) {
  12.      // Do not attempt to lookup tx manager if no tx attributes are set
  13.      // 空判断返回一个事务管理器
  14.      if (txAttr == null || this.beanFactory == null) {
  15.         return getTransactionManager();
  16.      }
  18.      // 属性是否有别名
  19.      String qualifier = txAttr.getQualifier();
  20.      // 如果有
  21.      if (StringUtils.hasText(qualifier)) {
  22.         // 从 ioc 容器中根据类型和名称获取事务管理器
  23.         return determineQualifiedTransactionManager(this.beanFactory, qualifier);
  24.      } else if (StringUtils.hasText(this.transactionManagerBeanName)) {
  25.         // 从 ioc 容器中根据类型和名称获取事务管理器
  26.         return determineQualifiedTransactionManager(this.beanFactory,
  27.               this.transactionManagerBeanName);
  28.      } else {
  29.         // 通过get方法获取
  30.         TransactionManager defaultTransactionManager = getTransactionManager();
  31.         // 如果没有
  32.         if (defaultTransactionManager == null) {
  33.            // 尝试从缓存中获取
  34.            defaultTransactionManager = this.transactionManagerCache
  35.                  .get(DEFAULT_TRANSACTION_MANAGER_KEY);
  36.            // 缓存里面没有从 ioc 容器中获取并且设置缓存
  37.            if (defaultTransactionManager == null) {
  38.               defaultTransactionManager = this.beanFactory.getBean(TransactionManager.class);
  39.               this.transactionManagerCache.putIfAbsent(
  40.                     DEFAULT_TRANSACTION_MANAGER_KEY, defaultTransactionManager);
  41.            }
  42.         }
  43.         return defaultTransactionManager;
  44.      }
  45.   }

image-20200729160650401

  • 类型转换

    1. @Nullable
    2. private PlatformTransactionManager asPlatformTransactionManager(
    3.       @Nullable Object transactionManager) {
    4.    if (transactionManager == null
    5.          || transactionManager instanceof PlatformTransactionManager) {
    6.       return (PlatformTransactionManager) transactionManager;
    7.    } else {
    8.       throw new IllegalStateException(
    9.             "Specified transaction manager is not a PlatformTransactionManager: "
    10.                   + transactionManager);
    11.    }
    12. }

  • 获取方法切面

    1. private String methodIdentification(Method method, @Nullable Class<?> targetClass,
    2.       @Nullable TransactionAttribute txAttr) {
    4.    String methodIdentification = methodIdentification(method, targetClass);
    5.    if (methodIdentification == null) {
    6.       if (txAttr instanceof DefaultTransactionAttribute) {
    7.          // 直接就获取了.方法签名.
    8.          methodIdentification = ((DefaultTransactionAttribute) txAttr).getDescriptor();
    9.       }
    10.       if (methodIdentification == null) {
    11.          methodIdentification = ClassUtils.getQualifiedMethodName(method, targetClass);
    12.       }
    13.    }
    14.    return methodIdentification;
    15. }

image-20200729161647214

  • 创建一个新的事务根据事务传播性

    1.     @SuppressWarnings("serial")
    2.     protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
    3.             @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
    5.         // If no name specified, apply method identification as transaction name.
    6.         // 把切面的地址放进去
    7.         if (txAttr != null && txAttr.getName() == null) {
    8.             txAttr = new DelegatingTransactionAttribute(txAttr) {
    9.                 @Override
    10.                 public String getName() {
    11.                     return joinpointIdentification;
    12.                 }
    13.             };
    14.         }
    16.         TransactionStatus status = null;
    17.         if (txAttr != null) {
    18.             if (tm != null) {
    19.                 // 事务状态
    20.                 // 获取事务
    21.                 status = tm.getTransaction(txAttr);
    22.             } else {
    23.                 if (logger.isDebugEnabled()) {
    24.                     logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
    25.                             "] because no transaction manager has been configured");
    26.                 }
    27.             }
    28.         }
    29.         // 处理出一个 TransactionInfo
    30.         return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
    31.     }

image-20200729163303000

  • tm.getTransaction

    1. @Override
    2. public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
    3.       throws TransactionException {
    5.    // Use defaults if no transaction definition given.
    6.    // 获取事务的定义
    7.    TransactionDefinition def = (definition != null ? definition
    8.          : TransactionDefinition.withDefaults());
    10.    // 获取事务
    11.    Object transaction = doGetTransaction();
    12.    boolean debugEnabled = logger.isDebugEnabled();
    14.    // 是否存在事务
    15.    if (isExistingTransaction(transaction)) {
    16.       // Existing transaction found -> check propagation behavior to find out how to behave.
    17.       // 存在事务后处理什么操作
    18.       return handleExistingTransaction(def, transaction, debugEnabled);
    19.    }
    21.    // Check definition settings for new transaction.
    22.    // 超时的校验. 小于默认值抛出异常
    23.    if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
    24.       throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
    25.    }
    27.    // No existing transaction found -> check propagation behavior to find out how to proceed.
    28.    // 没有事务抛出异常
    29.    if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
    30.       throw new IllegalTransactionStateException(
    31.             "No existing transaction found for transaction marked with propagation 'mandatory'");
    32.    } else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
    33.          def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
    34.          def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
    35.       SuspendedResourcesHolder suspendedResources = suspend(null);
    36.       if (debugEnabled) {
    37.          logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
    38.       }
    39.       try {
    40.          boolean newSynchronization = (getTransactionSynchronization()
    41.                != SYNCHRONIZATION_NEVER);
    42.          DefaultTransactionStatus status = newTransactionStatus(
    43.                def, transaction, true, newSynchronization, debugEnabled,
    44.                suspendedResources);
    45.          doBegin(transaction, def);
    46.          prepareSynchronization(status, def);
    47.          return status;
    48.       } catch (RuntimeException | Error ex) {
    49.          resume(null, suspendedResources);
    50.          throw ex;
    51.       }
    52.    } else {
    53.       // Create "empty" transaction: no actual transaction, but potentially synchronization.
    54.       if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger
    55.             .isWarnEnabled()) {
    56.          logger.warn(
    57.                "Custom isolation level specified but no actual transaction initiated; " +
    58.                      "isolation level will effectively be ignored: " + def);
    59.       }
    60.       boolean newSynchronization = (getTransactionSynchronization()
    61.             == SYNCHRONIZATION_ALWAYS);
    62.       return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled,
    63.             null);
    64.    }
    65. }

    • org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction

    • org.springframework.jdbc.datasource.DataSourceTransactionManager#doGetTransaction

      1. @Override
      2. protected Object doGetTransaction() {
      3.    DataSourceTransactionObject txObject = new DataSourceTransactionObject();
      4.    txObject.setSavepointAllowed(isNestedTransactionAllowed());
      5.    // 数据库链接对象
      6.    // 从事务管理器中获取数据库链接对象
      7.    ConnectionHolder conHolder =
      8.          (ConnectionHolder) TransactionSynchronizationManager
      9.                .getResource(obtainDataSource());
      10.    txObject.setConnectionHolder(conHolder, false);
      11.    return txObject;
      12. }

    • org.springframework.transaction.support.AbstractPlatformTransactionManager#suspend

      1. @Nullable
      2. protected final SuspendedResourcesHolder suspend(@Nullable Object transaction)
      3.       throws TransactionException {
      4.    if (TransactionSynchronizationManager.isSynchronizationActive()) {
      5.       List<TransactionSynchronization> suspendedSynchronizations = doSuspendSynchronization();
      6.       try {
      7.          Object suspendedResources = null;
      8.          if (transaction != null) {
      9.             suspendedResources = doSuspend(transaction);
      10.          }
      11.          // 线程名称
      12.          String name = TransactionSynchronizationManager.getCurrentTransactionName();
      13.          // 同步方法中设置
      14.          TransactionSynchronizationManager.setCurrentTransactionName(null);
      15.          // 只读设置
      16.          boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();
      17.          // 同步方法中设置
      18.          TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);
      19.          // 隔离级别
      20.          Integer isolationLevel = TransactionSynchronizationManager
      21.                .getCurrentTransactionIsolationLevel();
      22.          // 同步方法中设置
      23.          TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(null);
      24.          // 是否活跃
      25.          boolean wasActive = TransactionSynchronizationManager.isActualTransactionActive();
      26.          TransactionSynchronizationManager.setActualTransactionActive(false);
      27.          return new SuspendedResourcesHolder(
      28.                suspendedResources, suspendedSynchronizations, name, readOnly,
      29.                isolationLevel, wasActive);
      30.       } catch (RuntimeException | Error ex) {
      31.          // doSuspend failed - original transaction is still active...
      32.          doResumeSynchronization(suspendedSynchronizations);
      33.          throw ex;
      34.       }
      35.    } else if (transaction != null) {
      36.       // Transaction active but no synchronization active.
      37.       Object suspendedResources = doSuspend(transaction);
      38.       return new SuspendedResourcesHolder(suspendedResources);
      39.    } else {
      40.       // Neither transaction nor synchronization active.
      41.       return null;
      42.    }
      43. }

  • prepareTransactionInfo简单的new对象并且绑定线程

    1. protected TransactionInfo prepareTransactionInfo(@Nullable PlatformTransactionManager tm,
    2.       @Nullable TransactionAttribute txAttr, String joinpointIdentification,
    3.       @Nullable TransactionStatus status) {
    5.    // 初始化
    6.    TransactionInfo txInfo = new TransactionInfo(tm, txAttr, joinpointIdentification);
    7.    if (txAttr != null) {
    8.       // We need a transaction for this method...
    9.       if (logger.isTraceEnabled()) {
    10.          logger.trace(
    11.                "Getting transaction for [" + txInfo.getJoinpointIdentification() + "]");
    12.       }
    13.       // The transaction manager will flag an error if an incompatible tx already exists.
    14.       txInfo.newTransactionStatus(status);
    15.    } else {
    16.       // The TransactionInfo.hasTransaction() method will return false. We created it only
    17.       // to preserve the integrity of the ThreadLocal stack maintained in this class.
    18.       if (logger.isTraceEnabled()) {
    19.          logger.trace("No need to create transaction for [" + joinpointIdentification +
    20.                "]: This method is not transactional.");
    21.       }
    22.    }
    24.    // We always bind the TransactionInfo to the thread, even if we didn't create
    25.    // a new transaction here. This guarantees that the TransactionInfo stack
    26.    // will be managed correctly even if no transaction was created by this aspect.
    27.    // 和线程绑定
    28.    txInfo.bindToThread();
    29.    return txInfo;
    30. }

  • retVal = invocation.proceedWithInvocation();

    • 这里走的是 CGLIB 的方法直接会执行结果将结果返回具体方法在

      org.springframework.aop.framework.CglibAopProxy.CglibMethodInvocation#proceed

    1.         @Override
    2.         @Nullable
    3.         public Object proceed() throws Throwable {
    4.             try {
    5.                 return super.proceed();
    6.             }
    7.             catch (RuntimeException ex) {
    8.                 throw ex;
    9.             }
    10.             catch (Exception ex) {
    11.                 if (ReflectionUtils.declaresException(getMethod(), ex.getClass())) {
    12.                     throw ex;
    13.                 }
    14.                 else {
    15.                     throw new UndeclaredThrowableException(ex);
    16.                 }
    17.             }
    18.         }

  • 如果没有异常就直接处理完成返回了

  • 我们现在是有异常的

    1. try {
    2.    // This is an around advice: Invoke the next interceptor in the chain.
    3.    // This will normally result in a target object being invoked.
    4.    // 回调方法
    5.    retVal = invocation.proceedWithInvocation();
    6. } catch (Throwable ex) {
    7.    // target invocation exception
    8.    // 回滚异常
    9.    completeTransactionAfterThrowing(txInfo, ex);
    10.    throw ex;
    11. } finally {
    12.    // 消息清理
    13.    cleanupTransactionInfo(txInfo);
    14. }

  • completeTransactionAfterThrowing回滚异常的处理方法

    1. protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo,
    2.       Throwable ex) {
    3.    if (txInfo != null && txInfo.getTransactionStatus() != null) {
    4.       if (logger.isTraceEnabled()) {
    5.          logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() +
    6.                "] after exception: " + ex);
    7.       }
    8.       if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
    9.          try {
    10.             // 做回滚
    11.             txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
    12.          } catch (TransactionSystemException ex2) {
    13.             logger.error("Application exception overridden by rollback exception", ex);
    14.             ex2.initApplicationException(ex);
    15.             throw ex2;
    16.          } catch (RuntimeException | Error ex2) {
    17.             logger.error("Application exception overridden by rollback exception", ex);
    18.             throw ex2;
    19.          }
    20.       } else {
    21.          // We don't roll back on this exception.
    22.          // Will still roll back if TransactionStatus.isRollbackOnly() is true.
    23.          try {
    24.             // org.springframework.transaction.support.AbstractPlatformTransactionManager.commit 的方法
    25.             txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
    26.          } catch (TransactionSystemException ex2) {
    27.             logger.error("Application exception overridden by commit exception", ex);
    28.             ex2.initApplicationException(ex);
    29.             throw ex2;
    30.          } catch (RuntimeException | Error ex2) {
    31.             logger.error("Application exception overridden by commit exception", ex);
    32.             throw ex2;
    33.          }
    34.       }
    35.    }
    36. }

    • 整理一下这里的流程

      1. 有异常走回滚

        txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus())

      2. 没有异常直接提交

        txInfo.getTransactionManager().commit(txInfo.getTransactionStatus())

    • 注意: 这里的异常如果是 exception 不会走回滚

  • 判断是否需要回滚

    1. txInfo.transactionAttribute.rollbackOn

    • 链路

      • org.springframework.transaction.interceptor.DelegatingTransactionAttribute#rollbackOn

        • org.springframework.transaction.interceptor.RuleBasedTransactionAttribute#rollbackOn

          1. @Override
          2. public boolean rollbackOn(Throwable ex) {
          3.    if (logger.isTraceEnabled()) {
          4.       logger.trace(
          5.             "Applying rules to determine whether transaction should rollback on " + ex);
          6.    }
          8.    RollbackRuleAttribute winner = null;
          9.    int deepest = Integer.MAX_VALUE;
          11.    if (this.rollbackRules != null) {
          12.       for (RollbackRuleAttribute rule : this.rollbackRules) {
          13.          int depth = rule.getDepth(ex);
          14.          if (depth >= 0 && depth < deepest) {
          15.             deepest = depth;
          16.             winner = rule;
          17.          }
          18.       }
          19.    }
          21.    if (logger.isTraceEnabled()) {
          22.       logger.trace("Winning rollback rule is: " + winner);
          23.    }
          25.    // User superclass behavior (rollback on unchecked) if no rule matches.
          26.    if (winner == null) {
          27.       logger.trace("No relevant rollback rule found: applying default rules");
          28.       return super.rollbackOn(ex);
          29.    }
          31.    return !(winner instanceof NoRollbackRuleAttribute);
          32. }

          • org.springframework.transaction.interceptor.DefaultTransactionAttribute#rollbackOn

            1. @Override
            2. public boolean rollbackOn(Throwable ex) {
            3.    return (ex instanceof RuntimeException || ex instanceof Error);
            4. }
            • 这就是我们的异常判断是否需要回滚

  • cleanupTransactionInfo

    数据清理

    1. protected void cleanupTransactionInfo(@Nullable TransactionInfo txInfo) {
    2.    if (txInfo != null) {
    3.       txInfo.restoreThreadLocalStatus();
    4.    }
    5. }
    1. private void restoreThreadLocalStatus() {
    2.    // Use stack to restore old transaction TransactionInfo.
    3.    // Will be null if none was set.
    4.    transactionInfoHolder.set(this.oldTransactionInfo);
    5. }

编程式事务

DefaultTransactionDefinition

  • 默认的事务定义
    • 常见属性
      1. timeout
      2. readOnly
      3. ….

PlatformTransactionManager

  1. // 获取事务
  2. TransactionStatus getTransaction(@Nullable TransactionDefinition definition)throws TransactionException;
  3. // 提交事务
  4. void commit(TransactionStatus status) throws TransactionException;
  5. // 回滚事务
  6. void rollback(TransactionStatus status) throws TransactionException;
  • 贴出一部分

image-20200728105926218

  • AbstractPlatformTransactionManager 定义了一些基础属性 以及一些需要子类实现的方法
  1. // 属性
  2. defaultTimeout
  3. nestedTransactionAllowed
  4. validateExistingTransaction
  5. globalRollbackOnParticipationFailure
  6. failEarlyOnGlobalRollbackOnly
  7. rollbackOnCommitFailure
  8. // 方法
  9. doGetTransaction
  10. isExistingTransaction
  11. useSavepointForNestedTransaction
  12. doBegin
  13. doSuspend
  14. doResume
  15. shouldCommitOnGlobalRollbackOnly
  16. prepareForCommit
  17. doCommit
  18. doRollback
  19. doSetRollbackOnly
  20. registerAfterCompletionWithExistingTransaction
  21. doCleanupAfterCompletion

DataSourceTransactionManager

  • xml 配置如下
  1.     <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource">
  2.         <property name="url"
  3.                 value=""/>
  4.         <property name="username" value=""/>
  5.         <property name="password" value=""/>
  6.         <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
  7.     </bean>
  8.     <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
  9.         <property name="dataSource" ref="dataSource"/>
  10.     </bean>
  11.     <bean id="transactionManager"
  12.             class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
  13.         <property name="dataSource" ref="dataSource"/>
  14.     </bean>

  • 两个属性，通常我们会配置 datasource

    1.     @Nullable
    2.     private DataSource dataSource;
    4.     private boolean enforceReadOnly = false;
    • bean 的属性注入就不具体描述了

image-20200728133037075

  • InitializingBean

      1. @Override
      2. public void afterPropertiesSet() {
      3.    if (getDataSource() == null) {
      4.       throw new IllegalArgumentException("Property 'dataSource' is required");
      5.    }
      6. }
      • 如果dataSource为空会抛出异常
      • 默认单例会注册到 ioc 容器中.后续注册流程不具体描述

  • 方法注释

  1.     /**
  2.      * 获取datasource
  3.      */
  4.     protected DataSource obtainDataSource() {
  5.         DataSource dataSource = getDataSource();
  6.         Assert.state(dataSource != null, "No DataSource set");
  7.         return dataSource;
  8.     }
  10.     /**
  11.      * 创建事务
  12.      *
  13.      * @return 事务对象
  14.      */
  15.     @Override
  16.     protected Object doGetTransaction() {
  17.         DataSourceTransactionObject txObject = new DataSourceTransactionObject();
  18.         txObject.setSavepointAllowed(isNestedTransactionAllowed());
  19.         // 数据库链接对象
  20.         // 从事务管理器中获取数据库链接对象
  21.         ConnectionHolder conHolder =
  22.                 (ConnectionHolder) TransactionSynchronizationManager
  23.                         .getResource(obtainDataSource());
  24.         txObject.setConnectionHolder(conHolder, false);
  25.         return txObject;
  26.     }
  28.     /**
  29.      * 是否存在事务
  30.      *
  31.      * @param transaction transaction object returned by doGetTransaction
  32.      * @return
  33.      */
  34.     @Override
  35.     protected boolean isExistingTransaction(Object transaction) {
  36.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  37.         return (txObject.hasConnectionHolder() && txObject.getConnectionHolder()
  38.                 .isTransactionActive());
  39.     }
  42.     /**
  43.      * This implementation sets the isolation level but ignores the timeout. 事务的开始方法
  44.      */
  45.     @Override
  46.     protected void doBegin(Object transaction, TransactionDefinition definition) {
  47.         // 拿出事务
  48.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  49.         // 链接对象
  50.         Connection con = null;
  52.         try {
  53.             if (!txObject.hasConnectionHolder() ||
  54.                     txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
  55.                 // 数据库链接对象
  56.                 Connection newCon = obtainDataSource().getConnection();
  57.                 if (logger.isDebugEnabled()) {
  58.                     logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
  59.                 }
  60.                 // 设置数据库连接
  61.                 txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
  62.             }
  63.             // 拿出链接对象并且设置同步事务
  64.             txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
  65.             // 链接对象赋值
  66.             con = txObject.getConnectionHolder().getConnection();
  68.             // 获取事务级别
  69.             Integer previousIsolationLevel = DataSourceUtils
  70.                     .prepareConnectionForTransaction(con, definition);
  71.             // 设置事务隔离级别
  72.             txObject.setPreviousIsolationLevel(previousIsolationLevel);
  73.             // 设置只读
  74.             txObject.setReadOnly(definition.isReadOnly());
  76.             // Switch to manual commit if necessary. This is very expensive in some JDBC drivers,
  77.             // so we don't want to do it unnecessarily (for example if we've explicitly
  78.             // configured the connection pool to set it already).
  79.             // 判断是否自动提交
  80.             if (con.getAutoCommit()) {
  81.                 txObject.setMustRestoreAutoCommit(true);
  82.                 if (logger.isDebugEnabled()) {
  83.                     logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
  84.                 }
  85.                 con.setAutoCommit(false);
  86.             }
  88.             // 事务链接准备
  89.             prepareTransactionalConnection(con, definition);
  90.             // 事务激活
  91.             txObject.getConnectionHolder().setTransactionActive(true);
  93.             // 超时时间获取
  94.             int timeout = determineTimeout(definition);
  95.             // 默认超时时间设置
  96.             if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
  97.                 txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
  98.             }
  100.             // Bind the connection holder to the thread.
  101.             // 将链接和当前线程绑定
  102.             if (txObject.isNewConnectionHolder()) {
  103.                 // k: datasource v: connectionHolder
  104.                 TransactionSynchronizationManager
  105.                         .bindResource(obtainDataSource(), txObject.getConnectionHolder());
  106.             }
  107.         } catch (Throwable ex) {
  108.             if (txObject.isNewConnectionHolder()) {
  109.                 // 释放链接
  110.                 DataSourceUtils.releaseConnection(con, obtainDataSource());
  111.                 txObject.setConnectionHolder(null, false);
  112.             }
  113.             throw new CannotCreateTransactionException(
  114.                     "Could not open JDBC Connection for transaction", ex);
  115.         }
  116.     }
  119.     /**
  120.      * 挂起事务
  121.      *
  122.      * @param transaction transaction object returned by {@code doGetTransaction}
  123.      * @return 移除的链接
  124.      */
  125.     @Override
  126.     protected Object doSuspend(Object transaction) {
  127.         // 获取事务对象
  128.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  129.         // 连接置空
  130.         txObject.setConnectionHolder(null);
  131.         // 解除资源绑定
  132.         return TransactionSynchronizationManager.unbindResource(obtainDataSource());
  133.     }
  136.     /**
  137.      * 恢复事务
  138.      *
  139.      * @param transaction        transaction object returned by {@code doGetTransaction}
  140.      * @param suspendedResources the object that holds suspended resources, as returned by
  141.      *                           doSuspend
  142.      */
  143.     @Override
  144.     protected void doResume(@Nullable Object transaction, Object suspendedResources) {
  145.         // 资源绑定
  146.         TransactionSynchronizationManager.bindResource(obtainDataSource(), suspendedResources);
  147.     }
  150.     /**
  151.      * 做提交
  152.      *
  153.      * @param status the status representation of the transaction
  154.      */
  155.     @Override
  156.     protected void doCommit(DefaultTransactionStatus status) {
  157.         // 事务对象
  158.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) status
  159.                 .getTransaction();
  160.         // 获取链接
  161.         Connection con = txObject.getConnectionHolder().getConnection();
  162.         if (status.isDebug()) {
  163.             logger.debug("Committing JDBC transaction on Connection [" + con + "]");
  164.         }
  165.         try {
  166.             // 链接提交
  167.             con.commit();
  168.         } catch (SQLException ex) {
  169.             throw new TransactionSystemException("Could not commit JDBC transaction", ex);
  170.         }
  171.     }
  174.     /**
  175.      * 事务回滚
  176.      *
  177.      * @param status the status representation of the transaction
  178.      */
  179.     @Override
  180.     protected void doRollback(DefaultTransactionStatus status) {
  182.         // 事务对象
  183.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) status
  184.                 .getTransaction();
  185.         // 链接对象
  186.         Connection con = txObject.getConnectionHolder().getConnection();
  187.         if (status.isDebug()) {
  188.             logger.debug("Rolling back JDBC transaction on Connection [" + con + "]");
  189.         }
  190.         try {
  191.             // 回滚方法
  192.             con.rollback();
  193.         } catch (SQLException ex) {
  194.             throw new TransactionSystemException("Could not roll back JDBC transaction", ex);
  195.         }
  196.     }
  199.     /**
  200.      * 设置回滚
  201.      *
  202.      * @param status the status representation of the transaction
  203.      */
  204.     @Override
  205.     protected void doSetRollbackOnly(DefaultTransactionStatus status) {
  206.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) status
  207.                 .getTransaction();
  208.         if (status.isDebug()) {
  209.             logger.debug(
  210.                     "Setting JDBC transaction [" + txObject.getConnectionHolder().getConnection() +
  211.                             "] rollback-only");
  212.         }
  213.         txObject.setRollbackOnly();
  214.     }
  217.     /**
  218.      * 清除
  219.      *
  220.      * @param transaction transaction object returned by {@code doGetTransaction}
  221.      */
  222.     @Override
  223.     protected void doCleanupAfterCompletion(Object transaction) {
  224.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  226.         // Remove the connection holder from the thread, if exposed.
  227.         if (txObject.isNewConnectionHolder()) {
  228.             // 释放datasource绑定的资源
  229.             TransactionSynchronizationManager.unbindResource(obtainDataSource());
  230.         }
  232.         // Reset connection.
  233.         Connection con = txObject.getConnectionHolder().getConnection();
  234.         try {
  235.             if (txObject.isMustRestoreAutoCommit()) {
  236.                 con.setAutoCommit(true);
  237.             }
  238.             // 重置链接
  239.             DataSourceUtils.resetConnectionAfterTransaction(
  240.                     con, txObject.getPreviousIsolationLevel(), txObject.isReadOnly());
  241.         } catch (Throwable ex) {
  242.             logger.debug("Could not reset JDBC Connection after transaction", ex);
  243.         }
  245.         if (txObject.isNewConnectionHolder()) {
  246.             if (logger.isDebugEnabled()) {
  247.                 logger.debug("Releasing JDBC Connection [" + con + "] after transaction");
  248.             }
  249.             DataSourceUtils.releaseConnection(con, this.dataSource);
  250.         }
  252.         txObject.getConnectionHolder().clear();
  253.     }
  256.     /**
  257.     *
  258.     * 事务准备
  259.     */
  260.     protected void prepareTransactionalConnection(Connection con, TransactionDefinition definition)
  261.             throws SQLException {
  263.         if (isEnforceReadOnly() && definition.isReadOnly()) {
  264.             try (Statement stmt = con.createStatement()) {
  265.                 // 执行sql 类似事务隔离级别
  266.                 stmt.executeUpdate("SET TRANSACTION READ ONLY");
  267.             }
  268.         }
  269.     }

内部类 DataSourceTransactionObject

  1.     private static class DataSourceTransactionObject extends JdbcTransactionObjectSupport {
  3.         /**
  4.          * 是否有新的链接
  5.          */
  6.         private boolean newConnectionHolder;
  8.         /**
  9.          * 是否自动提交
  10.          */
  11.         private boolean mustRestoreAutoCommit;
  12.     }

AbstractPlatformTransactionManager

  • abstract 修饰具体定义的方法不具体展开。主要关注实现org.springframework.transaction.PlatformTransactionManager的几个方法

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.    }
  8.    // 事务状态
  9.    DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
  11.    if (defStatus.isLocalRollbackOnly()) {
  12.       if (defStatus.isDebug()) {
  13.          logger.debug("Transactional code has requested rollback");
  14.       }
  15.       // 处理回滚
  16.       processRollback(defStatus, false);
  17.       return;
  18.    }
  20.    if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
  21.       if (defStatus.isDebug()) {
  22.          logger.debug(
  23.                "Global transaction is marked as rollback-only but transactional code requested commit");
  24.       }
  25.       // 处理回滚
  26.       processRollback(defStatus, true);
  27.       return;
  28.    }
  29.    // 真正的处理提交
  30.    processCommit(defStatus);
  31. }
  1. private void processCommit(DefaultTransactionStatus status) throws TransactionException {
  2.    try {
  3.       boolean beforeCompletionInvoked = false;
  5.       try {
  6.          boolean unexpectedRollback = false;
  7.          //
  8.          prepareForCommit(status);
  9.          triggerBeforeCommit(status);
  10.          triggerBeforeCompletion(status);
  11.          // 前置任务是否已经执行
  12.          beforeCompletionInvoked = true;
  14.          // 嵌套事务. 是否有保存点
  15.          if (status.hasSavepoint()) {
  16.             if (status.isDebug()) {
  17.                logger.debug("Releasing transaction savepoint");
  18.             }
  19.             unexpectedRollback = status.isGlobalRollbackOnly();
  20.             status.releaseHeldSavepoint();
  21.          } else if (status.isNewTransaction()) {
  22.             if (status.isDebug()) {
  23.                logger.debug("Initiating transaction commit");
  24.             }
  25.             unexpectedRollback = status.isGlobalRollbackOnly();
  26.             doCommit(status);
  27.          } else if (isFailEarlyOnGlobalRollbackOnly()) {
  28.             unexpectedRollback = status.isGlobalRollbackOnly();
  29.          }
  31.          // Throw UnexpectedRollbackException if we have a global rollback-only
  32.          // marker but still didn't get a corresponding exception from commit.
  33.          if (unexpectedRollback) {
  34.             throw new UnexpectedRollbackException(
  35.                   "Transaction silently rolled back because it has been marked as rollback-only");
  36.          }
  37.       } catch (UnexpectedRollbackException ex) {
  38.          // can only be caused by doCommit
  39.          // 事务的同步状态: 回滚
  40.          triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
  41.          throw ex;
  42.       } catch (TransactionException ex) {
  43.          // can only be caused by doCommit
  44.          // 提交失败 做回滚
  45.          if (isRollbackOnCommitFailure()) {
  46.             doRollbackOnCommitException(status, ex);
  47.          } else {
  48.             // 事务的同步状态: 未知
  49.             triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
  50.          }
  51.          throw ex;
  52.       } catch (RuntimeException | Error ex) {
  53.          if (!beforeCompletionInvoked) {
  54.             triggerBeforeCompletion(status);
  55.          }
  56.          doRollbackOnCommitException(status, ex);
  57.          throw ex;
  58.       }
  60.       // Trigger afterCommit callbacks, with an exception thrown there
  61.       // propagated to callers but the transaction still considered as committed.
  62.       try {
  63.          triggerAfterCommit(status);
  64.       } finally {
  65.          triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
  66.       }
  68.    } finally {
  69.       // 完成后清理
  70.       cleanupAfterCompletion(status);
  71.    }
  72. }

rollback 方法

  1. @Override
  2. public final void rollback(TransactionStatus status) throws TransactionException {
  3.    // 是否已完成
  4.    if (status.isCompleted()) {
  5.       throw new IllegalTransactionStateException(
  6.             "Transaction is already completed - do not call commit or rollback more than once per transaction");
  7.    }
  9.    DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
  10.    // 执行回滚
  11.    processRollback(defStatus, false);
  12. }
  1. private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
  2.    try {
  3.       boolean unexpectedRollback = unexpected;
  5.       try {
  6.          triggerBeforeCompletion(status);
  8.          // 嵌套事务
  9.          if (status.hasSavepoint()) {
  10.             if (status.isDebug()) {
  11.                logger.debug("Rolling back transaction to savepoint");
  12.             }
  13.             // 回滚保存点
  14.             status.rollbackToHeldSavepoint();
  15.          }
  16.          // 独立事务
  17.          else if (status.isNewTransaction()) {
  18.             if (status.isDebug()) {
  19.                logger.debug("Initiating transaction rollback");
  20.             }
  21.             // 执行回滚
  22.             doRollback(status);
  23.          } else {
  24.             // Participating in larger transaction
  25.             if (status.hasTransaction()) {
  26.                if (status.isLocalRollbackOnly()
  27.                      || isGlobalRollbackOnParticipationFailure()) {
  28.                   if (status.isDebug()) {
  29.                      logger.debug(
  30.                            "Participating transaction failed - marking existing transaction as rollback-only");
  31.                   }
  32.                   // 设置回滚
  33.                   doSetRollbackOnly(status);
  34.                } else {
  35.                   if (status.isDebug()) {
  36.                      logger.debug(
  37.                            "Participating transaction failed - letting transaction originator decide on rollback");
  38.                   }
  39.                }
  40.             } else {
  41.                logger.debug(
  42.                      "Should roll back transaction but cannot - no transaction available");
  43.             }
  44.             // Unexpected rollback only matters here if we're asked to fail early
  45.             if (!isFailEarlyOnGlobalRollbackOnly()) {
  46.                unexpectedRollback = false;
  47.             }
  48.          }
  49.       } catch (RuntimeException | Error ex) {
  50.          triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
  51.          throw ex;
  52.       }
  54.       triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
  56.       // Raise UnexpectedRollbackException if we had a global rollback-only marker
  57.       if (unexpectedRollback) {
  58.          throw new UnexpectedRollbackException(
  59.                "Transaction rolled back because it has been marked as rollback-only");
  60.       }
  61.    } finally {
  62.       cleanupAfterCompletion(status);
  63.    }
  64. }

TransactionSynchronizationManager

  • 事务同步管理器

  • 一些基本属性

  1.     /**
  2.      * 资源
  3.      */
  4.     private static final ThreadLocal<Map<Object, Object>> resources =
  5.             new NamedThreadLocal<>("Transactional resources");
  6.     /**
  7.      * 同步器
  8.      */
  9.     private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =
  10.             new NamedThreadLocal<>("Transaction synchronizations");
  12.     /**
  13.      * 事务名称
  14.      */
  15.     private static final ThreadLocal<String> currentTransactionName =
  16.             new NamedThreadLocal<>("Current transaction name");
  18.     /**
  19.      * 是否只读
  20.      */
  21.     private static final ThreadLocal<Boolean> currentTransactionReadOnly =
  22.             new NamedThreadLocal<>("Current transaction read-only status");
  24.     /**
  25.      * 事务隔离级别
  26.      */
  27.     private static final ThreadLocal<Integer> currentTransactionIsolationLevel =
  28.             new NamedThreadLocal<>("Current transaction isolation level");
  30.     /**
  31.      * 事务激活状态
  32.      */
  33.     private static final ThreadLocal<Boolean> actualTransactionActive =
  34.             new NamedThreadLocal<>("Actual transaction active");

资源方法

获取资源
  1. public static Map<Object, Object> getResourceMap() {
  2.    // 线程变量中获取
  3.    Map<Object, Object> map = resources.get();
  4.    // 判空 如果为空给个空map如果有就返回
  5.    return (map != null ? Collections.unmodifiableMap(map) : Collections.emptyMap());
  6. }
判断是否存在资源
  1. public static boolean hasResource(Object key) {
  2.    // 资源key获取
  3.    // 通过 unwrapResourceIfNecessary 会走一次资源对象转换.
  4.    // 1. InfrastructureProxy
  5.    // 2. ScopedObject
  6.    Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
  7.    Object value = doGetResource(actualKey);
  8.    return (value != null);
  9. }

  • unwrapResourceIfNecessary方法会将资源具体化到接口，从接口中调用方法获取具体的资源

    1. static Object unwrapResourceIfNecessary(Object resource) {
    2.    Assert.notNull(resource, "Resource must not be null");
    3.    Object resourceRef = resource;
    4.    // unwrap infrastructure proxy
    5.    if (resourceRef instanceof InfrastructureProxy) {
    6.       resourceRef = ((InfrastructureProxy) resourceRef).getWrappedObject();
    7.    }
    8.    if (aopAvailable) {
    9.       // now unwrap scoped proxy
    10.       resourceRef = ScopedProxyUnwrapper.unwrapIfNecessary(resourceRef);
    11.    }
    12.    return resourceRef;
    13. }
    15.     private static class ScopedProxyUnwrapper {
    17.         public static Object unwrapIfNecessary(Object resource) {
    18.             if (resource instanceof ScopedObject) {
    19.                 return ((ScopedObject) resource).getTargetObject();
    20.             } else {
    21.                 return resource;
    22.             }
    23.         }
    24.     }

  • doGetResource 方法去获取资源

    1. @Nullable
    2. private static Object doGetResource(Object actualKey) {
    3.    Map<Object, Object> map = resources.get();
    4.    if (map == null) {
    5.       return null;
    6.    }
    7.    Object value = map.get(actualKey);
    8.    // Transparently remove ResourceHolder that was marked as void...
    9.    // 如果资源是下面两种的其中一个就删除这个资源
    10.    if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
    11.       map.remove(actualKey);
    12.       // Remove entire ThreadLocal if empty...
    13.       if (map.isEmpty()) {
    14.          resources.remove();
    15.       }
    16.       value = null;
    17.    }
    18.    return value;
    19. }
资源绑定
  1. public static void bindResource(Object key, Object value) throws IllegalStateException {
  2.    // 将资源转换为正真的key
  3.    Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
  4.    Assert.notNull(value, "Value must not be null");
  5.    Map<Object, Object> map = resources.get();
  6.    // set ThreadLocal Map if none found
  7.    // 资源对象为空初始化
  8.    if (map == null) {
  9.       map = new HashMap<>();
  10.       resources.set(map);
  11.    }
  12.    // 原来的值
  13.    Object oldValue = map.put(actualKey, value);
  14.    // Transparently suppress a ResourceHolder that was marked as void...
  15.    // 如果原来的值是下面的两种 抛出异常
  16.    if (oldValue instanceof ResourceHolder && ((ResourceHolder) oldValue).isVoid()) {
  17.       oldValue = null;
  18.    }
  19.    if (oldValue != null) {
  20.       throw new IllegalStateException("Already value [" + oldValue + "] for key [" +
  21.             actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
  22.    }
  23.    if (logger.isTraceEnabled()) {
  24.       logger.trace("Bound value [" + value + "] for key [" + actualKey + "] to thread [" +
  25.             Thread.currentThread().getName() + "]");
  26.    }
  27. }
  • debug 使用的是 druid 的数据源

image-20200729090322058

  • unwrapResourceIfNecessary 方法
  1. static Object unwrapResourceIfNecessary(Object resource) {
  2.    Assert.notNull(resource, "Resource must not be null");
  3.    Object resourceRef = resource;
  4.    // unwrap infrastructure proxy
  5.    if (resourceRef instanceof InfrastructureProxy) {
  6.       resourceRef = ((InfrastructureProxy) resourceRef).getWrappedObject();
  7.    }
  8.    if (aopAvailable) {
  9.       // now unwrap scoped proxy
  10.       resourceRef = ScopedProxyUnwrapper.unwrapIfNecessary(resourceRef);
  11.    }
  12.    return resourceRef;
  13. }

显然com.alibaba.druid.pool.DruidDataSource不是InfrastructureProxy

  • aopAvailable

    1. private static final boolean aopAvailable = ClassUtils.isPresent(
    2.       "org.springframework.aop.scope.ScopedObject",
    3.       TransactionSynchronizationUtils.class.getClassLoader());
    1. public static boolean isPresent(String className, @Nullable ClassLoader classLoader) {
    2.    try {
    3.       forName(className, classLoader);
    4.       return true;
    5.    }
    6.    catch (IllegalAccessError err) {
    7.       throw new IllegalStateException("Readability mismatch in inheritance hierarchy of class [" +
    8.             className + "]: " + err.getMessage(), err);
    9.    }
    10.    catch (Throwable ex) {
    11.       // Typically ClassNotFoundException or NoClassDefFoundError...
    12.       return false;
    13.    }
    14. }

    看是否可以解析如果解析成功返回true 解析失败返回false

  • ScopedProxyUnwrapper.unwrapIfNecessary

    1. private static class ScopedProxyUnwrapper {
    3.    public static Object unwrapIfNecessary(Object resource) {
    4.       if (resource instanceof ScopedObject) {
    5.          return ((ScopedObject) resource).getTargetObject();
    6.       } else {
    7.          return resource;
    8.       }
    9.    }
    10. }
    • com.alibaba.druid.pool.DruidDataSource不是ScopedObject 直接返回

后续就是一个map的put方法不具体展开

解除资源绑定
  1. public static Object unbindResource(Object key) throws IllegalStateException {
  2.    // 获取真正的资源对象
  3.    Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
  4.    // map 移除key
  5.    Object value = doUnbindResource(actualKey);
  6.    if (value == null) {
  7.       throw new IllegalStateException(
  8.             "No value for key [" + actualKey + "] bound to thread [" + Thread
  9.                   .currentThread().getName() + "]");
  10.    }
  11.    return value;
  12. }
  15.     @Nullable
  16.     private static Object doUnbindResource(Object actualKey) {
  17.         Map<Object, Object> map = resources.get();
  18.         if (map == null) {
  19.             return null;
  20.         }
  21.         Object value = map.remove(actualKey);
  22.         // Remove entire ThreadLocal if empty...
  23.         if (map.isEmpty()) {
  24.             resources.remove();
  25.         }
  26.         // Transparently suppress a ResourceHolder that was marked as void...
  27.         if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
  28.             value = null;
  29.         }
  30.         if (value != null && logger.isTraceEnabled()) {
  31.             logger.trace("Removed value [" + value + "] for key [" + actualKey + "] from thread [" +
  32.                     Thread.currentThread().getName() + "]");
  33.         }
  34.         return value;
  35.     }

map 对象的 remove 操作

其他

  • 其他几个都是使用ThreadLocal进行数据设置操作即可.

TransactionTemplate

  • 属性

    1.     @Nullable
    2.     private PlatformTransactionManager transactionManager;

    前文说到 DataSourceTransactionManager 实现了 PlatformTransactionManager 因此配置的时候我们有如下片段

    1. <bean id="transactionTemplate"
    2.       class="org.springframework.transaction.support.TransactionTemplate">
    3.    <property name="transactionManager" ref="transactionManager"/>
    4. </bean>

  • 事务操作模板类图

    image-20200728094658684

  • org.springframework.beans.factory.InitializingBean接口的实现

    1. @Override
    2. public void afterPropertiesSet() {
    3.         if (this.transactionManager == null) {
    4.             throw new IllegalArgumentException("Property 'transactionManager' is required");
    5.         }
    6.     }

execute

  1.    @Override
  2.    @Nullable
  3.    public <T> T execute(TransactionCallback<T> action) throws TransactionException {
  4.       Assert.state(this.transactionManager != null, "No PlatformTransactionManager set");
  6.       // 事务管理是否是 xxx接口
  7.       if (this.transactionManager instanceof CallbackPreferringPlatformTransactionManager) {
  8. //       强转执行
  9.          return ((CallbackPreferringPlatformTransactionManager) this.transactionManager)
  10.                .execute(this, action);
  11.       } else {
  12.          // 获取事务状态
  13.          TransactionStatus status = this.transactionManager.getTransaction(this);
  14.          // 返回结果
  15.          T result;
  16.          try {
  17.             // 事务回调执行
  18.             result = action.doInTransaction(status);
  19.          } catch (RuntimeException | Error ex) {
  20.             // Transactional code threw application exception -> rollback
  21.             // 回滚异常
  22.             rollbackOnException(status, ex);
  23.             throw ex;
  24.          } catch (Throwable ex) {
  25.             // Transactional code threw unexpected exception -> rollback
  26.             // 回滚异常
  27.             rollbackOnException(status, ex);
  28.             throw new UndeclaredThrowableException(ex,
  29.                   "TransactionCallback threw undeclared checked exception");
  30.          }
  31.          // 提交
  32.          this.transactionManager.commit(status);
  33.          return result;
  34.       }
  35.    }