part4-datasource-transaction-manager

约 3021 字大约 10 分钟

2026-05-24

系列文章:《深入Spring事务与数据源:从注解到数据库连接的全链路解析》
第四篇:DataSourceTransactionManager与数据源协作机制
基于版本:Spring Framework 5.3.x
前置阅读:第三篇:事务管理器的核心

前言

在前面的文章中,我们深入分析了Spring事务管理的核心机制,但一直没有回答一个关键问题:

事务管理器是如何从数据源获取数据库连接的?

@Override
protected void doBegin(Object transaction, TransactionDefinition definition) {
    // 这一行代码背后发生了什么?
    Connection newCon = obtainDataSource().getConnection();
}

本文将深入分析DataSourceTransactionManager的完整实现,揭示它如何与数据源(连接池)协作,管理数据库连接的生命周期。

一、DataSourceTransactionManager的架构设计

1.1 类的继承体系

PlatformTransactionManager (接口)
    ↑
    │ getTransaction()
    │ commit()
    │ rollback()
    │
AbstractPlatformTransactionManager (抽象类)
    ↑ 实现标准事务管理流程
    │
DataSourceTransactionManager (具体实现)
    ↑ 实现JDBC特定的事务管理
    │
    ├─ doGetTransaction()
    ├─ doBegin()
    ├─ doCommit()
    ├─ doRollback()
    └─ doCleanupAfterCompletion()

1.2 核心职责

DataSourceTransactionManager的核心职责:

管理JDBC事务:通过Connection的setAutoCommit(false)、commit()、rollback()
绑定连接到线程:确保同一事务内使用同一个连接
与连接池协作:从数据源获取连接,使用后归还

二、DataSourceTransactionManager的完整实现

2.1 核心属性

源码位置:spring-jdbc/src/main/java/org/springframework/jdbc/datasource/DataSourceTransactionManager.java

public class DataSourceTransactionManager extends AbstractPlatformTransactionManager
        implements ResourceTransactionManager, InitializingBean {

    // 数据源(通常是连接池)
    @Nullable
    private DataSource dataSource;
    
    // 是否强制只读
    private boolean enforceReadOnly = false;
    
    // 构造函数
    public DataSourceTransactionManager() {
        setNestedTransactionAllowed(true);  // 默认允许嵌套事务
    }
    
    public DataSourceTransactionManager(DataSource dataSource) {
        this();
        setDataSource(dataSource);
        afterPropertiesSet();
    }
}

2.2 数据源的设置

public void setDataSource(@Nullable DataSource dataSource) {
    // 如果是TransactionAwareDataSourceProxy,需要解包
    if (dataSource instanceof TransactionAwareDataSourceProxy) {
        this.dataSource = ((TransactionAwareDataSourceProxy) dataSource)
                .getTargetDataSource();
    }
    else {
        this.dataSource = dataSource;
    }
}

TransactionAwareDataSourceProxy:

这是一个DataSource代理
使标准JDBC代码也能参与Spring事务
通常不需要,因为Spring的JdbcTemplate等已经支持事务

三、doGetTransaction:获取当前事务对象

