事物增强器（上篇）

TransactionInterceptor支撑着整个事务功能的架构，逻辑还是相对复杂的，那么我们现在切入正题来分析此拦截器是如何实现事务特性的。TransactionInterceptor类继承自MethodInterceptor，所以调用该类是从其invoke方法开始的，首先看下这个方法的源码：

public Object invoke(MethodInvocation invocation) throws Throwable {
        Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
        return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
    }

protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
            final InvocationCallback invocation) throws Throwable {

        TransactionAttributeSource tas = getTransactionAttributeSource();
        //获取对应事务属性
        final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
        //获取BeanFactory的TransactionManager
        final PlatformTransactionManager tm = determineTransactionManager(txAttr);
        //构造方法唯一标识
        final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
        //声明式事务处理
        if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
            //创建TransactionInfo
            TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
            Object retVal = null;
            try {
                // 执行被增强方法
                retVal = invocation.proceedWithInvocation();
            }
            catch (Throwable ex) {
                //异常回滚
                completeTransactionAfterThrowing(txInfo, ex);
                throw ex;
            }
            finally {
                //清除消息
                cleanupTransactionInfo(txInfo);
            }
            //提交事务
            commitTransactionAfterReturning(txInfo);
            return retVal;
        }

        else {
            final ThrowableHolder throwableHolder = new ThrowableHolder();

            // 编程式事务处理
            try {
                Object result = ((CallbackPreferringPlatformTransactionManager) tm).execute(txAttr, status -> {
                    TransactionInfo txInfo = prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
                    try {
                        return invocation.proceedWithInvocation();
                    }
                    catch (Throwable ex) {
                        if (txAttr.rollbackOn(ex)) {
                            // A RuntimeException: will lead to a rollback.
                            if (ex instanceof RuntimeException) {
                                throw (RuntimeException) ex;
                            }
                            else {
                                throw new ThrowableHolderException(ex);
                            }
                        }
                        else {
                            // A normal return value: will lead to a commit.
                            throwableHolder.throwable = ex;
                            return null;
                        }
                    }
                    finally {
                        cleanupTransactionInfo(txInfo);
                    }
                });

                // Check result state: It might indicate a Throwable to rethrow.
                if (throwableHolder.throwable != null) {
                    throw throwableHolder.throwable;
                }
                return result;
            }
            catch (ThrowableHolderException ex) {
                throw ex.getCause();
            }
            catch (TransactionSystemException ex2) {
                if (throwableHolder.throwable != null) {
                    logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
                    ex2.initApplicationException(throwableHolder.throwable);
                }
                throw ex2;
            }
            catch (Throwable ex2) {
                if (throwableHolder.throwable != null) {
                    logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
                }
                throw ex2;
            }
        }
    }

从上面的函数中，我们尝试整理下事务处理的脉络，在Spring中支持两种事务处理的方式，分别是声明式事务处理与编程式事务处理，两者对于开发人员来讲差别很大，但是对于Spring中的实现来讲，大同小异。在invoke中我们也可以看到这两种方式的实现。考虑到对事务的应用比声明式的事务处理使用起来更方便，也相对流行些，我们就以此种方式进行分析。对于声明式的事务处理主要有以下几个步骤：

（1）获取事务的属性。

对于事务处理来说，最基础或者说最首要的工作便是获取事务属性了，这是支撑整个事务功能的基石，如果没有事务属性，其他功能也无从谈起，在分析事务准备阶段时我们已经分析了事务属性提取的功能，大家应该有所了解。

（2）加载配置中配置的TransactionManager。

（3）不同事务处理方式使用不同的逻辑。

　　对于声明式事务的处理与编程式事务的处理，第一点区别在于事务属性上，因为编程式的事务处理是不需要有事务属性的，第二点区别就是在TransactionManager上，CallbackPreferringPlatformTransactionManager实现PlatformTransactionManager接口，暴露出一个方法用于执行事务处理中的回调。所以，这两种方式都可以用作事务处理方式的判断。

