PlatformTransactionManager 事务管理器

概述

PlatformTransactionManager 是 Spring 事务管理的核心接口，它是 Spring 框架中事务管理抽象层的顶层接口，定义了事务管理的三个基本操作：获取事务、提交事务和回滚事务。这个接口的设计体现了策略模式的思想，为不同的事务管理实现提供了统一的编程接口。

Spring 通过这个接口实现了对各种事务技术的统一管理，包括 JDBC 事务(DataSourceTransactionManager)、JTA 全局事务(JtaTransactionManager)、JPA 事务(JpaTransactionManager)、响应式事务（ReactiveTransactionManager）等。无论底层使用何种事务技术，开发者都可以通过相同的编程模型来管理事务，这大大提高了代码的可移植性和维护性。

本文将从源码层面深入分析 PlatformTransactionManager 的设计原理、实现机制和扩展点，帮助读者全面理解 Spring 事务管理的内部工作机制。

事务管理整体架构

核心接口源码分析

PlatformTransactionManager 接口定义

PlatformTransactionManager 接口设计非常简洁，只定义了三个核心方法。这种简洁的设计体现了单一职责原则，专注于事务的生命周期管理。

package org.springframework.transaction;

public interface PlatformTransactionManager extends TransactionManager {

    /**
     * 根据事务定义获取事务状态
     * 这是最复杂的方法，需要处理事务传播行为、隔离级别、超时设置等
     * 可能返回一个新的事务或现有的事务（如果当前线程已经有匹配的事务）
     *
     * @param definition 事务定义，包含传播行为、隔离级别、超时等配置
     * @return 事务状态对象，包含事务的当前状态和相关信息
     * @throws TransactionException 获取事务失败时抛出异常
     */
    TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
            throws TransactionException;

    /**
     * 提交事务
     * 如果事务已被标记为仅回滚（rollback-only），则执行回滚操作而不是提交
     * 这种设计确保了事务的一致性，即使上层代码调用提交，但底层发现异常时仍会回滚
     *
     * @param status 事务状态，由 getTransaction 方法返回
     * @throws TransactionException 提交失败时抛出异常
     */
    void commit(TransactionStatus status) throws TransactionException;

    /**
     * 回滚事务
     * 撤销事务中的所有数据库操作，恢复到事务开始前的状态
     *
     * @param status 事务状态，由 getTransaction 方法返回
     * @throws TransactionException 回滚失败时抛出异常
     */
    void rollback(TransactionStatus status) throws TransactionException;
}

这三个方法构成了事务管理的完整生命周期：获取事务 → 执行业务逻辑 → 提交或回滚事务。接口的设计遵循了以下原则：

• 职责单一：每个方法都有明确的职责，不承担额外的功能
• 异常安全：所有方法都声明了 TransactionException，确保异常情况得到正确处理
• 状态管理：通过 TransactionStatus 对象维护事务状态，实现了状态的一致性管理

核心方法详解

getTransaction 方法的复杂性分析

getTransaction 方法是整个事务管理机制中最复杂的部分，它需要处理各种事务传播行为的组合情况。该方法的执行流程可以用以下流程图表示：

getTransaction 方法源码分析

以下是 AbstractPlatformTransactionManager 中 getTransaction 方法的具体实现，这个方法体现了模板方法模式的经典应用：

// AbstractPlatformTransactionManager 中的实现
@Override
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
        throws TransactionException {

    // 1. 获取事务定义，如果为null则使用默认定义
    TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());

    // 2. 获取事务对象（由具体实现类提供）
    Object transaction = doGetTransaction();
    boolean debugEnabled = logger.isDebugEnabled();

    // 3. 检查是否已存在事务
    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) {
        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) {
        // 创建新事务
        SuspendedResourcesHolder suspendedResources = suspend(null);
        try {
            return startTransaction(def, transaction, debugEnabled, suspendedResources);
        }
        catch (RuntimeException | Error ex) {
            resume(null, suspendedResources);
            throw ex;
        }
    }
    else {
        // 创建"空"事务：没有实际事务，但可能有同步
        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);
    }
}

