spring：声明式事务源码

本文主要介绍spring事务（JDBC单数据库事务管理）的核心源码实现

1、核心类介绍

我们可以先看一下核心类图。

  • PlatformTransactionManager
    底层接口，定义事务方法
    1
    2
    3
    4
    5
    public interface PlatformTransactionManager extends TransactionManager {  
        TransactionStatus getTransaction(@Nullable TransactionDefinition definition) 
        void commit(TransactionStatus status) throws TransactionException;
        void rollback(TransactionStatus status) throws TransactionException;
    }
  • AbstractPlatformTransactionManager
    抽象类，实现PlatformTransactionManager接口，并定义了一些抽象方法比如doCommit、doRollback留给子类实现（这里使用了设计模式中的模板模式）。
  • DataSourceTransactionmanager
    JDBC单数据库事务管理器，基于Connection实现
  • JtaTransactionManager
    分布式事务管理器（本文不展开介绍）
  • KafkaTransactionManager
    Kafka事务管理器（本文不展开介绍）

2、核心方法实现

2.1、getTransaction获取事务

  • AbstractPlatformTransactionManager.getTransaction实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@Override  
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)  
       throws TransactionException {  
  
    // Use defaults if no transaction definition given.  
    TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());  
  
    //模板模式，由子类实现  
    Object transaction = doGetTransaction();  
    boolean debugEnabled = logger.isDebugEnabled();  
  
    //当前已存在事务  
    if (isExistingTransaction(transaction)) {  
       // Existing transaction found -> check propagation behavior to find out how to behave.  
       return handleExistingTransaction(def, transaction, debugEnabled);  
    }  
  
    // Check definition settings for new transaction.  
    if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {  
       throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());  
    }  
  
    // No existing transaction found -> check propagation behavior to find out how to proceed.  
    //传播类型为MANDATORY(强制性的)，如果当前无事务则抛出异常  
    if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {  
       throw new IllegalTransactionStateException(  
             "No existing transaction found for transaction marked with propagation 'mandatory'");  
    }  
    //传播类型为REQUIRED、REQUIRES_NEW、NESTED则新开启一个事务  
    else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||  
          def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||  
          def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {  
       SuspendedResourcesHolder suspendedResources = suspend(null);  
       if (debugEnabled) {  
          logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);  
       }  
       try {  
          return startTransaction(def, transaction, debugEnabled, suspendedResources);  
       }  
       catch (RuntimeException | Error ex) {  
          resume(null, suspendedResources);  
          throw ex;  
       }  
    }  
    //传播类型为SUPPORTS、NOT_SUPPORTED、NEVER  
    else {  
       // Create "empty" transaction: no actual transaction, but potentially synchronization.  
       //如果当前事务隔离级别不是-1，并且开启warn提醒，则打印一条warn日T志  
       if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {  
          logger.warn("Custom isolation level specified but no actual transaction initiated; " +  
                "isolation level will effectively be ignored: " + def);  
       }  
       boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);  
       return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);  
    }  
}
  • AbstractPlatformTransactionManager.handleExistingTransaction实现
    该方法用于处理当前已存在事务的情况
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
private TransactionStatus handleExistingTransaction(  
       TransactionDefinition definition, Object transaction, boolean debugEnabled)  
       throws TransactionException {  
  
    //事务传播类型为NEVER，抛出异常  
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {  
       throw new IllegalTransactionStateException(  
             "Existing transaction found for transaction marked with propagation 'never'");  
    }  
  
    //事务传播类型为NOT_SUPPORTED，挂起当前事务  
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {  
       if (debugEnabled) {  
          logger.debug("Suspending current transaction");  
       }  
       //挂起当前事务  
       Object suspendedResources = suspend(transaction);  
       boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);  
       //创建空事务  
       return prepareTransactionStatus(  
             definition, null, false, newSynchronization, debugEnabled, suspendedResources);  
    }  
  
    //事务传播类型为REQUIRES_NEW，挂起当前事务，开启新事物  
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {  
       if (debugEnabled) {  
          logger.debug("Suspending current transaction, creating new transaction with name [" +  
                definition.getName() + "]");  
       }  
       SuspendedResourcesHolder suspendedResources = suspend(transaction);  
       try {  
          return startTransaction(definition, transaction, debugEnabled, suspendedResources);  
       }  
       catch (RuntimeException | Error beginEx) {  
          resumeAfterBeginException(transaction, suspendedResources, beginEx);  
          throw beginEx;  
       }  
    }  
  
    //事务传播类型为NESTED  
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {  
       if (!isNestedTransactionAllowed()) {  
          throw new NestedTransactionNotSupportedException(  
                "Transaction manager does not allow nested transactions by default - " +  
                "specify 'nestedTransactionAllowed' property with value 'true'");  
       }  
       if (debugEnabled) {  
          logger.debug("Creating nested transaction with name [" + definition.getName() + "]");  
       }  
       if (useSavepointForNestedTransaction()) {  
          // Create savepoint within existing Spring-managed transaction,  
          // through the SavepointManager API implemented by TransactionStatus.          // Usually uses JDBC savepoints. Never activates Spring synchronization.          //创建一个保存点  
          DefaultTransactionStatus status =  
                prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);  
          status.createAndHoldSavepoint();  
          return status;  
       }  
       else {  
          // Nested transaction through nested begin and commit/rollback calls.  
          // Usually only for JTA: Spring synchronization might get activated here          // in case of a pre-existing JTA transaction.          
          return startTransaction(definition, transaction, debugEnabled, null);  
       }  
    }
  • AbstractPlatformTransactionManager.suspend
    事务挂起
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
protected final SuspendedResourcesHolder suspend(@Nullable Object transaction) throws TransactionException {  
    //判断事务同步状态是否激活  
    if (TransactionSynchronizationManager.isSynchronizationActive()) {  
       List<TransactionSynchronization> suspendedSynchronizations = doSuspendSynchronization();  
       try {  
          Object suspendedResources = null;  
          if (transaction != null) {  
             //挂起当前事务  
             suspendedResources = doSuspend(transaction);  
          }  
          String name = TransactionSynchronizationManager.getCurrentTransactionName();  
          //清空当前事务名称  
          TransactionSynchronizationManager.setCurrentTransactionName(null);  
          boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();  
          //readOnly置为false  
          TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);  
          Integer isolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();  
          //事务隔离级别设置为null  
          TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(null);  
          boolean wasActive = TransactionSynchronizationManager.isActualTransactionActive();  
          //事务实际活跃状态设置为false  
          TransactionSynchronizationManager.setActualTransactionActive(false);  
          return new SuspendedResourcesHolder(  
                suspendedResources, suspendedSynchronizations, name, readOnly, isolationLevel, wasActive);  
       }  
       catch (RuntimeException | Error ex) {  
          // doSuspend failed - original transaction is still active...  
          //异常情况恢复同步信息  
          doResumeSynchronization(suspendedSynchronizations);  
          throw ex;  
       }  
    }  
    else if (transaction != null) {  
       // Transaction active but no synchronization active.  
       Object suspendedResources = doSuspend(transaction);  
       return new SuspendedResourcesHolder(suspendedResources);  
    }  
    else {  
       // Neither transaction nor synchronization active.  
       return null;  
    }  
}