（4）在目标方法执行前获取事务并收集事务信息。

　　事务信息与事务属性并不相同，也就是TransactionInfo与TransactionAttribute并不相同，TransactionInfo中包含TransactionAttribute信息，但是，除了TransactionAttribute外还有其他事务信息，例如PlatformTransactionManager以及TransactionStatus相关信息。

（5）执行目标方法。

（6）一旦出现异常，尝试异常处理。

　　并不是所有异常，Spring都会将其回滚，默认只对RuntimeException回滚。

（7）提交事务前的事务信息清除。

（8）提交事务。

上面的步骤分析旨在让大家对事务功能与步骤有大致的了解，下面将其步骤进行详细的分析。

创建事务

先来分析事务的创建过程。

protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
            @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {

        // 如果没有名称指定则使用方法唯一标识，并使用DelegatingTransactionAttribute封装txAttr
        if (txAttr != null && txAttr.getName() == null) {
            txAttr = new DelegatingTransactionAttribute(txAttr) {
                @Override
                public String getName() {
                    return joinpointIdentification;
                }
            };
        }

        TransactionStatus status = null;
        if (txAttr != null) {
            if (tm != null) {
                //获取TransactionStatus
                status = tm.getTransaction(txAttr);
            }
            else {
                if (logger.isDebugEnabled()) {
                    logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
                            "] because no transaction manager has been configured");
                }
            }
        }
        //根据指定的属性与status准备一个TransactionInfo
        return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
    }

对于createTransactionIfNecessary函数主要做了这几件事情：

（1）使用DelegatingTransactionAttribute封装传入的TransactionAttribute实例。

　　对于传入的TransactionAttribute类型的参数txAttr，当前的实例类型是RuleBasedTransactionAttribute，是由获取事务属性时生成，主要用于数据承载，而这里之所以使用DelegatingTransactionAttribute进行封装，当然是提供了更多的功能。

（2）获取事务。

　　事务处理当然是以事务为核心，那么获取事务就是最重要的事情。

（3）构建事务信息。

根据之前几个步骤获取的信息构建TransactionInfo并返回。

分别对以上信息进行详细的解析。

1.获取事务

Spring中使用getTransaction来处理事务的准备工作，包括事务获取以及信息的构建。

public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
        Object transaction = doGetTransaction();

        //缓存调试标志以避免重复检查。
        boolean debugEnabled = logger.isDebugEnabled();

        if (definition == null) {
            // 如果没有给出事务定义，则使用默认值。
            definition = new DefaultTransactionDefinition();
        }

        if (isExistingTransaction(transaction)) {
            // 判断当前线程是否存在事务，判断依据为当前线程记录的连接不为空且连接中的TransactionActive属性不为空
            return handleExistingTransaction(definition, transaction, debugEnabled);
        }

        // 事务超时设置验证
        if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
            throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
        }

        //如果当前线程不存在事务，但是PropagationBehavior却被声明为PROPAGATION_MANDATORY抛出异常
        if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
            throw new IllegalTransactionStateException(
                    "No existing transaction found for transaction marked with propagation 'mandatory'");
        }
        //PROPAGATION_REQUIRED、PROPAGATION_REQUIRES_NEW、PROPAGATION_REQUIRES_NEW都需要新建事务
        else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
                definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
                definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
            //空挂起
            SuspendedResourcesHolder suspendedResources = suspend(null);
            if (debugEnabled) {
                logger.debug("Creating new transaction with name [" + definition.getName() + "]: " + definition);
            }
            try {
                boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
                DefaultTransactionStatus status = newTransactionStatus(
                        definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
                //构造Transaction，包括设置ConnectionHolder、隔离级别、timeout，如果是新连接，绑定到当前线程
                doBegin(transaction, definition);
                //新同步事务的设置，针对于当前线程的设置
                prepareSynchronization(status, definition);
                return status;
            }
            catch (RuntimeException | Error ex) {
                resume(null, suspendedResources);
                throw ex;
            }
        }
        else {
            // Create "empty" transaction: no actual transaction, but potentially synchronization.
            if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
                logger.warn("Custom isolation level specified but no actual transaction initiated; " +
                        "isolation level will effectively be ignored: " + definition);
            }
            boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
            return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null);
        }
    }

