526互联

Spring的事务实现原理

发布时间 2023-12-24 08:45:46作者: 残城碎梦

Spring事务

Spring本身并不实现事务，Spring事务的本质还是底层数据库对事务的支持，没有数据库事务的支持，Spring事务就不会生效。

例如：使用JDBC操作数据库，使用事务的步骤主要分为如下5步:

第一步：获取连接Connection con = DriverManager.getConnection();

第二步：开启事务con.setAutoCommit(true/false);

第三步：执行CRUD;

第四步：提交事务/回滚事务con.commit() / con.rollback();

第五步：关闭连接conn.close();

采用Spring事务后，只需要关注第三步的实现，其他的步骤都是Spring自动完成。

Spring事务的本质其实就是数据库对事务的支持，Spring只提供统一事务管理接口，具体实现都是由各数据库自己实现。

Spring事务使用

Spring事务管理有两种方式：编程式事务管理、声明式事务管理。

编程式事务

所谓编程式事务指的是通过编码方式实现事务，允许用户在代码中精确定义事务的边界。

Spring团队通常建议使用TransactionTemplate进行程序化事务管理，如下所示:

@Autowired
private TransactionTemplate transactionTemplate;
public void testTransaction() {
    transactionTemplate.execute (new TransactionCallbackWithoutResult() {
        @Override
        protected void doInTransactionWithoutResult(TransactionStatus transactionStatus)
            try {
                // l....业务代码
            }catch (Exception e){
                //回滚
                transactionStatus.setRollbackOnly(O);
            }
    });
}

声明式事务

声明式事务最大的优点：就是不需要通过编程的方式管理事务，只需在需要使用的地方标注@Transactional注解，就能为该方法开启事务了。

如下所示：

@Transactional
public void insert (string userName){
    this.jdbcTemplate.update ("insert into t_user (name) values (?)"，userName);
}

Spring事务实现原理

想要了解Spring事务的实现原理，一个绕不开的点就是Spring AOP，因为事务就是依靠Spring AOP实现的。

@EnableTransactionManagement是开启注解式事务的事务，如果注解式事务真的有玄机，那么@EnableTransactionManagement就是我们揭开秘密的突破口。

@EnableTransactionManagement注解源码如下：

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import({TransactionManagementConfigurationSelector.class})
public @interface EnableTransactionManagement {
    
    //用来表示默认使用JDK Dynamic Proxy还是CGLIB Proxy
    boolean proxyTargetClass() default false;
    
    //表示以Proxy-based方式实现AOP还是以weaving-based方式实现AOP
    AdviceMode mode() default AdviceMode.PROXY;

    //顺序
    int order() default 2147483647;
}

@EnableTransactionManagement注解，主要通过@lmport引入了另一个配置TransactionManagentConfigurationSelector。

在Spring中Selector通常都是用来选择一些Bean，向容器注册：AutoProxyRegistrar、ProxyTransactionManagementConfiguration两个Bean。 

ProxyTransactionManagementConfiguration是一个配置类，实现类主要是TransactionInterceptor，这里会涉及Spring AOP实现切面逻辑织入。

要实现AOP需要明确两个点：

1）定义切点：需要在哪里做增强?

2）定义增强逻辑：需要做什么样的增强逻辑?

如果对于这两点，Spring主要通过事务代理管理配置类: ProxyTransactionManagementConfiguration来实现的

首先看下核心调用接口TransactionInterceptor实现的invoke()方法的源码：

public Object invoke(final MethodInvocation invocation) throws Throwable {
    //获取需要织入事务逻辑的目标类
    Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);

    //进行事务逻辑的织入
    return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
}

在invokeWithinTransaction中会调用createTransactionlfNecessary方法：

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

    //如果TransactionAttribute的名称为空，则创建一个代理的TransactionAttribute,
    //并且将其名称设置为需要织入事务的方法的名称
    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) {
            //如果事务属性不为空，并且TransactionManager都存在，
            //则通过TransactionManager获取当前事务状态的对象
            status = tm.getTransaction(txAttr);
        }
        else {
            if (logger.isDebugEnabled()) {
                logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
                             "] because no transaction manager has been configured");
            }
        }
    }
    //将当前事务属性和事务状态封装为一个TransactionInfo，这里主要做的工作是将事务属性绑定到当前线程
    return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