事务传播行为详细说明

事务传播行为是 Spring 事务管理的重要特性，它定义了在存在事务的情况下，方法如何在事务上下文中执行。理解传播行为对于正确使用 Spring 事务至关重要。

各种传播行为的使用场景和特点：

• REQUIRED（最常用）：适用于大部分业务方法，确保操作在事务中执行
• REQUIRES_NEW：适用于需要独立提交的操作，如日志记录、审计等
• NESTED：适用于需要部分回滚能力的场景，如批量处理中的单个项目失败
• SUPPORTS：适用于查询操作，既可以在事务中执行也可以独立执行
• NOT_SUPPORTED：适用于不需要事务的操作，如读取配置信息等
• MANDATORY：适用于必须在现有事务中执行的操作，确保数据一致性
• NEVER：适用于禁止在事务中执行的操作，如某些第三方系统调用

抽象基类源码分析

AbstractPlatformTransactionManager 设计分析

AbstractPlatformTransactionManager 是 Spring 事务管理器的抽象基类，它实现了 PlatformTransactionManager 接口的大部分逻辑。这个类采用了模板方法模式，将事务管理的通用流程在父类中实现，而将具体的事务技术相关的操作留给子类实现。

这种设计的优势在于：

1. 代码复用：通用的事务管理逻辑只需要实现一次
2. 扩展性强：新的事务技术只需要实现几个抽象方法即可
3. 维护性好：通用逻辑的修改不会影响具体实现
4. 一致性保证：所有实现都遵循相同的事务管理流程

public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {

    // 事务同步策略
    public static final int SYNCHRONIZATION_ALWAYS = 0;
    public static final int SYNCHRONIZATION_ON_ACTUAL_TRANSACTION = 1;
    public static final int SYNCHRONIZATION_NEVER = 2;

    // 默认事务超时时间
    public static final int DEFAULT_TIMEOUT = TransactionDefinition.TIMEOUT_DEFAULT;

    // 事务同步设置
    private int transactionSynchronization = SYNCHRONIZATION_ALWAYS;

    // 默认超时时间
    private int defaultTimeout = DEFAULT_TIMEOUT;

    // 是否嵌套事务回滚到保存点
    private boolean nestedTransactionAllowed = false;

    // 验证现有事务
    private boolean validateExistingTransaction = false;

    // 是否在提交失败时全局回滚
    private boolean globalRollbackOnParticipationFailure = true;

    // 是否在回滚时调用doRollback
    private boolean rollbackOnCommitFailure = false;

    //=== 抽象方法，由子类实现 ===

    /**
     * 获取事务对象
     */
    protected abstract Object doGetTransaction() throws TransactionException;

    /**
     * 检查是否存在现有事务
     */
    protected boolean isExistingTransaction(Object transaction) throws TransactionException {
        return false;
    }

    /**
     * 开始新事务
     */
    protected abstract void doBegin(Object transaction, TransactionDefinition definition)
            throws TransactionException;

    /**
     * 挂起当前事务
     */
    @Nullable
    protected Object doSuspend(Object transaction) throws TransactionException {
        throw new TransactionSuspensionNotSupportedException(
                "Transaction manager [" + getClass().getName() + "] does not support transaction suspension");
    }

    /**
     * 恢复挂起的事务
     */
    protected void doResume(@Nullable Object transaction, Object suspendedResources)
            throws TransactionException {
        throw new TransactionSuspensionNotSupportedException(
                "Transaction manager [" + getClass().getName() + "] does not support transaction suspension");
    }

    /**
     * 提交事务
     */
    protected abstract void doCommit(DefaultTransactionStatus status) throws TransactionException;

    /**
     * 回滚事务
     */
    protected abstract void doRollback(DefaultTransactionStatus status) throws TransactionException;
}

