Spring源码解析——事务增强器

徐锦洪  金牌会员 | 2023-11-5 00:39:45 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 857|帖子 857|积分 2571

正文
上一篇文章我们讲解了事务的Advisor是如何注册进Spring容器的，也讲解了Spring是如何将有配置事务的类配置上事务的，实际上也就是用了AOP那一套，也讲解了Advisor，pointcut验证流程，至此，事务的初始化工作都已经完成了，在之后的调用过程，如果代理类的方法被调用，都会调用BeanFactoryTransactionAttributeSourceAdvisor这个Advisor的增强方法，也就是我们还未提到的那个Advisor里面的advise，还记得吗，在自定义标签的时候我们将TransactionInterceptor这个Advice作为bean注册进IOC容器，并且将其注入进Advisor中，这个Advice在代理类的invoke方法中会被封装到拦截器链中，最终事务的功能都在advise中体现，所以我们先来关注一下TransactionInterceptor这个类吧。最全面的Java面试网站
TransactionInterceptor类继承自MethodInterceptor，所以调用该类是从其invoke方法开始的，首先预览下这个方法：
  1. @Override
  2. @Nullable
  3. public Object invoke(final 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. }
复制代码
重点来了，进入invokeWithinTransaction方法：
  1. @Nullable
  2. protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
  3.         final InvocationCallback invocation) throws Throwable {
  5.     // If the transaction attribute is null, the method is non-transactional.
  6.     TransactionAttributeSource tas = getTransactionAttributeSource();
  7.     // 获取对应事务属性
  8.     final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
  9.     // 获取beanFactory中的transactionManager
  10.     final PlatformTransactionManager tm = determineTransactionManager(txAttr);
  11.     // 构造方法唯一标识（类.方法，如：service.UserServiceImpl.save）
  12.     final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
  14.     // 声明式事务处理
  15.     if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
  16.         // 创建TransactionInfo
  17.         TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
  18.         Object retVal = null;
  19.         try {
  20.             // 执行原方法
  21.             // 继续调用方法拦截器链,这里一般将会调用目标类的方法,如:AccountServiceImpl.save方法
  22.             retVal = invocation.proceedWithInvocation();
  23.         }
  24.         catch (Throwable ex) {
  25.             // 异常回滚
  26.             completeTransactionAfterThrowing(txInfo, ex);
  27.             // 手动向上抛出异常，则下面的提交事务不会执行
  28.             // 如果子事务出异常，则外层事务代码需catch住子事务代码，不然外层事务也会回滚
  29.             throw ex;
  30.         }
  31.         finally {
  32.             // 消除信息
  33.             cleanupTransactionInfo(txInfo);
  34.         }
  35.         // 提交事务
  36.         commitTransactionAfterReturning(txInfo);
  37.         return retVal;
  38.     }
  40.     else {
  41.         final ThrowableHolder throwableHolder = new ThrowableHolder();
  42.         try {
  43.             // 编程式事务处理
  44.             Object result = ((CallbackPreferringPlatformTransactionManager) tm).execute(txAttr, status -> {
  45.                 TransactionInfo txInfo = prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
  46.                 try {
  47.                     return invocation.proceedWithInvocation();
  48.                 }
  49.                 catch (Throwable ex) {
  50.                     if (txAttr.rollbackOn(ex)) {
  51.                         // A RuntimeException: will lead to a rollback.
  52.                         if (ex instanceof RuntimeException) {
  53.                             throw (RuntimeException) ex;
  54.                         }
  55.                         else {
  56.                             throw new ThrowableHolderException(ex);
  57.                         }
  58.                     }
  59.                     else {
  60.                         // A normal return value: will lead to a commit.
  61.                         throwableHolder.throwable = ex;
  62.                         return null;
  63.                     }
  64.                 }
  65.                 finally {
  66.                     cleanupTransactionInfo(txInfo);
  67.                 }
  68.             });
  70.             // Check result state: It might indicate a Throwable to rethrow.
  71.             if (throwableHolder.throwable != null) {
  72.                 throw throwableHolder.throwable;
  73.             }
  74.             return result;
  75.         }
  76.         catch (ThrowableHolderException ex) {
  77.             throw ex.getCause();
  78.         }
  79.         catch (TransactionSystemException ex2) {
  80.             if (throwableHolder.throwable != null) {
  81.                 logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
  82.                 ex2.initApplicationException(throwableHolder.throwable);
  83.             }
  84.             throw ex2;
  85.         }
  86.         catch (Throwable ex2) {
  87.             if (throwableHolder.throwable != null) {
  88.                 logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
  89.             }
  90.             throw ex2;
  91.         }
  92.     }
  93. }
