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

1、核心类介绍

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

• PlatformTransactionManager 底层接口，定义事务方法

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实现

@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实现 该方法用于处理当前已存在事务的情况

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 事务挂起

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实现如下：

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，如下所示：

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

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

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

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

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);  
    }  
}