事务状态管理

// DefaultTransactionStatus 类
public class DefaultTransactionStatus extends AbstractTransactionStatus {

    @Nullable
    private final Object transaction;

    private final boolean newTransaction;

    private final boolean newSynchronization;

    private final boolean readOnly;

    private final boolean debug;

    @Nullable
    private final Object suspendedResources;

    // 构造函数
    public DefaultTransactionStatus(
            @Nullable Object transaction, boolean newTransaction, boolean newSynchronization,
            boolean readOnly, boolean debug, @Nullable Object suspendedResources) {

        this.transaction = transaction;
        this.newTransaction = newTransaction;
        this.newSynchronization = newSynchronization;
        this.readOnly = readOnly;
        this.debug = debug;
        this.suspendedResources = suspendedResources;
    }

    // 获取事务对象
    @Override
    public Object getTransaction() {
        Assert.state(this.transaction != null, "No transaction active");
        return this.transaction;
    }

    // 是否为新事务
    @Override
    public boolean isNewTransaction() {
        return (hasTransaction() && this.newTransaction);
    }

    // 设置回滚标记
    @Override
    public void setRollbackOnly() {
        super.setRollbackOnly();
        // 如果有事务对象，也要在事务对象上设置回滚标记
        if (this.transaction instanceof SmartTransactionObject) {
            ((SmartTransactionObject) this.transaction).rollbackOnly();
        }
    }
}

主要实现类源码分析

DataSourceTransactionManager 实现原理

DataSourceTransactionManager 是基于 JDBC DataSource 的事务管理器实现，它是 Spring 中最常用的事务管理器之一。该类专门针对单一数据源的 JDBC 事务进行了优化，提供了完整的本地事务管理功能。

核心工作原理

关键设计特点

1. 连接管理：通过 ConnectionHolder 管理数据库连接的生命周期
2. 线程绑定：使用 TransactionSynchronizationManager 将连接绑定到当前线程
3. 自动提交控制：自动关闭连接的自动提交模式，确保事务的原子性
4. 资源清理：事务结束后自动清理连接资源，避免连接泄露

public class DataSourceTransactionManager extends AbstractPlatformTransactionManager
        implements ResourceTransactionManager, InitializingBean {

    @Nullable
    private DataSource dataSource;

    private boolean enforceReadOnly = false;

    // 事务对象内部类
    private static final class DataSourceTransactionObject extends JdbcTransactionObjectSupport {

        private boolean newConnectionHolder;

        private boolean mustRestoreAutoCommit;

        public void setConnectionHolder(@Nullable ConnectionHolder connectionHolder, boolean newConnectionHolder) {
            super.setConnectionHolder(connectionHolder);
            this.newConnectionHolder = newConnectionHolder;
        }

        // 其他方法...
    }

    @Override
    protected Object doGetTransaction() {
        DataSourceTransactionObject txObject = new DataSourceTransactionObject();
        txObject.setSavepointAllowed(isNestedTransactionAllowed());

        // 从事务同步管理器中获取连接持有者
        ConnectionHolder conHolder =
            (ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
        txObject.setConnectionHolder(conHolder, false);
        return txObject;
    }

    @Override
    protected boolean isExistingTransaction(Object transaction) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
        return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
    }

    @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();
                txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
            }

            txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
            con = txObject.getConnectionHolder().getConnection();

            // 设置只读和隔离级别
            Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
            txObject.setPreviousIsolationLevel(previousIsolationLevel);
            txObject.setReadOnly(definition.isReadOnly());

            // 关闭自动提交
            if (con.getAutoCommit()) {
                txObject.setMustRestoreAutoCommit(true);
                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);
        }
    }

    @Override
    protected void doCommit(DefaultTransactionStatus status) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
        Connection con = txObject.getConnectionHolder().getConnection();
        try {
            con.commit();
        }
        catch (SQLException ex) {
            throw new TransactionSystemException("Could not commit JDBC transaction", ex);
        }
    }

    @Override
    protected void doRollback(DefaultTransactionStatus status) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
        Connection con = txObject.getConnectionHolder().getConnection();
        try {
            con.rollback();
        }
        catch (SQLException ex) {
            throw new TransactionSystemException("Could not roll back JDBC transaction", ex);
        }
    }
}