复制代码
创建事务Info对象

我们先分析事务创建的过程。
  1. protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
  2.         @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
  4.     // If no name specified, apply method identification as transaction name.
  5.     // 如果没有名称指定则使用方法唯一标识，并使用DelegatingTransactionAttribute封装txAttr
  6.     if (txAttr != null && txAttr.getName() == null) {
  7.         txAttr = new DelegatingTransactionAttribute(txAttr) {
  8.             @Override
  9.             public String getName() {
  10.                 return joinpointIdentification;
  11.             }
  12.         };
  13.     }
  15.     TransactionStatus status = null;
  16.     if (txAttr != null) {
  17.         if (tm != null) {
  18.             // 获取TransactionStatus
  19.             status = tm.getTransaction(txAttr);
  20.         }
  21.         else {
  22.             if (logger.isDebugEnabled()) {
  23.                 logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
  24.                         "] because no transaction manager has been configured");
  25.             }
  26.         }
  27.     }
  28.     // 根据指定的属性与status准备一个TransactionInfo
  29.     return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
  30. }
复制代码
对于createTransactionlfNecessary函数主要做了这样几件事情。
（1）使用 DelegatingTransactionAttribute 封装传入的 TransactionAttribute 实例。
对于传入的TransactionAttribute类型的参数txAttr，当前的实际类型是RuleBasedTransactionAttribute，是由获取事务属性时生成，主要用于数据承载，而这里之所以使用DelegatingTransactionAttribute进行封装，当然是提供了更多的功能。
（2）获取事务。
事务处理当然是以事务为核心，那么获取事务就是最重要的事情。
（3）构建事务信息。
根据之前几个步骤获取的信息构建Transactionlnfo并返回。
分享一份大彬精心整理的大厂面试手册，包含计算机基础、Java基础、多线程、JVM、数据库、Redis、Spring、Mybatis、SpringMVC、SpringBoot、分布式、微服务、设计模式、架构、校招社招分享等高频面试题，非常实用，有小伙伴靠着这份手册拿过字节offer~


需要的小伙伴可以自行下载：
http://mp.weixin.qq.com/s?__biz=Mzg2OTY1NzY0MQ==&mid=2247485445&idx=1&sn=1c6e224b9bb3da457f5ee03894493dbc&chksm=ce98f543f9ef7c55325e3bf336607a370935a6c78dbb68cf86e59f5d68f4c51d175365a189f8#rd
获取事务