其中doSuspend会执行当前事务的挂起操作，AbstractPlatformTransactionManager抽象类doSuspend()会直接抛出异常，依赖于具体事务执行器的实现，DataSourceTransactionManager实现如下：

1
2
3
4
5
6
7
protected Object doSuspend(Object transaction) {  
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;  
    //连接持有对象置为null  
    txObject.setConnectionHolder(null);  
    //删除datasource  
    return TransactionSynchronizationManager.unbindResource(obtainDataSource());  
}

其中核心操作包含两步
1、当前事务清空ConnectionHolder
2、当前线程解绑datasource（ThreadLocal中移除resources对应的key）

  • TransactionSynchronizationManager
    事务同步管理器，该类维护了多个线程本地变量ThreadLocal，如下所示：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    public abstract class TransactionSynchronizationManager {  
      
        //事务资源  
        private static final ThreadLocal<Map<Object, Object>> resources =  
              new NamedThreadLocal<>("Transactional resources");  
      
        //事务同步信息  
        private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =  
              new NamedThreadLocal<>("Transaction synchronizations");  
      
        //当前事务名称  
        private static final ThreadLocal<String> currentTransactionName =  
              new NamedThreadLocal<>("Current transaction name");  
      
        //事务只读状态  
        private static final ThreadLocal<Boolean> currentTransactionReadOnly =  
              new NamedThreadLocal<>("Current transaction read-only status");  
      
        //事务隔离级别  
        private static final ThreadLocal<Integer> currentTransactionIsolationLevel =  
              new NamedThreadLocal<>("Current transaction isolation level");  
      
        //事务活跃状态  
        private static final ThreadLocal<Boolean> actualTransactionActive =  
              new NamedThreadLocal<>("Actual transaction active");
        //......
    }

  • DataSourceTransactionManager.doBegin

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
protected void doBegin(Object transaction, TransactionDefinition definition) {  
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;  
    Connection con = null;  
  
    try {  
       //如果事务对象还没有ConnectionHolder或者其事务同步中  
       if (!txObject.hasConnectionHolder() ||  
             txObject.getConnectionHolder().isSynchronizedWithTransaction()) {  
          //获取新的数据库连接对象  
          Connection newCon = obtainDataSource().getConnection();  
          if (logger.isDebugEnabled()) {  
             logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");  
          }  
          //设置ConnectionHolder  
          txObject.setConnectionHolder(new ConnectionHolder(newCon), true);  
       }  
  
       //设置新的ConnectionHolder为事务同步中  
       txObject.getConnectionHolder().setSynchronizedWithTransaction(true);  
       con = txObject.getConnectionHolder().getConnection();  
  
       Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);  
       //设置事务隔离级别、readOnly  
       txObject.setPreviousIsolationLevel(previousIsolationLevel);  
       txObject.setReadOnly(definition.isReadOnly());  
  
       // Switch to manual commit if necessary. This is very expensive in some JDBC drivers,  
       // so we don't want to do it unnecessarily (for example if we've explicitly       // configured the connection pool to set it already).       if (con.getAutoCommit()) {  
          txObject.setMustRestoreAutoCommit(true);  
          if (logger.isDebugEnabled()) {  
             logger.debug("Switching JDBC Connection [" + con + "] to manual commit");  
          }  
          //如果是自动提交切换到手动提交  
          con.setAutoCommit(false);  
       }  
  
       //如果只读，执行sql设置事务为只读  
       prepareTransactionalConnection(con, definition);  
       //设置事务active状态  
       txObject.getConnectionHolder().setTransactionActive(true);  
  
       //设置超时时间  
       int timeout = determineTimeout(definition);  
       if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {  
          txObject.getConnectionHolder().setTimeoutInSeconds(timeout);  
       }  
  
       // Bind the connection holder to the thread.  
       //绑定ConnectionHolder到当前线程(thread的ThreadLocalMap)  
       if (txObject.isNewConnectionHolder()) {  
          TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());  
       }  
    }  
  
    catch (Throwable ex) {  
       if (txObject.isNewConnectionHolder()) {  
          //出现异常释放连接并清空ConnectionHolder  
          DataSourceUtils.releaseConnection(con, obtainDataSource());  
          txObject.setConnectionHolder(null, false);  
       }  
       throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);  
    }  
}