当然，在Spring中每个复杂的功能实现，并不是一次完成的，而是会通过入口函数进行一个框架的搭建，初步构建完整的逻辑，而将实现的细节分摊给不同的函数。那么，让我们来看看事务的准备工作都包括哪些：

（1）获取事务

　　创建对应的事务实例，这里使用的是DataSourceTransactionManager中的doGetTransaction方法，创建基于JDBC的事务实例。如果当前线程中存在关于DataSource的连接，那么直接使用。这里有一个对保存点的设置，是否开启允许保存点取决于是否设置了允许嵌入式事务。

protected Object doGetTransaction() {
        DataSourceTransactionObject txObject = new DataSourceTransactionObject();
        txObject.setSavepointAllowed(isNestedTransactionAllowed());
        //如果当前线程已经记录数据库连接则使用原有连接
        ConnectionHolder conHolder =
                (ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
        //false表示非新创建连接
        txObject.setConnectionHolder(conHolder, false);
        return txObject;
    }

（2）如果当前线程存在事务，则转向嵌套事务的处理。

（3）事务超时设置验证。

（4）事务propagationBehavior属性的设置验证。

（5）构建DefaultTransactionStatus。

（6）完善Transaction，包括设置ConnectionHolder、隔离级别、timeout，如果是新连接，则绑定到当前线程。

对于一些隔离级别，timeout等功能的设置并不是由Spring来完成的，而是委托给底层的数据库连接去做的，而对于数据库连接的设置就是在doBegin函数中处理的。

/**
     * 完善Transaction，包括设置ConnectionHolder、隔离级别、timeout，如果是新连接，则绑定到当前线程。
     */
    @Override
    protected void doBegin(Object transaction, TransactionDefinition definition) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
        Connection con = null;

        try {
            if (!txObject.hasConnectionHolder() ||
                    txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
                Connection newCon = obtainDataSource().getConnection();
                if (logger.isDebugEnabled()) {
                    logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
                }
                txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
            }

            txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
            con = txObject.getConnectionHolder().getConnection();
            //设置隔离级别
            Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
            txObject.setPreviousIsolationLevel(previousIsolationLevel);

            // 更改自动提交设置，由Spring控制提交
            if (con.getAutoCommit()) {
                txObject.setMustRestoreAutoCommit(true);
                if (logger.isDebugEnabled()) {
                    logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
                }
                con.setAutoCommit(false);
            }

            prepareTransactionalConnection(con, definition);
            //设置判断当前线程是否存在事务的依据
            txObject.getConnectionHolder().setTransactionActive(true);

            int timeout = determineTimeout(definition);
            if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
                txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
            }

            // 将当前获取到的连接绑定到当前线程
            if (txObject.isNewConnectionHolder()) {
                TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
            }
        }

        catch (Throwable ex) {
            if (txObject.isNewConnectionHolder()) {
                DataSourceUtils.releaseConnection(con, obtainDataSource());
                txObject.setConnectionHolder(null, false);
            }
            throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
        }
    }

可以说事务是从这个函数开始的，因为这个函数中已经开始尝试了对数据库连接的获取，当然，在获取数据库连接的同时，一些必要的设置也是需要同步设置的。

❤ 尝试获取连接

　　当然不是每次都会获取新的连接，如果当前线程中的connectionHolder已经存在，则没有必要再次获取，或者，对于事务同步表示设置为true的需要重新获取连接。

❤ 设置隔离级别以及只读标识

　　Spring中确实是针对只读操作做了一些处理，但是核心的实现是设置connection上的readOnly属性。同样，对于隔离级别的控制也是交由connection去控制的。