其中核心是在getTransaction方法中：
  1. @Override
  2. public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
  3.     // 获取一个transaction
  4.     Object transaction = doGetTransaction();
  6.     boolean debugEnabled = logger.isDebugEnabled();
  8.     if (definition == null) {
  9.         definition = new DefaultTransactionDefinition();
  10.     }
  11.     // 如果在这之前已经存在事务了，就进入存在事务的方法中
  12.     if (isExistingTransaction(transaction)) {
  13.         return handleExistingTransaction(definition, transaction, debugEnabled);
  14.     }
  15.    
  16.     // 事务超时设置验证
  17.     if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
  18.         throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
  19.     }
  21.     // 走到这里说明此时没有存在事务，如果传播特性是MANDATORY时抛出异常
  22.     if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
  23.         throw new IllegalTransactionStateException(
  24.                 "No existing transaction found for transaction marked with propagation 'mandatory'");
  25.     }
  26.     // 如果此时不存在事务，当传播特性是REQUIRED或NEW或NESTED都会进入if语句块
  27.     else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
  28.             definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
  29.             definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
  30.         // PROPAGATION_REQUIRED、PROPAGATION_REQUIRES_NEW、PROPAGATION_NESTED都需要新建事务
  31.         // 因为此时不存在事务，将null挂起
  32.         SuspendedResourcesHolder suspendedResources = suspend(null);
  33.         if (debugEnabled) {
  34.             logger.debug("Creating new transaction with name [" + definition.getName() + "]: " + definition);
  35.         }
  36.         try {
  37.             boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
  38.             // new一个status，存放刚刚创建的transaction，然后将其标记为新事务！
  39.             // 这里transaction后面一个参数决定是否是新事务！
  40.             DefaultTransactionStatus status = newTransactionStatus(
  41.                     definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
  42.             // 新开一个连接的地方，非常重要
  43.             doBegin(transaction, definition);
  44.             prepareSynchronization(status, definition);
  45.             return status;
  46.         }
  47.         catch (RuntimeException | Error ex) {
  48.             resume(null, suspendedResources);
  49.             throw ex;
  50.         }
  51.     }
  52.     else {
  53.         // Create "empty" transaction: no actual transaction, but potentially synchronization.
  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.         // 其他的传播特性一律返回一个空事务，transaction = null
  59.         //当前不存在事务，且传播机制=PROPAGATION_SUPPORTS/PROPAGATION_NOT_SUPPORTED/PROPAGATION_NEVER，这三种情况，创建“空”事务
  60.         boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
  61.         return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null);
  62.     }
  63. }