@Override
protected Object doGetTransaction() {
    // 1. 创建事务对象
    DataSourceTransactionObject txObject = new DataSourceTransactionObject();
    txObject.setSavepointAllowed(isNestedTransactionAllowed());
    
    // 2. 从TransactionSynchronizationManager获取ConnectionHolder
    ConnectionHolder conHolder =
            (ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
    
    // 3. 设置到事务对象
    txObject.setConnectionHolder(conHolder, false);
    
    return txObject;
}

关键点:

创建DataSourceTransactionObject封装事务状态
从TransactionSynchronizationManager查找当前线程绑定的ConnectionHolder
如果找到了,说明已经存在事务
如果没找到(conHolder == null),说明还没有事务

3.1 DataSourceTransactionObject结构

private static class DataSourceTransactionObject extends JdbcTransactionObjectSupport {

    // 是否是新创建的ConnectionHolder
    private boolean newConnectionHolder;
    
    // 是否需要恢复自动提交
    private boolean mustRestoreAutoCommit;

    public void setConnectionHolder(@Nullable ConnectionHolder connectionHolder, 
                                    boolean newConnectionHolder) {
        super.setConnectionHolder(connectionHolder);
        this.newConnectionHolder = newConnectionHolder;
    }
    
    public boolean isNewConnectionHolder() {
        return this.newConnectionHolder;
    }
    
    public void setMustRestoreAutoCommit(boolean mustRestoreAutoCommit) {
        this.mustRestoreAutoCommit = mustRestoreAutoCommit;
    }
    
    public boolean isMustRestoreAutoCommit() {
        return this.mustRestoreAutoCommit;
    }
    
    public void setRollbackOnly() {
        getConnectionHolder().setRollbackOnly();
    }
    
    @Override
    public boolean isRollbackOnly() {
        return getConnectionHolder().isRollbackOnly();
    }
}

职责:

封装ConnectionHolder
标记是否是新连接(用于后续清理)
标记是否需要恢复自动提交
支持savepoint(嵌套事务)

四、isExistingTransaction:判断是否存在事务

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

判断标准:

hasConnectionHolder():有ConnectionHolder
isTransactionActive():连接的事务状态为活跃

ConnectionHolder.isTransactionActive():

public boolean isTransactionActive() {
    return this.transactionActive;
}

这个标记在doBegin()方法中设置为true。

五、doBegin:开启事务(核心方法)

这是DataSourceTransactionManager最核心的方法,真正实现了数据库事务的开启。

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

    try {
        // ========== 步骤1:获取物理连接 ==========
        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);
        }

        // ========== 步骤2:标记连接与事务同步 ==========
        txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
        con = txObject.getConnectionHolder().getConnection();

        // ========== 步骤3:设置事务属性 ==========
        // 设置隔离级别
        Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(
            con, definition);
        txObject.setPreviousIsolationLevel(previousIsolationLevel);
        // 设置只读标记
        txObject.setReadOnly(definition.isReadOnly());

        // ========== 步骤4:关闭自动提交(关键!) ==========
        if (con.getAutoCommit()) {
            txObject.setMustRestoreAutoCommit(true);
            if (logger.isDebugEnabled()) {
                logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
            }
            con.setAutoCommit(false);  // ← 这就是开启事务!
        }

        // ========== 步骤5:处理只读事务 ==========
        prepareTransactionalConnection(con, definition);
        
        // ========== 步骤6:标记事务活跃 ==========
        txObject.getConnectionHolder().setTransactionActive(true);

        // ========== 步骤7:设置超时时间 ==========
        int timeout = determineTimeout(definition);
        if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
            txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
        }

        // ========== 步骤8:绑定到ThreadLocal ==========
        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);
    }
}

5.1 步骤详解

步骤1:获取物理连接

Connection newCon = obtainDataSource().getConnection();

这里的obtainDataSource()返回什么?

场景1:原生Druid数据源

@Bean
public DataSource dataSource() {
    DruidDataSource ds = new DruidDataSource();
    ds.setUrl("jdbc:mysql://localhost:3306/test");
    ds.setUsername("root");
    ds.setPassword("password");
    ds.setInitialSize(5);
    ds.setMaxActive(20);
    return ds;
}

// getConnection()的实现
public Connection getConnection() throws SQLException {
    // 1. 从连接池获取空闲连接
    DruidConnectionHolder holder = pool.pollFirst();
    
    if (holder == null) {
        // 2. 如果连接池为空,创建新连接
        holder = createPhysicalConnection();
    }
    
    return holder.getConnection();
}

场景2:Spring Boot自动配置的HikariCP

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/test
    username: root
    password: password

Spring Boot默认使用HikariCP:

@Bean
public DataSource dataSource() {
    HikariDataSource ds = new HikariDataSource();
    ds.setJdbcUrl("jdbc:mysql://localhost:3306/test");
    ds.setUsername("root");
    ds.setPassword("password");
    return ds;
}

步骤3:设置事务属性

Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(
    con, definition);

DataSourceUtils.prepareConnectionForTransaction:

@Nullable
public static Integer prepareConnectionForTransaction(Connection con,
        @Nullable TransactionDefinition definition) throws SQLException {
    
    // 1. 设置只读标记(使用JDBC hint)
    if (definition != null && definition.isReadOnly()) {
        con.setReadOnly(true);
    }
    
    // 2. 设置隔离级别
    Integer previousIsolationLevel = null;
    if (definition != null && definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
        previousIsolationLevel = con.getTransactionIsolation();
        con.setTransactionIsolation(definition.getIsolationLevel());
    }
    
    return previousIsolationLevel;
}

隔离级别对照:

