Knowledge4J

切换主题

03.2 声明式事务源码分析 ​

本章深入 TransactionInterceptor → invokeWithinTransaction → DataSourceTransactionManager 的完整源码链路，理解声明式事务的 AOP 实现机理。

1. 声明式事务的 AOP 实现机制（核心） ​

1.1 TransactionInterceptor 源码 ​

java
// TransactionInterceptor 是事务的 AOP 入口
// 它实现了 MethodInterceptor（AOP Alliance 接口）
public class TransactionInterceptor extends TransactionAspectSupport
        implements MethodInterceptor, Serializable {
    
    @Override
    @Nullable
    public Object invoke(MethodInvocation invocation) throws Throwable {
        Class<?> targetClass = invocation.getThis() != null 
            ? AopUtils.getTargetClass(invocation.getThis()) : null;
        
        // 适配并调用父类 TransactionAspectSupport 的核心方法
        return invokeWithinTransaction(
            invocation.getMethod(), targetClass, invocation::proceed);
    }
}

1.2 TransactionAspectSupport.invokeWithinTransaction（核心源码） ​

java
// TransactionAspectSupport — 事务拦截的核心类
@Nullable
protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
        final InvocationCallback invocation) throws Throwable {
    
    // 1. 获取 @Transactional 属性
    TransactionAttributeSource tas = getTransactionAttributeSource();
    TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
    
    // 2. 确定事务管理器
    PlatformTransactionManager tm = determineTransactionManager(txAttr);
    
    // 3. 获取方法标识（用于日志）
    String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
    
    // 4. 创建/获取事务（关键！）
    TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
    
    Object retVal;
    try {
        // 5. 执行目标业务方法（回调模式）
        retVal = invocation.proceedWithInvocation();
    } catch (Throwable ex) {
        // 6. 异常时：根据回滚规则决定 commit 还是 rollback
        completeTransactionAfterThrowing(txInfo, ex);
        throw ex;
    } finally {
        // 7. 清理事务资源
        cleanupTransactionInfo(txInfo);
    }
    
    // 8. 正常返回：提交事务
    commitTransactionAfterReturning(txInfo);
    return retVal;
}

1.3 createTransactionIfNecessary（事务创建） ​

java
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
        @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
    
    // ... 省略日志和前缀逻辑
    
    TransactionStatus status = null;
    if (txAttr != null) {
        if (tm != null) {
            // 调用事务管理器的 getTransaction 方法
            // 这里决定了：是创建新事务，还是加入已有事务
            status = tm.getTransaction(txAttr);
        }
    }
    
    // 包装 TransactionInfo（绑定到当前线程）
    return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

2. DataSourceTransactionManager 源码深度剖析 ​

2.1 doGetTransaction（获取当前事务） ​

java
public class DataSourceTransactionManager extends AbstractPlatformTransactionManager
        implements ResourceTransactionManager, InitializingBean {
    
    @Override
    protected Object doGetTransaction() {
        // 1. 创建数据源事务对象
        DataSourceTransactionObject txObject = new DataSourceTransactionObject();
        
        // 2. 设置是否允许保存点（NESTED 传播行为需要）
        txObject.setSavepointAllowed(isNestedTransactionAllowed());
        
        // 3. 从 TransactionSynchronizationManager 获取当前线程绑定的连接
        //    这是事务传播的关键：同一个线程的多次调用共享同一个 Connection
        ConnectionHolder conHolder = (ConnectionHolder)
            TransactionSynchronizationManager.getResource(obtainDataSource());
        txObject.setConnectionHolder(conHolder, false);
        
        return txObject;
    }
    
    private static class DataSourceTransactionObject extends JdbcTransactionObjectSupport {
        private boolean newConnectionHolder;  // 是否新创建了连接
        private boolean mustRestoreAutoCommit; // 恢复 autoCommit 标志
        
        public void setConnectionHolder(@Nullable ConnectionHolder connectionHolder, 
                boolean newConnectionHolder) {
            super.setConnectionHolder(connectionHolder);
            this.newConnectionHolder = newConnectionHolder;
        }
    }
}