复制代码
先来看看transaction是如何被创建出来的：
  1. @Override
  2. protected Object doGetTransaction() {
  3.     // 这里DataSourceTransactionObject是事务管理器的一个内部类
  4.     // DataSourceTransactionObject就是一个transaction，这里new了一个出来
  5.     DataSourceTransactionObject txObject = new DataSourceTransactionObject();
  6.     txObject.setSavepointAllowed(isNestedTransactionAllowed());
  7.     // 解绑与绑定的作用在此时体现，如果当前线程有绑定的话，将会取出holder
  8.     // 第一次conHolder肯定是null
  9.     ConnectionHolder conHolder =
  10.     (ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
  11.     // 此时的holder被标记成一个旧holder
  12.     txObject.setConnectionHolder(conHolder, false);
  13.     return txObject;
  14. }
复制代码
创建transaction过程很简单，接着就会判断当前是否存在事务：
  1. @Override
  2. protected boolean isExistingTransaction(Object transaction) {
  3.     DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  4.     return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
  5. }
  7. public boolean hasConnectionHolder() {
  8.     return (this.connectionHolder != null);
  9. }
复制代码
这里判断是否存在事务的依据主要是获取holder中的transactionActive变量是否为true，如果是第一次进入事务，holder直接为null判断不存在了，如果是第二次进入事务transactionActive变量是为true的（后面会提到是在哪里把它变成true的），由此来判断当前是否已经存在事务了。
至此，源码分成了2条处理线：
1.当前已存在事务：isExistingTransaction()判断是否存在事务，存在事务handleExistingTransaction()根据不同传播机制不同处理
2.当前不存在事务: 不同传播机制不同处理
当前不存在事务

如果不存在事务，传播特性又是REQUIRED或NEW或NESTED，将会先挂起null，这个挂起方法我们后面再讲，然后创建一个DefaultTransactionStatus ，并将其标记为新事务，然后执行doBegin(transaction, definition);这个方法也是一个关键方法
神秘又关键的status对象

TransactionStatus接口
  1. public interface TransactionStatus extends SavepointManager, Flushable {
  2.     // 返回当前事务是否为新事务（否则将参与到现有事务中，或者可能一开始就不在实际事务中运行）
  3.     boolean isNewTransaction();
  4.     // 返回该事务是否在内部携带保存点，也就是说，已经创建为基于保存点的嵌套事务。
  5.     boolean hasSavepoint();
  6.     // 设置事务仅回滚。
  7.     void setRollbackOnly();
  8.     // 返回事务是否已标记为仅回滚
  9.     boolean isRollbackOnly();
  10.     // 将会话刷新到数据存储区
  11.     @Override
  12.     void flush();
  13.     // 返回事物是否已经完成，无论提交或者回滚。
  14.     boolean isCompleted();
  15. }
复制代码
再来看看实现类DefaultTransactionStatus
DefaultTransactionStatus
  1. public class DefaultTransactionStatus extends AbstractTransactionStatus {
  3.     //事务对象
  4.     @Nullable
  5.     private final Object transaction;
  7.     //事务对象
  8.     private final boolean newTransaction;
  10.     private final boolean newSynchronization;
  12.     private final boolean readOnly;
  14.     private final boolean debug;
  16.     //事务对象
  17.     @Nullable
  18.     private final Object suspendedResources;
  19.    
  20.         public DefaultTransactionStatus(
  21.             @Nullable Object transaction, boolean newTransaction, boolean newSynchronization,
  22.             boolean readOnly, boolean debug, @Nullable Object suspendedResources) {
  24.         this.transaction = transaction;
  25.         this.newTransaction = newTransaction;
  26.         this.newSynchronization = newSynchronization;
  27.         this.readOnly = readOnly;
  28.         this.debug = debug;
  29.         this.suspendedResources = suspendedResources;
  30.     }
  31.    
  32.     //略...
  33. }
复制代码
我们看看这行代码 DefaultTransactionStatus status = newTransactionStatus( definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
  1. // 这里是构造一个status对象的方法
  2. protected DefaultTransactionStatus newTransactionStatus(
  3.     TransactionDefinition definition, @Nullable Object transaction, boolean newTransaction,
  4.     boolean newSynchronization, boolean debug, @Nullable Object suspendedResources) {
  6.     boolean actualNewSynchronization = newSynchronization &&
  7.         !TransactionSynchronizationManager.isSynchronizationActive();
  8.     return new DefaultTransactionStatus(
  9.         transaction, newTransaction, actualNewSynchronization,
  10.         definition.isReadOnly(), debug, suspendedResources);
  11. }
复制代码
实际上就是封装了事务属性definition，新创建的transaction，并且将事务状态属性设置为新事务，最后一个参数为被挂起的事务。
简单了解一下关键参数即可：
第二个参数transaction：事务对象，在一开头就有创建，其就是事务管理器的一个内部类。
第三个参数newTransaction：布尔值，一个标识，用于判断是否是新的事务，用于提交或者回滚方法中，是新的才会提交或者回滚。
最后一个参数suspendedResources：被挂起的对象资源，挂起操作会返回旧的holder，将其与一些事务属性一起封装成一个对象，就是这个suspendedResources这个对象了，它会放在status中，在最后的清理工作方法中判断status中是否有这个挂起对象，如果有会恢复它
接着我们来看看关键代码 doBegin(transaction, definition);
  1. @Override
  2. protected void doBegin(Object transaction, TransactionDefinition definition) {
  3.      DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  4.      Connection con = null;
  6.      try {
  7.          // 判断如果transaction没有holder的话，才去从dataSource中获取一个新连接
  8.          if (!txObject.hasConnectionHolder() ||
  9.                  txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
  10.              //通过dataSource获取连接
  11.              Connection newCon = this.dataSource.getConnection();
  12.              if (logger.isDebugEnabled()) {
  13.                  logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
  14.              }
  15.              // 所以，只有transaction中的holder为空时，才会设置为新holder
  16.              // 将获取的连接封装进ConnectionHolder，然后封装进transaction的connectionHolder属性
  17.              txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
  18.          }
  19. 　　　　　//设置新的连接为事务同步中
  20.          txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
  21.          con = txObject.getConnectionHolder().getConnection();
  22. 　　　　 //conn设置事务隔离级别,只读
  23.          Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
  24.          txObject.setPreviousIsolationLevel(previousIsolationLevel);//DataSourceTransactionObject设置事务隔离级别
  26.          // 如果是自动提交切换到手动提交
  27.          if (con.getAutoCommit()) {
  28.              txObject.setMustRestoreAutoCommit(true);
  29.              if (logger.isDebugEnabled()) {
  30.                  logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
  31.              }
  32.              con.setAutoCommit(false);
  33.          }
  34. 　　　　　// 如果只读，执行sql设置事务只读
  35.          prepareTransactionalConnection(con, definition);
  36.          // 设置connection持有者的事务开启状态
  37.          txObject.getConnectionHolder().setTransactionActive(true);
  39.          int timeout = determineTimeout(definition);
  40.          if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
  41.              // 设置超时秒数
  42.              txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
  43.          }
  45.          // 将当前获取到的连接绑定到当前线程
  46.          if (txObject.isNewConnectionHolder()) {
  47.              TransactionSynchronizationManager.bindResource(getDataSource(), txObject.getConnectionHolder());
  48.          }
  49.      }catch (Throwable ex) {
  50.          if (txObject.isNewConnectionHolder()) {
  51.              DataSourceUtils.releaseConnection(con, this.dataSource);
  52.              txObject.setConnectionHolder(null, false);
  53.          }
  54.          throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
  55.      }
  56. }
复制代码
conn设置事务隔离级别
  1. @Nullable
  2. public static Integer prepareConnectionForTransaction(Connection con, @Nullable TransactionDefinition definition)
  3.         throws SQLException {
  5.     Assert.notNull(con, "No Connection specified");
  7.     // Set read-only flag.
  8.     // 设置数据连接的只读标识
  9.     if (definition != null && definition.isReadOnly()) {
  10.         try {
  11.             if (logger.isDebugEnabled()) {
  12.                 logger.debug("Setting JDBC Connection [" + con + "] read-only");
  13.             }
  14.             con.setReadOnly(true);
  15.         }
  16.         catch (SQLException | RuntimeException ex) {
  17.             Throwable exToCheck = ex;
  18.             while (exToCheck != null) {
  19.                 if (exToCheck.getClass().getSimpleName().contains("Timeout")) {
  20.                     // Assume it's a connection timeout that would otherwise get lost: e.g. from JDBC 4.0
  21.                     throw ex;
  22.                 }
  23.                 exToCheck = exToCheck.getCause();
  24.             }
  25.             // "read-only not supported" SQLException -> ignore, it's just a hint anyway
  26.             logger.debug("Could not set JDBC Connection read-only", ex);
  27.         }
  28.     }
  30.     // Apply specific isolation level, if any.
  31.     // 设置数据库连接的隔离级别
  32.     Integer previousIsolationLevel = null;
  33.     if (definition != null && definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
  34.         if (logger.isDebugEnabled()) {
  35.             logger.debug("Changing isolation level of JDBC Connection [" + con + "] to " +
  36.                     definition.getIsolationLevel());
  37.         }
  38.         int currentIsolation = con.getTransactionIsolation();
  39.         if (currentIsolation != definition.getIsolationLevel()) {
  40.             previousIsolationLevel = currentIsolation;
  41.             con.setTransactionIsolation(definition.getIsolationLevel());
  42.         }
  43.     }
  45.     return previousIsolationLevel;
  46. }
复制代码
我们看到都是通过 Connection 去设置
线程变量的绑定

我们看 doBegin 方法的47行，将当前获取到的连接绑定到当前线程，绑定与解绑围绕一个线程变量，此变量在TransactionSynchronizationManager类中：
  1. private static final ThreadLocal<Map<Object, Object>> resources =  new NamedThreadLocal<>("Transactional resources");
复制代码
这是一个 static final 修饰的 线程变量，存储的是一个Map，我们来看看47行的静态方法，bindResource
  1. public static void bindResource(Object key, Object value) throws IllegalStateException {
  2.     // 从上面可知，线程变量是一个Map，而这个Key就是dataSource
  3.     // 这个value就是holder
  4.     Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
  5.     Assert.notNull(value, "Value must not be null");
  6.     // 获取这个线程变量Map
  7.     Map<Object, Object> map = resources.get();
  8.     // set ThreadLocal Map if none found
  9.     if (map == null) {
  10.         map = new HashMap<>();
  11.         resources.set(map);
  12.     }
  13.     // 将新的holder作为value，dataSource作为key放入当前线程Map中
  14.     Object oldValue = map.put(actualKey, value);
  15.     // Transparently suppress a ResourceHolder that was marked as void...
  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.     }    Thread.currentThread().getName() + "]");
  23.     }
  24.     // 略...
  25. }