TransactionDefinition.ISOLATION_READ_UNCOMMITTED = 1  // 读未提交
TransactionDefinition.ISOLATION_READ_COMMITTED = 2    // 读已提交
TransactionDefinition.ISOLATION_REPEATABLE_READ = 4   // 可重复读
TransactionDefinition.ISOLATION_SERIALIZABLE = 8      // 串行化

步骤4:关闭自动提交

if (con.getAutoCommit()) {
    txObject.setMustRestoreAutoCommit(true);
    con.setAutoCommit(false);  // ← 开启事务的关键!
}

底层实现(MySQL驱动):

// MySQL JDBC Driver
public void setAutoCommit(boolean autoCommit) throws SQLException {
    if (this.autoCommit != autoCommit) {
        this.autoCommit = autoCommit;
        
        if (!autoCommit) {
            // 发送SQL关闭自动提交
            executeSql("SET autocommit=0");
        } else {
            executeSql("SET autocommit=1");
        }
    }
}

效果:

autocommit=1:每条SQL执行后自动提交
autocommit=0:需要手动COMMIT或ROLLBACK

步骤5:处理只读事务

prepareTransactionalConnection(con, definition);

protected void prepareTransactionalConnection(Connection con, TransactionDefinition definition)
        throws SQLException {
    
    // 如果是只读事务,执行SET TRANSACTION READ ONLY
    if (isEnforceReadOnly() && definition.isReadOnly()) {
        try (Statement stmt = con.createStatement()) {
            stmt.executeUpdate("SET TRANSACTION READ ONLY");
        }
    }
}

MySQL的只读事务:

SET TRANSACTION READ ONLY;
-- 之后只能执行SELECT,不能执行INSERT/UPDATE/DELETE

步骤8:绑定到ThreadLocal

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

这是确保同一事务内使用同一个连接的关键!

完整流程:

第一次调用doBegin:
  isNewConnectionHolder = true
  → bindResource(dataSource, connectionHolder)
  → ThreadLocal绑定

后续在同一事务内调用:
  doGetTransaction()
  → getResource(dataSource)
  → 从ThreadLocal获取connectionHolder
  → isExistingTransaction = true
  → 加入现有事务

六、doCommit:提交事务

@Override
protected void doCommit(DefaultTransactionStatus status) {
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
    Connection con = txObject.getConnectionHolder().getConnection();
    
    if (status.isDebug()) {
        logger.debug("Committing JDBC transaction on Connection [" + con + "]");
    }
    
    try {
        con.commit();  // ← JDBC提交
    }
    catch (SQLException ex) {
        throw translateException("JDBC commit", ex);
    }
}

底层实现(MySQL驱动):

public void commit() throws SQLException {
    executeSql("COMMIT");
}

七、doRollback:回滚事务

@Override
protected void doRollback(DefaultTransactionStatus status) {
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
    Connection con = txObject.getConnectionHolder().getConnection();
    
    if (status.isDebug()) {
        logger.debug("Rolling back JDBC transaction on Connection [" + con + "]");
    }
    
    try {
        con.rollback();  // ← JDBC回滚
    }
    catch (SQLException ex) {
        throw translateException("JDBC rollback", ex);
    }
}

底层实现(MySQL驱动):

public void rollback() throws SQLException {
    executeSql("ROLLBACK");
}

八、doCleanupAfterCompletion:清理事务资源

@Override
protected void doCleanupAfterCompletion(Object transaction) {
    DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;

    // 1. 从ThreadLocal移除ConnectionHolder
    if (txObject.isNewConnectionHolder()) {
        TransactionSynchronizationManager.unbindResource(obtainDataSource());
    }

    // 2. 重置连接状态
    Connection con = txObject.getConnectionHolder().getConnection();
    try {
        // 恢复自动提交
        if (txObject.isMustRestoreAutoCommit()) {
            con.setAutoCommit(true);
        }
        // 重置隔离级别和只读标记
        DataSourceUtils.resetConnectionAfterTransaction(
                con, txObject.getPreviousIsolationLevel(), txObject.isReadOnly());
    }
    catch (Throwable ex) {
        logger.debug("Could not reset JDBC Connection after transaction", ex);
    }

    // 3. 释放连接(返回连接池)
    if (txObject.isNewConnectionHolder()) {
        if (logger.isDebugEnabled()) {
            logger.debug("Releasing JDBC Connection [" + con + "] after transaction");
        }
        DataSourceUtils.releaseConnection(con, this.dataSource);
    }

    // 4. 清空ConnectionHolder
    txObject.getConnectionHolder().clear();
}