该方法的核心操作是：
1、获取新的数据库连接对象，设置ConnectionHolder
2、在当前线程的ThreadLocal变量resources中设置key-value（key=DataSource对象，value=ConnectionHolder）

2.2、commit提交事务

  • AbstractPlatformTransactionManager.commit
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public final void commit(TransactionStatus status) throws TransactionException {  
    //如果事务已完成，抛出异常  
    if (status.isCompleted()) {  
       throw new IllegalTransactionStateException(  
             "Transaction is already completed - do not call commit or rollback more than once per transaction");  
    }  
  
    DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;  
    //事务标记为回滚  
    if (defStatus.isLocalRollbackOnly()) {  
       if (defStatus.isDebug()) {  
          logger.debug("Transactional code has requested rollback");  
       }  
       //执行回滚  
       processRollback(defStatus, false);  
       return;  
    }  
  
    //不需要全局回滚时提交 && 全局回滚  
    if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {  
       if (defStatus.isDebug()) {  
          logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");  
       }  
       processRollback(defStatus, true);  
       return;  
    }  
  
    //执行事务提交  
    processCommit(defStatus);  
}
  • AbstractPlatformTransactionManager.processCommit
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    private void processCommit(DefaultTransactionStatus status) throws TransactionException {  
        try {  
           boolean beforeCompletionInvoked = false;  
      
           try {  
              boolean unexpectedRollback = false;  
              //3个前置操作  
              prepareForCommit(status);  
              triggerBeforeCommit(status);  
              triggerBeforeCompletion(status);  
              beforeCompletionInvoked = true;  
      
              //有保存点，即嵌套事务  
              if (status.hasSavepoint()) {  
                 if (status.isDebug()) {  
                    logger.debug("Releasing transaction savepoint");  
                 }  
                 unexpectedRollback = status.isGlobalRollbackOnly();  
                 //释放保存点  
                 status.releaseHeldSavepoint();  
              }  
              //新事务  
              else if (status.isNewTransaction()) {  
                 if (status.isDebug()) {  
                    logger.debug("Initiating transaction commit");  
                 }  
                 unexpectedRollback = status.isGlobalRollbackOnly();  
                 //执行提交  
                 doCommit(status);  
              }  
              //全局回滚失败  
              else if (isFailEarlyOnGlobalRollbackOnly()) {  
                 unexpectedRollback = status.isGlobalRollbackOnly();  
              }  
      
              // Throw UnexpectedRollbackException if we have a global rollback-only  
              // marker but still didn't get a corresponding exception from commit.          if (unexpectedRollback) {  
                 throw new UnexpectedRollbackException(  
                       "Transaction silently rolled back because it has been marked as rollback-only");  
              }  
           }  
           catch (UnexpectedRollbackException ex) {  
              // can only be caused by doCommit  
              //触发完成后事务同步，状态为回滚  
              triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);  
              throw ex;  
           }  
           catch (TransactionException ex) {  
              // can only be caused by doCommit  
              //提交失败回滚  
              if (isRollbackOnCommitFailure()) {  
                 doRollbackOnCommitException(status, ex);  
              }  
              else {  
                 //触发完成后回调，事务同步状态为未知  
                 triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);  
              }  
              throw ex;  
           }  
           catch (RuntimeException | Error ex) {  
              //如果3个前置步骤没有执行完，调用前置最后一步操作  
              if (!beforeCompletionInvoked) {  
                 triggerBeforeCompletion(status);  
              }  
              doRollbackOnCommitException(status, ex);  
              throw ex;  
           }  
      
           // Trigger afterCommit callbacks, with an exception thrown there  
           // propagated to callers but the transaction still considered as committed.       try {  
              triggerAfterCommit(status);  
           }  
           finally {  
              triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);  
           }  
      
        }  
        finally {  
           cleanupAfterCompletion(status);  
        }  
    }