复制代码
扩充知识点

这里再扩充一点，mybatis中获取的数据库连接，就是根据 dataSource 从ThreadLocal中获取的
以查询举例，会调用Executor#doQuery方法：

最终会调用DataSourceUtils#doGetConnection获取，真正的数据库连接，其中TransactionSynchronizationManager中保存的就是方法调用前，spring增强方法中绑定到线程的connection，从而保证整个事务过程中connection的一致性


我们看看TransactionSynchronizationManager.getResource(Object key)这个方法
  1. @Nullable
  2. public static Object getResource(Object key) {
  3.     Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
  4.     Object value = doGetResource(actualKey);
  5.     if (value != null && logger.isTraceEnabled()) {
  6.         logger.trace("Retrieved value [" + value + "] for key [" + actualKey + "] bound to thread [" +
  7.                 Thread.currentThread().getName() + "]");
  8.     }
  9.     return value;
  10. }
  12.     @Nullable
  13. private static Object doGetResource(Object actualKey) {
  14.     Map<Object, Object> map = resources.get();
  15.     if (map == null) {
  16.         return null;
  17.     }
  18.     Object value = map.get(actualKey);
  19.     // Transparently remove ResourceHolder that was marked as void...
  20.     if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
  21.         map.remove(actualKey);
  22.         // Remove entire ThreadLocal if empty...
  23.         if (map.isEmpty()) {
  24.             resources.remove();
  25.         }
  26.         value = null;
  27.     }
  28.     return value;
  29. }