事务传播行为源码实现

事务传播行为的实现机制

事务传播行为的处理是 Spring 事务管理中最精妙的部分之一。它通过复杂的条件判断和状态管理，实现了不同传播行为的精确控制。下面通过流程图和源码分析来深入理解这一机制：

传播行为处理源码详解

handleExistingTransaction 方法是处理现有事务情况下各种传播行为的核心方法。它的设计体现了策略模式的思想，根据不同的传播行为采用不同的处理策略：

// AbstractPlatformTransactionManager 中的实现
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) {
        // 挂起现有事务
        Object suspendedResources = suspend(transaction);
        boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
        return prepareTransactionStatus(
                definition, null, false, newSynchronization, debugEnabled, suspendedResources);
    }

    if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
        // 挂起现有事务，开始新事务
        SuspendedResourcesHolder suspendedResources = suspend(transaction);
        try {
            return startTransaction(definition, transaction, debugEnabled, suspendedResources);
        }
        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 (useSavepointForNestedTransaction()) {
            DefaultTransactionStatus status =
                    prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
            status.createAndHoldSavepoint();
            return status;
        }
        else {
            // 通过嵌套的begin和commit/rollback调用创建嵌套事务
            return startTransaction(definition, transaction, debugEnabled, null);
        }
    }

    // PROPAGATION_SUPPORTS 或 PROPAGATION_REQUIRED
    boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
}

事务同步机制源码分析

TransactionSynchronizationManager 的设计哲学

TransactionSynchronizationManager 是 Spring 事务管理中的核心组件，它采用 ThreadLocal 技术实现了事务资源的线程级隔离和管理。这个类的设计体现了以下重要原则：

1. 线程安全：每个线程维护独立的事务上下文，避免线程间的数据污染
2. 资源管理：统一管理事务相关的所有资源，如数据库连接、Session 等
3. 生命周期回调：提供事务各个阶段的回调机制，支持扩展功能
4. 透明化访问：业务代码无需关心事务资源的获取和释放

事务同步机制架构图

public abstract class TransactionSynchronizationManager {

    // 线程本地变量存储事务资源
    private static final ThreadLocal<Map<Object, Object>> resources = new ThreadLocal<>();

    // 线程本地变量存储同步器
    private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations = new ThreadLocal<>();

    // 当前事务名称
    private static final ThreadLocal<String> currentTransactionName = new ThreadLocal<>();

    // 当前事务只读标记
    private static final ThreadLocal<Boolean> currentTransactionReadOnly = new ThreadLocal<>();

    // 当前事务隔离级别
    private static final ThreadLocal<Integer> currentTransactionIsolationLevel = new ThreadLocal<>();

    // 实际事务激活标记
    private static final ThreadLocal<Boolean> actualTransactionActive = new ThreadLocal<>();

    /**
     * 获取当前线程绑定的资源
     */
    @Nullable
    public static Object getResource(Object key) {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Object value = doGetResource(actualKey);
        return value;
    }