2.3、rollback回滚事务

  • AbstractPlatformTransactionManager.processRollback
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
private void processRollback(DefaultTransactionStatus status, boolean unexpected) {  
    try {  
       boolean unexpectedRollback = unexpected;  
  
       try {  
          //触发前置操作：遍历事务同步信息触发beforeCompletion  
          triggerBeforeCompletion(status);  
  
          //有保存点(即嵌套式事务)，回滚到保存点  
          if (status.hasSavepoint()) {  
             if (status.isDebug()) {  
                logger.debug("Rolling back transaction to savepoint");  
             }  
             status.rollbackToHeldSavepoint();  
          }  
          else if (status.isNewTransaction()) {  
             if (status.isDebug()) {  
                logger.debug("Initiating transaction rollback");  
             }  
             //执行回滚  
             doRollback(status);  
          }  
          else {  
             //非新事务&无保存点，加入当前事务  
             // Participating in larger transaction  
             if (status.hasTransaction()) {  
                if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {  
                   if (status.isDebug()) {  
                      logger.debug("Participating transaction failed - marking existing transaction as rollback-only");  
                   }  
                   doSetRollbackOnly(status);  
                }  
                else {  
                   if (status.isDebug()) {  
                      logger.debug("Participating transaction failed - letting transaction originator decide on rollback");  
                   }  
                }  
             }  
             else {  
                logger.debug("Should roll back transaction but cannot - no transaction available");  
             }  
             // Unexpected rollback only matters here if we're asked to fail early  
             if (!isFailEarlyOnGlobalRollbackOnly()) {  
                unexpectedRollback = false;  
             }  
          }  
       }  
       catch (RuntimeException | Error ex) {  
          triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);  
          throw ex;  
       }  
  
       //遍历事务同步信息执行invokeAfterCompletion  
       triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);  
  
       // Raise UnexpectedRollbackException if we had a global rollback-only marker  
       if (unexpectedRollback) {  
          throw new UnexpectedRollbackException(  
                "Transaction rolled back because it has been marked as rollback-only");  
       }  
    }  
    finally {  
       //完成后清理  
       cleanupAfterCompletion(status);  
    }  
}


-------------本文结束，感谢您的阅读-------------



本文作者： 陈进涛
本文链接： https://zhizhi123.com/2021/11/07/spring-transaction-core-implementation/
版权声明： 本作品采用 署名—非商业性使用—相同方式共享 4.0 协议国际版 进行许可。转载请注明出处！
知识共享许可协议

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

很惭愧<br/><br/>只做了一点微小的工作<br/>谢谢大家