复制代码
就是从线程变量的Map中根据 DataSource获取 ConnectionHolder
已经存在的事务

前面已经提到，第一次事务开始时必会新创一个holder然后做绑定操作，此时线程变量是有holder的且avtive为true，如果第二个事务进来，去new一个transaction之后去线程变量中取holder，holder是不为空的且active是为true的，所以会进入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.          // 这里会将原来的事务挂起,并返回被挂起的对象
  15.          Object suspendedResources = suspend(transaction);
  16.          boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
  17.          // 这里可以看到，第二个参数transaction传了一个空事务，第三个参数false为旧标记
  18.          // 最后一个参数就是将前面挂起的对象封装进新的Status中，当前事务执行完后，就恢复suspendedResources
  19.          return prepareTransactionStatus(definition, null, false, newSynchronization, debugEnabled, suspendedResources);
  20.      }
  21. 　　 // 3.REQUIRES_NEW挂起当前事务，创建新事务
  22.      if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
  23.          if (debugEnabled) {
  24.              logger.debug("Suspending current transaction, creating new transaction with name [" +
  25.                      definition.getName() + "]");
  26.          }
  27.          // 将原事务挂起，此时新建事务，不与原事务有关系
  28.          // 会将transaction中的holder设置为null，然后解绑！
  29.          SuspendedResourcesHolder suspendedResources = suspend(transaction);
  30.          try {
  31.              boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
  32.              // new一个status出来，传入transaction，并且为新事务标记，然后传入挂起事务
  33.              DefaultTransactionStatus status = newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
  34.              // 这里也做了一次doBegin，此时的transaction中holer是为空的，因为之前的事务被挂起了
  35.              // 所以这里会取一次新的连接，并且绑定！
  36.              doBegin(transaction, definition);
  37.              prepareSynchronization(status, definition);
  38.              return status;
  39.          }
  40.          catch (RuntimeException beginEx) {
  41.              resumeAfterBeginException(transaction, suspendedResources, beginEx);
  42.              throw beginEx;
  43.          }
  44.          catch (Error beginErr) {
  45.              resumeAfterBeginException(transaction, suspendedResources, beginErr);
  46.              throw beginErr;
  47.          }
  48.      }
  49. 　　// 如果此时的传播特性是NESTED，不会挂起事务
  50.      if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
  51.          if (!isNestedTransactionAllowed()) {
  52.              throw new NestedTransactionNotSupportedException(
  53.                      "Transaction manager does not allow nested transactions by default - " +
  54.                      "specify 'nestedTransactionAllowed' property with value 'true'");
  55.          }
  56.          if (debugEnabled) {
  57.              logger.debug("Creating nested transaction with name [" + definition.getName() + "]");
  58.          }
  59.          // 这里如果是JTA事务管理器，就不可以用savePoint了，将不会进入此方法
  60.          if (useSavepointForNestedTransaction()) {
  61.              // 这里不会挂起事务，说明NESTED的特性是原事务的子事务而已
  62.              // new一个status，传入transaction，传入旧事务标记，传入挂起对象=null
  63.              DefaultTransactionStatus status =prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
  64.              // 这里是NESTED特性特殊的地方，在先前存在事务的情况下会建立一个savePoint
  65.              status.createAndHoldSavepoint();
  66.              return status;
  67.          }
  68.          else {
  69.              // JTA事务走这个分支，创建新事务
  70.              boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
  71.              DefaultTransactionStatus status = newTransactionStatus(
  72.                      definition, transaction, true, newSynchronization, debugEnabled, null);
  73.              doBegin(transaction, definition);
  74.              prepareSynchronization(status, definition);
  75.              return status;
  76.          }
  77.      }
  79.      // 到这里PROPAGATION_SUPPORTS 或 PROPAGATION_REQUIRED或PROPAGATION_MANDATORY，存在事务加入事务即可，标记为旧事务，空挂起
  80.      boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
  81.      return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
  82. }