2.2 doBegin（开启事务 — 核心） ​

java
@Override
protected void doBegin(Object transaction, TransactionDefinition definition) {
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
    Connection con = null;
    
    try {
        // 1. 如果当前事务对象还没有 ConnectionHolder（没有已有连接）
        if (!txObject.hasConnectionHolder() || 
                txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
            // 从数据源获取一个新的数据库连接
            Connection newCon = obtainDataSource().getConnection();
            txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
        }
        
        // 2. 设置连接持有者与事务同步
        txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
        con = txObject.getConnectionHolder().getConnection();
        
        // 3. 设置事务隔离级别
        Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
        txObject.setPreviousIsolationLevel(previousIsolationLevel);
        
        // 4. 关闭 autoCommit（重要！）
        //    JDBC 事务通过关闭 autoCommit 开启
        //    在 autoCommit=true 模式下，每条 SQL 都是独立事务
        if (con.getAutoCommit() != false) {
            txObject.setMustRestoreAutoCommit(true);
            con.setAutoCommit(false);  // 关闭自动提交 = 开启事务
        }
        
        // 5. 设置只读提示
        prepareTransactional(con, definition);
        
        // 6. 将当前连接绑定到当前线程
        //    这样同一个线程后续的 DAO 操作都使用这个连接
        txObject.getConnectionHolder().setTransactionActive(true);
        int timeout = determineTimeout(definition);
        if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
            txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
        }
        
        // 7. 将连接绑定到 TransactionSynchronizationManager
        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);
    }
}

2.3 doCommit / doRollback（提交与回滚） ​

java
@Override
protected void doCommit(DefaultTransactionStatus status) {
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
    Connection con = txObject.getConnectionHolder().getConnection();
    
    // 执行 JDBC Connection.commit()
    con.commit();  // ← 本质就是 JDBC 的 commit!
    
    // 注意：如果 status 有保存点（NESTED），这里实际上执行的是 releaseSavepoint
    // 而不是调用 con.commit()
}

@Override
protected void doRollback(DefaultTransactionStatus status) {
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
    Connection con = txObject.getConnectionHolder().getConnection();
    
    // 执行 JDBC Connection.rollback()
    con.rollback();  // ← 本质就是 JDBC 的 rollback!
}

2.4 doCleanupAfterCompletion（事务完成后的清理） ​

java
@Override
protected void doCleanupAfterCompletion(Object transaction) {
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
    
    // 1. 如果当前是新连接，从线程解绑
    if (txObject.isNewConnectionHolder()) {
        TransactionSynchronizationManager.unbindResource(obtainDataSource());
    }
    
    Connection con = txObject.getConnectionHolder().getConnection();
    
    try {
        // 2. 恢复 autoCommit 原始状态
        if (txObject.isMustRestoreAutoCommit()) {
            con.setAutoCommit(true);
        }
        
        // 3. 重置隔离级别
        DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel());
    } catch (Throwable ex) {
        logger.debug("Could not reset JDBC Connection after transaction", ex);
    }
    
    // 4. 如果是新连接，释放回连接池
    if (txObject.isNewConnectionHolder()) {
        DataSourceUtils.releaseConnection(con, obtainDataSource());
    }
    
    // 5. 清空连接持有者
    txObject.setConnectionHolder(null, false);
}

3. AbstractPlatformTransactionManager — 顶级骨架实现 ​