    /**
     * 将资源绑定到当前线程
     */
    public static void bindResource(Object key, Object value) throws IllegalStateException {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Map<Object, Object> map = resources.get();

        if (map == null) {
            map = new HashMap<>();
            resources.set(map);
        }

        Object oldValue = map.put(actualKey, value);
        if (oldValue != null) {
            throw new IllegalStateException("Already value [" + oldValue + "] for key [" +
                    actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
        }
    }

    /**
     * 从当前线程解绑资源
     */
    public static Object unbindResource(Object key) throws IllegalStateException {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Object value = doUnbindResource(actualKey);
        if (value == null) {
            throw new IllegalStateException(
                    "No value for key [" + actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
        }
        return value;
    }

    /**
     * 注册事务同步器
     */
    public static void registerSynchronization(TransactionSynchronization synchronization)
            throws IllegalStateException {

        if (!isSynchronizationActive()) {
            throw new IllegalStateException("Transaction synchronization is not active");
        }

        Set<TransactionSynchronization> synchs = synchronizations.get();
        if (synchs == null) {
            synchs = new LinkedHashSet<>();
            synchronizations.set(synchs);
        }
        synchs.add(synchronization);
    }

    /**
     * 触发事务同步器的各个生命周期回调
     */
    public static void triggerBeforeCommit(boolean readOnly) {
        for (TransactionSynchronization synchronization : getSynchronizations()) {
            synchronization.beforeCommit(readOnly);
        }
    }

    public static void triggerBeforeCompletion() {
        for (TransactionSynchronization synchronization : getSynchronizations()) {
            try {
                synchronization.beforeCompletion();
            }
            catch (Throwable tsex) {
                logger.debug("TransactionSynchronization.beforeCompletion threw exception", tsex);
            }
        }
    }

    public static void triggerAfterCommit() {
        triggerAfterCommit(getSynchronizations());
    }

    private static void triggerAfterCommit(List<TransactionSynchronization> synchronizations) {
        if (synchronizations != null) {
            for (TransactionSynchronization synchronization : synchronizations) {
                synchronization.afterCommit();
            }
        }
    }

    public static void triggerAfterCompletion(int completionStatus) {
        List<TransactionSynchronization> synchronizations = getSynchronizations();
        triggerAfterCompletion(synchronizations, completionStatus);
    }

    private static void triggerAfterCompletion(List<TransactionSynchronization> synchronizations, int completionStatus) {
        if (synchronizations != null) {
            for (TransactionSynchronization synchronization : synchronizations) {
                try {
                    synchronization.afterCompletion(completionStatus);
                }
                catch (Throwable tsex) {
                    logger.debug("TransactionSynchronization.afterCompletion threw exception", tsex);
                }
            }
        }
    }
}

Spring 事务管理的完整生命周期

为了更好地理解 Spring 事务管理的整体工作机制，让我们通过一个完整的时序图来展示从事务开始到结束的全过程：

实际应用场景和最佳实践

典型应用场景

场景1：电商订单处理

@Service
public class OrderService {

    @Transactional
    public void createOrder(Order order) {
        // 在同一个事务中完成多个相关操作
        orderRepository.save(order);           // 保存订单
        inventoryService.reduceStock(order);   // 减少库存
        paymentService.processPayment(order);  // 处理支付
        // 如果任何一步失败，整个事务回滚
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void logOrderEvent(OrderEvent event) {
        // 独立事务记录日志，不受主业务事务影响
        eventLogRepository.save(event);
    }
}

场景2：批量数据处理

@Service
public class BatchProcessService {

    @Transactional
    public void processBatch(List<DataItem> items) {
        for (DataItem item : items) {
            try {
                processItem(item);  // NESTED传播行为
            } catch (Exception e) {
                // 单个项目失败不影响其他项目
                log.error("Failed to process item: " + item.getId(), e);
            }
        }
    }

    @Transactional(propagation = Propagation.NESTED)
    private void processItem(DataItem item) {
        // 使用嵌套事务处理单个项目
        // 可以回滚到保存点，不影响整体事务
        dataRepository.update(item);
    }
}

建议

1. 合理设置事务边界：事务范围应该尽可能小，只包含必要的操作
2. 避免长事务：长时间运行的事务会导致锁竞争和性能问题
3. 合理使用只读事务：对于查询操作，使用 @Transactional(readOnly = true)
4. 选择合适的传播行为：根据业务需求选择最合适的传播行为
5. 异常处理：明确哪些异常会导致事务回滚，哪些不会