复制代码
对于已经存在事务的处理过程中，我们看到了很多熟悉的操作，但是，也有些不同的地方，函数中对已经存在的事务处理考虑两种情况。
（1）PROPAGATION_REQUIRES_NEW表示当前方法必须在它自己的事务里运行，一个新的事务将被启动，而如果有一个事务正在运行的话，则在这个方法运行期间被挂起。而Spring中对于此种传播方式的处理与新事务建立最大的不同点在于使用suspend方法将原事务挂起。 将信息挂起的目的当然是为了在当前事务执行完毕后在将原事务还原。
（2）PROPAGATION_NESTED表示如果当前正有一个事务在运行中，则该方法应该运行在一个嵌套的事务中，被嵌套的事务可以独立于封装事务进行提交或者回滚，如果封装事务不存在，行为就像PROPAGATION_REQUIRES_NEW。对于嵌入式事务的处理，Spring中主要考虑了两种方式的处理。

  • Spring中允许嵌入事务的时候，则首选设置保存点的方式作为异常处理的回滚。
  • 对于其他方式，比如JTA无法使用保存点的方式，那么处理方式与PROPAGATION_ REQUIRES_NEW相同，而一旦出现异常，则由Spring的事务异常处理机制去完成后续操作。
对于挂起操作的主要目的是记录原有事务的状态，以便于后续操作对事务的恢复
小结

到这里我们可以知道，在当前存在事务的情况下，根据传播特性去决定是否为新事务，是否挂起当前事务。
NOT_SUPPORTED ：会挂起事务，不运行doBegin方法传空transaction，标记为旧事务。封装status对象：
  1. return prepareTransactionStatus(definition, null, false, newSynchronization, debugEnabled, suspendedResources)
复制代码
REQUIRES_NEW ：将会挂起事务且运行doBegin方法，标记为新事务。封装status对象：
  1. DefaultTransactionStatus status = newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
复制代码
NESTED ：不会挂起事务且不会运行doBegin方法，标记为旧事务，但会创建savePoint。封装status对象：
  1. DefaultTransactionStatus status =prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
复制代码
其他事务例如REQUIRED ：不会挂起事务，封装原有的transaction不会运行doBegin方法，标记旧事务，封装status对象：
  1. return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
复制代码
挂起