这里对事务的创建，首先会判断封装事务属性的对象名称是否为空，如果不为空，则以目标方法的标识符作为其名称，然后通过TransactionManager创建事务。

如下是PlatformTransactionManager.getTransaction()方法的源码：

public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
    Object transaction = doGetTransaction();
    boolean debugEnabled = logger.isDebugEnabled();

    if (definition == null) {
        //如果TransactionDefinition为空则使用默认配置
        definition = new DefaultTransactionDefinition();
    }

    if (isExistingTransaction(transaction)) {
        //如果有存在的事务，则处理存在的事物
        return handleExistingTransaction(definition, transaction, debugEnabled);
    }

    //判断事务是否超时
    if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
        throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
    }

    //使用PROPAGAT工ON_MANDATORY这个级别，没有可用的事务则抛异常
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
        throw new IllegalTransactionStateException(
            "No existing transaction found for transaction marked with propagation 'mandatory'");
    }
    else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
             definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
             definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
        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);
            //根据隔离级别开始事务
            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);
    }
}

doBegin会调用DataSourceTransactionManager的doBegin方法设置当前连接为手动提交事务。

protected void doBegin(Object transaction, TransactionDefinition definition) {
    DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
    Connection con = null;

    try {
        //获取数据库连接
        if (!txObject.hasConnectionHolder() || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
            Connection newCon = this.obtainDataSource().getConnection();
            if (this.logger.isDebugEnabled()) {
                this.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);
        //如果连接是自动提交，则设置成手动

        if (con.getAutoCommit()) {
            txObject.setMustRestoreAutoCommit(true);
            if (this.logger.isDebugEnabled()) {
                this.logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
            }

            con.setAutoCommit(false);
        }
        //这里会和数据库通信
        this.prepareTransactionalConnection(con, definition);
        txObject.getConnectionHolder().setTransactionActive(true);
        int timeout = this.determineTimeout(definition);
        if (timeout != -1) {
            txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
        }

        if (txObject.isNewConnectionHolder()) {
            TransactionSynchronizationManager.bindResource(this.obtainDataSource(), txObject.getConnectionHolder());
        }

    } catch (Throwable var7) {
        if (txObject.isNewConnectionHolder()) {
            DataSourceUtils.releaseConnection(con, this.obtainDataSource());
            txObject.setConnectionHolder((ConnectionHolder)null, false);
        }

        throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", var7);
    }
}

再看回org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction方法，这里显示了aop 如何处理事务

if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
    // 获取事务
    TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
    Object retVal = null;
    try {
        //触发后面的拦截器和方法本身
        retVal = invocation.proceedWithInvocation();
    }
    catch (Throwable ex) {
        //捕获异常, 处理回滚
        completeTransactionAfterThrowing(txInfo, ex);
        throw ex;
    }
    finally {
        //重置threadLocal中事务状态
        cleanupTransactionInfo(txInfo);
    }
    commitTransactionAfterReturning(txInfo);
    return retVal;
}

再看一下completeTransactionAfterThrowing方法，如果是需要回滚的异常则执行回滚，否则执行提交

protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {
    if (txInfo != null && txInfo.getTransactionStatus() != null) {
        if (logger.isTraceEnabled()) {
            logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() +
                         "] after exception: " + ex);
        }
        //需要回滚的异常
        if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
            try {
                //执行回滚
                txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
            }
            catch (TransactionSystemException ex2) {
                logger.error("Application exception overridden by rollback exception", ex);
                ex2.initApplicationException(ex);
                throw ex2;
            }
            catch (RuntimeException | Error ex2) {
                logger.error("Application exception overridden by rollback exception", ex);
                throw ex2;
            }
        }
        else {
            try {
                //提交
                txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
            }
            catch (TransactionSystemException ex2) {
                logger.error("Application exception overridden by commit exception", ex);
                ex2.initApplicationException(ex);
                throw ex2;
            }
            catch (RuntimeException | Error ex2) {
                logger.error("Application exception overridden by commit exception", ex);
                throw ex2;
            }
        }
    }
}

Spring事务原理总结

Spring事务的本质其实就是Spring AOP和数据库事务，Spring将数据库的事务操作提取为切面，通过AOP在方法执行前后增加数据库事务操作。