❤ 更改默认的提交设置

　　如果事务属性是自动提交，那么需要改变这种设置，而将提交操作委托给Spring来处理。

❤ 设置标志位，标识当前连接已经被事务激活

❤ 设置过期时间

❤ 将connectionHolder绑定到当前线程

设置隔离级别的prepareConnectionForTransaction函数用于负责对底层数据库连接的设置，当然，只是包含只读标识和隔离级别的设置。由于强大的日志及异常处理，显得函数代码量比较大，但是单从业务角度去看，关键代码其实是不多的。

public static Integer prepareConnectionForTransaction(Connection con, @Nullable TransactionDefinition definition)
            throws SQLException {

        Assert.notNull(con, "No Connection specified");

        // 设置数据库连接的只读标识
        if (definition != null && definition.isReadOnly()) {
            try {
                if (logger.isDebugEnabled()) {
                    logger.debug("Setting JDBC Connection [" + con + "] read-only");
                }
                con.setReadOnly(true);
            }
            catch (SQLException | RuntimeException ex) {
                Throwable exToCheck = ex;
                while (exToCheck != null) {
                    if (exToCheck.getClass().getSimpleName().contains("Timeout")) {
                        // Assume it's a connection timeout that would otherwise get lost: e.g. from JDBC 4.0
                        throw ex;
                    }
                    exToCheck = exToCheck.getCause();
                }
                // "read-only not supported" SQLException -> ignore, it's just a hint anyway
                logger.debug("Could not set JDBC Connection read-only", ex);
            }
        }

        // 设置数据库连接的隔离级别
        Integer previousIsolationLevel = null;
        if (definition != null && definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
            if (logger.isDebugEnabled()) {
                logger.debug("Changing isolation level of JDBC Connection [" + con + "] to " +
                        definition.getIsolationLevel());
            }
            int currentIsolation = con.getTransactionIsolation();
            if (currentIsolation != definition.getIsolationLevel()) {
                previousIsolationLevel = currentIsolation;
                con.setTransactionIsolation(definition.getIsolationLevel());
            }
        }

        return previousIsolationLevel;
    }

（7）将事务信息记录在当前线程中

protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
        if (status.isNewSynchronization()) {
            TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
            TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(
                    definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ?
                            definition.getIsolationLevel() : null);
            TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());
            TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());
            TransactionSynchronizationManager.initSynchronization();
        }
    }

2.处理已经存在的事务

之前讲述了普通事务的建立的过程，但是Spring中支持多种事务的传播规则，比如PROPAGATION_REQUIRED、PROPAGATION_REQUIRES_NEW、PROPAGATION_REQUIRES_NEW等，这些都是在已经存在事务的基础上进行进一步的处理，那么，对于已经存在的事务，准备操作是如何进行的呢？