对于挂起操作的主要目的是记录原有事务的状态，以便于后续操作对事务的恢复：
  1. @Nullable
  2. protected final SuspendedResourcesHolder suspend(@Nullable Object transaction) throws TransactionException {
  3.     if (TransactionSynchronizationManager.isSynchronizationActive()) {
  4.         List<TransactionSynchronization> suspendedSynchronizations = doSuspendSynchronization();
  5.         try {
  6.             Object suspendedResources = null;
  7.             if (transaction != null) {
  8.                 // 这里是真正做挂起的方法，这里返回的是一个holder
  9.                 suspendedResources = doSuspend(transaction);
  10.             }
  11.             // 这里将名称、隔离级别等信息从线程变量中取出并设置对应属性为null到线程变量
  12.             String name = TransactionSynchronizationManager.getCurrentTransactionName();
  13.             TransactionSynchronizationManager.setCurrentTransactionName(null);
  14.             boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();
  15.             TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);
  16.             Integer isolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
  17.             TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(null);
  18.             boolean wasActive = TransactionSynchronizationManager.isActualTransactionActive();
  19.             TransactionSynchronizationManager.setActualTransactionActive(false);
  20.             // 将事务各个属性与挂起的holder一并封装进SuspendedResourcesHolder对象中
  21.             return new SuspendedResourcesHolder(
  22.                 suspendedResources, suspendedSynchronizations, name, readOnly, isolationLevel, wasActive);
  23.         }
  24.         catch (RuntimeException | Error ex) {
  25.             // doSuspend failed - original transaction is still active...
  26.             doResumeSynchronization(suspendedSynchronizations);
  27.             throw ex;
  28.         }
  29.     }
  30.     else if (transaction != null) {
  31.         // Transaction active but no synchronization active.
  32.         Object suspendedResources = doSuspend(transaction);
  33.         return new SuspendedResourcesHolder(suspendedResources);
  34.     }
  35.     else {
  36.         // Neither transaction nor synchronization active.
  37.         return null;
  38.     }
  39. }
复制代码
  1. @Override
  2. protected Object doSuspend(Object transaction) {
  3.     DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  4.     // 将transaction中的holder属性设置为空
  5.     txObject.setConnectionHolder(null);
  6.     // ConnnectionHolder从线程变量中解绑！
  7.     return TransactionSynchronizationManager.unbindResource(obtainDataSource());
  8. }
复制代码
我们来看看 unbindResource
  1. private static Object doUnbindResource(Object actualKey) {
  2.     // 取得当前线程的线程变量Map
  3.     Map<Object, Object> map = resources.get();
  4.     if (map == null) {
  5.         return null;
  6.     }
  7.     // 将key为dataSourece的value移除出Map，然后将旧的Holder返回
  8.     Object value = map.remove(actualKey);
  9.     // Remove entire ThreadLocal if empty...
  10.     // 如果此时map为空，直接清除线程变量
  11.     if (map.isEmpty()) {
  12.         resources.remove();
  13.     }
  14.     // Transparently suppress a ResourceHolder that was marked as void...
  15.     if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
  16.         value = null;
  17.     }
  18.     if (value != null && logger.isTraceEnabled()) {
  19.         logger.trace("Removed value [" + value + "] for key [" + actualKey + "] from thread [" +
  20.                      Thread.currentThread().getName() + "]");
  21.     }
  22.     // 将旧Holder返回
  23.     return value;
  24. }
复制代码
可以回头看一下解绑操作的介绍。这里挂起主要干了三件事：

  • 将transaction中的holder属性设置为空
  • 解绑（会返回线程中的那个旧的holder出来，从而封装到SuspendedResourcesHolder对象中）
  • 将SuspendedResourcesHolder放入status中，方便后期子事务完成后，恢复外层事务
最后给大家分享一个Github仓库，上面有大彬整理的300多本经典的计算机书籍PDF，包括C语言、C++、Java、Python、前端、数据库、操作系统、计算机网络、数据结构和算法、机器学习、编程人生等，可以star一下，下次找书直接在上面搜索，仓库持续更新中~


Github地址
如果访问不了Github，可以访问码云地址。
码云地址

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

徐锦洪

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表