8.1 DataSourceUtils.releaseConnection

public static void releaseConnection(@Nullable Connection con, @Nullable DataSource dataSource) {
    if (con != null && dataSource != null) {
        try {
            // 调用DataSource的release方法
            // 对于连接池,这是归还连接到池
            // 对于普通DataSource,这是关闭连接
            con.close();
        }
        catch (SQLException ex) {
            logger.debug("Could not close JDBC Connection", ex);
        }
    }
}

连接池的实现:

// Druid连接池
public void close() throws SQLException {
    // 不是真正关闭连接
    // 而是将连接归还到连接池
    pool.offerLast(this.connectionHolder);
}

九、ConnectionHolder:连接的封装

9.1 ConnectionHolder的作用

ConnectionHolder是Spring对Connection的封装,用于:

管理连接的生命周期
存储事务状态
支持引用计数(嵌套事务)
支持超时检查

9.2 ConnectionHolder的核心属性

public class ConnectionHolder extends ResourceHolderSupport {

    // 数据库连接
    @Nullable
    private Connection connection;
    
    // 是否是新创建的连接
    private boolean newConnection;
    
    // 事务是否活跃
    private boolean transactionActive = false;
    
    // 是否同步
    private boolean synchronizedWithTransaction = false;
    
    // 是否只读
    private boolean readOnly = false;
    
    // 之前的隔离级别(用于恢复)
    @Nullable
    private Integer previousIsolationLevel;
    
    // 引用计数(嵌套事务)
    private int referenceCount = 0;
    
    // 超时时间
    @Nullable
    private Integer timeout;
    
    // 截止时间
    @Nullable
    private Date deadline;
}

9.3 ResourceHolderSupport:引用计数

public abstract class ResourceHolderSupport {

    // 引用计数
    private int referenceCount = 0;
    
    // 是否已释放
    private boolean released = false;

    public void requested() {
        this.referenceCount++;
    }

    public void released() {
        this.referenceCount--;
    }

    public int getReferenceCount() {
        return this.referenceCount;
    }

    public boolean isOpen() {
        return (this.referenceCount > 0);
    }
}

应用场景:嵌套事务

@Transactional  // 外层事务
public void outerMethod() {
    // referenceCount = 1
    
    innerMethod();  // 内层事务(REQUIRES_NEW)
    // 挂起外层:保存referenceCount
    // 内层:referenceCount = 1
    
    // 内层结束:恢复外层
    // referenceCount = 1
}

十、与连接池的完整协作流程

10.1 使用Druid连接池的完整流程

// 1. 配置Druid数据源
@Bean
public DataSource dataSource() {
    DruidDataSource ds = new DruidDataSource();
    ds.setUrl("jdbc:mysql://localhost:3306/test");
    ds.setUsername("root");
    ds.setPassword("password");
    ds.setInitialSize(5);      // 初始连接数
    ds.setMaxActive(20);       // 最大活跃连接
    ds.setMinIdle(5);          // 最小空闲连接
    ds.setMaxWait(60000);      // 获取连接最大等待时间
    return ds;
}

// 2. 配置事务管理器
@Bean
public PlatformTransactionManager transactionManager() {
    return new DataSourceTransactionManager(dataSource());
}

// 3. 业务代码
@Service
public class UserService {
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    @Transactional
    public void createUser(User user) {
        jdbcTemplate.update("INSERT INTO users ...");
    }
}

10.2 完整的执行流程

┌─────────────────────────────────────────────────────────┐
│  调用: userService.createUser(user)                      │
└────────────────────┬────────────────────────────────────┘
                     │
                     ↓
┌─────────────────────────────────────────────────────────┐
│  DataSourceTransactionManager.doBegin()                 │
│                                                         │
│  1. dataSource.getConnection()                          │
│     → DruidDataSource.getConnection()                   │
│     → 从连接池获取连接                                    │
│     → 如果连接池为空,创建新连接                           │
│     → 返回DruidPooledConnection                          │
│                                                         │
│  2. con.setAutoCommit(false)                            │
│     → 发送SQL: SET autocommit=0                         │
│                                                         │
│  3. TransactionSynchronizationManager.bindResource()    │
│     → 将Connection绑定到ThreadLocal                      │
└────────────────────┬────────────────────────────────────┘
                     │
                     ↓