private TransactionStatus handleExistingTransaction(
            TransactionDefinition definition, Object transaction, boolean debugEnabled)
            throws TransactionException {

        if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
            throw new IllegalTransactionStateException(
                    "Existing transaction found for transaction marked with propagation 'never'");
        }

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

        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 {
                boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
                DefaultTransactionStatus status = newTransactionStatus(
                        definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
                doBegin(transaction, definition);
                prepareSynchronization(status, definition);
                return status;
            }
            catch (RuntimeException | Error beginEx) {
                resumeAfterBeginException(transaction, suspendedResources, beginEx);
                throw beginEx;
            }
        }
        //嵌入式事务的处理
        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()) {
                //如果没有可以使用保存点的方式控制事务回滚，那么在嵌入式事务的建立初始建立保存点
                DefaultTransactionStatus status =
                        prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
                status.createAndHoldSavepoint();
                return status;
            }
            else {
                // 有些情况下是不能使用保存点操作，那么建立新事务
                boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
                DefaultTransactionStatus status = newTransactionStatus(
                        definition, transaction, true, newSynchronization, debugEnabled, null);
                doBegin(transaction, definition);
                prepareSynchronization(status, definition);
                return status;
            }
        }

        // 假设propagation ation_supports或propagation ation_required。
        if (debugEnabled) {
            logger.debug("Participating in existing transaction");
        }
        if (isValidateExistingTransaction()) {
            if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
                Integer currentIsolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
                if (currentIsolationLevel == null || currentIsolationLevel != definition.getIsolationLevel()) {
                    Constants isoConstants = DefaultTransactionDefinition.constants;
                    throw new IllegalTransactionStateException("Participating transaction with definition [" +
                            definition + "] specifies isolation level which is incompatible with existing transaction: " +
                            (currentIsolationLevel != null ?
                                    isoConstants.toCode(currentIsolationLevel, DefaultTransactionDefinition.PREFIX_ISOLATION) :
                                    "(unknown)"));
                }
            }
            if (!definition.isReadOnly()) {
                if (TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
                    throw new IllegalTransactionStateException("Participating transaction with definition [" +
                            definition + "] is not marked as read-only but existing transaction is");
                }
            }
        }
        boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
        return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
    }

对于已经存在的事务处理过程中，我们看到了很多熟悉的操作，但是也有不同的地方，上述函数对于已经存在的事务处理考虑两种情况：

（1）PROPAGATION_REQUIRES_NEW 表示当前方法必须在它自己的事务里运行，一个新的事务将被启动，而一个事务正在运行的话，则在这个方法运行期间被挂起。而Spring中对于此种传播方式的处理与新事务建立最大的不同点在于使用suspend方法将原事务挂起。将信息挂起的目的当然是为了在当前事务执行完毕后再将原事务还原。

（2）PROPAGATION_NESTED 表示如果当前正有一个事务在运行中，则该方法应该运行在一个嵌套的事务中，被嵌套的事务可以独立于封装事务进行提交或者回滚，如果封装事务不存在，行为就像PROPAGATION_REQUIRES_NEW。对于嵌入式事务的处理，Spring中主要考虑两种方式的处理。

　　❤ Spring中允许嵌入事务的时候，则首选设置保存点的方式作为异常处理的回滚。

　　❤ 对于其他方式，比如JTA 无法使用保存点的方式，那么处理方式与PROPAGATION_REQUIRES_NEW相同，而一旦出现异常，则由Spring的事务异常处理机制去完成后续的操作。

对于挂起操作的主要目的是记录原有事务的状态，以便于后续操作对事务的恢复。

3.准备事务信息

当已经建立事务连接并完成了事务信息的提取后，我们需要将所有的事务信息统一记录在TransactionInfo类型的实例中，这个实例包含了目标方法开始前的所有状态信息，一旦事务执行失败，Spring会通过TransactionInfo类型的实例中的信息来进行回滚等后续工作。

protected TransactionInfo prepareTransactionInfo(@Nullable PlatformTransactionManager tm,
            @Nullable TransactionAttribute txAttr, String joinpointIdentification,
            @Nullable TransactionStatus status) {

        TransactionInfo txInfo = new TransactionInfo(tm, txAttr, joinpointIdentification);
        if (txAttr != null) {
            if (logger.isTraceEnabled()) {
                logger.trace("Getting transaction for [" + txInfo.getJoinpointIdentification() + "]");
            }
            // 记录事务状态
            txInfo.newTransactionStatus(status);
        }
        else {
            if (logger.isTraceEnabled()) {
                logger.trace("Don't need to create transaction for [" + joinpointIdentification +
                        "]: This method isn't transactional.");
            }
        }

        // We always bind the TransactionInfo to the thread, even if we didn't create
        // a new transaction here. This guarantees that the TransactionInfo stack
        // will be managed correctly even if no transaction was created by this aspect.
        txInfo.bindToThread();
        return txInfo;
    }

参考：《Spring源码深度解析》 郝佳 编著：