java
public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
    
    // ===== 传播行为处理的核心逻辑 =====
    
    @Override
    public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) {
        // 1. 获取当前线程的已有事务
        Object transaction = doGetTransaction();
        
        // 2. 如果定义了 propagration 行为，缓存它
        //    否则使用默认（REQUIRED）
        if (definition != null && !definition.isReadOnly()) {
            // ...
        }
        
        TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
        
        // 3. 检查当前线程是否已有事务
        //    通过 TransactionSynchronizationManager 检查
        if (isExistingTransaction(transaction)) {
            // 已有事务 → 根据传播行为处理
            return handleExistingTransaction(def, transaction, debugEnabled);
        }
        
        // 4. 没有已有事务 → 检查超时时间
        if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
            throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
        }
        
        // 5. 没有已有事务 → 根据传播行为决定
        if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
            // MANDATORY 必须有事务，没有则抛异常
            throw new IllegalTransactionStateException(
                "No existing transaction found for transaction marked with propagation 'mandatory'");
        } else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
                   def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
                   def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
            // REQUIRED / REQUIRES_NEW / NESTED 在没有事务时都创建新事务
            return startNewTransaction(def, transaction, debugEnabled);
        } else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_SUPPORTS ||
                   def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED ||
                   def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
            // SUPPORTS / NOT_SUPPORTED / NEVER 在没有事务时都以非事务运行
            return prepareNonTransaction(def, transaction);
        } else {
            throw new IllegalTransactionStateException("Unsupported propagation behavior: " + def.getPropagationBehavior());
        }
    }
}

3.1 handleExistingTransaction — 已有事务时的传播行为处理 ​

java
private TransactionStatus handleExistingTransaction(
        TransactionDefinition definition, Object transaction, boolean debugEnabled) {
    
    // PROPAGATION_NEVER：有事务就抛异常
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
        throw new IllegalTransactionStateException(
            "Existing transaction found for transaction marked with propagation 'never'");
    }
    
    // PROPAGATION_NOT_SUPPORTED：挂起当前事务，非事务执行
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
        // 挂起当前事务，保存其状态以便后续恢复
        Object suspendedResources = suspend(transaction);
        // 创建非事务状态
        boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
        return prepareTransactionStatus(
            transaction, false, false, newSynchronization, debugEnabled, suspendedResources);
    }
    
    // PROPAGATION_REQUIRES_NEW：挂起当前事务，创建全新事务
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
        // 挂起当前事务（保存 Connection、同步器等）
        SuspendedResourcesHolder suspendedResources = suspend(transaction);
        try {
            // 创建全新的独立事务
            return startNewTransaction(definition, transaction, debugEnabled, suspendedResources);
        } catch (RuntimeException | Error ex) {
            // 创建失败：恢复被挂起的事务
            resumeAfterBeginException(transaction, suspendedResources, ex);
            throw ex;
        }
    }
    
    // PROPAGATION_NESTED：创建保存点（嵌套事务）
    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
        if (useSavepointForNestedTransaction()) {
            // 默认使用 JDBC 3.0 保存点机制
            // 创建保存点，后续可以只回滚到保存点而不是全部
            DefaultTransactionStatus status = 
                prepareTransactionStatus(transaction, false, false, true, debugEnabled, null);
            status.createAndHoldSavepoint();  // ← 关键：创建保存点
            return status;
        } else {
            // 如果不支持保存点，退化为 REQUIRES_NEW 行为
            return startNewTransaction(definition, transaction, debugEnabled, null);
        }
    }
    
    // PROPAGATION_REQUIRED（默认）：加入当前事务
    // PROPAGATION_SUPPORTS：加入当前事务
    // PROPAGATION_MANDATORY：加入当前事务
    boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    return prepareTransactionStatus(transaction, false, false, newSynchronization, debugEnabled, null);
}

核心面试题 ​

  1. Spring 声明式事务的底层实现原理？从 @Transactional 到数据库 commit() 的完整调用链？
  2. REQUIRES_NEW 在同一个事务管理器下真的会新建物理连接吗？挂起和恢复的机制？
  3. NESTED 传播行为底层如何实现？如何做到只回滚部分操作？

← 返回 **返回 事务管理索引**

Knowledge4J — Java 知识库

MIT License