┌─────────────────────────────────────────────────────────┐
│  执行业务: jdbcTemplate.update(...)                      │
│                                                         │
│  1. DataSourceUtils.getConnection(dataSource)           │
│     → 从ThreadLocal获取ConnectionHolder                  │
│     → 返回已绑定的Connection                             │
│                                                         │
│  2. 使用Connection执行SQL                                │
│     → PreparedStatement.executeQuery()                  │
│     → 发送到MySQL服务器                                  │
└────────────────────┬────────────────────────────────────┘
                     │
                     ↓
┌─────────────────────────────────────────────────────────┐
│  DataSourceTransactionManager.doCommit()                │
│                                                         │
│  1. con.commit()                                        │
│     → 发送SQL: COMMIT                                   │
│                                                         │
│  2. doCleanupAfterCompletion()                          │
│     → 恢复autoCommit=true                               │
│     → con.close()  ← 归还到连接池(不是真正关闭)          │
└─────────────────────────────────────────────────────────┘

十一、关键问题解答

Q1:为什么需要TransactionSynchronizationManager?

问题:直接从DataSource获取连接不行吗?为什么要绑定到ThreadLocal?

答案:确保同一事务内使用同一个连接!

// 如果没有ThreadLocal绑定
@Transactional
public void method() {
    Connection conn1 = dataSource.getConnection();  // conn1
    jdbcTemplate.update(...);  // 可能获取conn2
    
    // conn1和conn2不是同一个连接!
    // 事务失效!
}

// 有了ThreadLocal绑定
@Transactional
public void method() {
    Connection conn1 = dataSource.getConnection();  // conn1,绑定到ThreadLocal
    jdbcTemplate.update(...);  // 从ThreadLocal获取conn1
    
    // conn1和conn1是同一个连接!
    // 事务生效!
}

Q2:连接池中的连接归还后,事务状态会怎样?

// 事务提交后
con.commit();

// 清理时
con.setAutoCommit(true);  // 恢复自动提交
con.close();  // 归还到连接池

连接归还后的状态:

autoCommit = true:恢复为默认值
隔离级别:恢复为默认值
只读标记:清除
连接本身:不关闭,归还到连接池

下一个事务使用时:

// 从连接池获取连接
Connection conn = dataSource.getConnection();

// 这是一个"干净"的连接
// autoCommit = true
// 隔离级别 = 默认
// 没有事务

Q3:多线程环境下,线程池复用线程会导致事务混乱吗?

不会! 原因:

ThreadLocal会在事务完成后清理:

doCleanupAfterCompletion() {
    TransactionSynchronizationManager.unbindResource(dataSource);
    // ThreadLocal被清除
}

线程池复用线程时,ThreadLocal已经被清理:

线程1:
  执行事务A
  → ThreadLocal绑定Connection
  → 事务提交
  → ThreadLocal清理
  → 归还到线程池

线程1被复用:
  ThreadLocal为null
  → 执行事务B
  → 重新绑定新的Connection

十二、总结与伏笔

通过本文的分析,我们完整理解了DataSourceTransactionManager与数据源的协作机制:

doGetTransaction:从ThreadLocal查找现有连接
doBegin:从数据源获取连接,关闭自动提交,绑定到ThreadLocal
doCommit:调用Connection.commit()提交事务
doRollback:调用Connection.rollback()回滚事务
doCleanupAfterCompletion:重置连接状态,归还到连接池
ConnectionHolder:封装连接,管理事务状态和引用计数

但还有一个关键问题:TransactionSynchronizationManager的ThreadLocal机制是如何实现的?

ThreadLocal的底层数据结构是什么?
bindResource和getResource的完整实现?
如何支持多个数据源?
事务同步回调的注册与触发机制?

下一篇预告:《ThreadLocal的魔法:TransactionSynchronizationManager资源管理揭秘》,我们将深入分析ThreadLocal在事务管理中的精妙应用。

参考资料

Spring Framework 5.3.x 源码
spring-jdbc/src/main/java/org/springframework/jdbc/datasource/DataSourceTransactionManager.java
spring-tx/src/main/java/org/springframework/transaction/support/TransactionSynchronizationManager.java
Druid源码: DruidDataSource.getConnection()
《Spring源码深度解析》

版权所有

版权归属：今朝· Liushigong · Relax Box

许可证：署名-非商业性 4.0 国际 (CC-BY-NC-4.0)