Ch3nyang's blog | 深入 Spring 源码 (3) -JDBC 和事务

Spring 是一个开源的轻量级 JavaEE 框架。它的核心是控制反转（IoC）和面向切面编程（AOP）。Spring 的 IoC 容器负责管理 JavaBean 的生命周期，而 AOP 容器负责管理切面。Spring 还提供了一系列的模块，如 Spring MVC、Spring JDBC、Spring Security 等。

JDBC

在使用 JDBC 之前，我们需要先配置 jdbc.properties 文件：

jdbc.driver=com.mysql.cj.jdbc.Driver
jdbc.url=jdbc:mysql://localhost:3306/bookstore
jdbc.username=root
jdbc.password=password

使用配置类：

@Configuration
@ComponentScan(basePackages = "com.example")
public class AppConfig {
    @Bean
    public DataSource dataSource() {
        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver"); // 驱动类
        dataSource.setUrl("jdbc:mysql://localhost:3306/bookstore"); // 数据库连接 URL
        dataSource.setUsername("root"); // 用户名
        dataSource.setPassword("password"); // 密码
        return dataSource;
    }

    @Bean
    public JdbcTemplate jdbcTemplate() {
        JdbcTemplate jdbcTemplate = new JdbcTemplate();
        jdbcTemplate.setDataSource(dataSource());
        return jdbcTemplate;
    }
}

这里，我们创建了 DataSource 来连接数据库，创建了 JdbcTemplate 来操作数据库。

查看使用 XML 配置

配置 XML 文件：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns:context="http://www.springframework.org/schema/context"
        xmlns:jdbc="http://www.springframework.org/schema/jdbc"
        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
        http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd">
  
      <context:property-placeholder location="classpath:jdbc.properties"/>

      <bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
          <property name="driverClassName" value="${jdbc.driver}"/>
          <property name="url" value="${jdbc.url}"/>
          <property name="username" value="${jdbc.username}"/>
          <property name="password" value="${jdbc.password}"/>
      </bean>

      <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
          <property name="dataSource" ref="dataSource"/>
      </bean>

</beans>

然后，我们进行具体的增删改查操作：

@Repository
public class BookDaoImpl implements BookDao {
    @Autowired
    private JdbcTemplate jdbcTemplate;

    public void addBook(Book book) {
        String sql = "INSERT INTO book VALUES(?, ?)";
        jdbcTemplate.update(sql, book.getTitle(), book.getAuthor());
    }

    public void updateBookAuthorByTitle(String title, String author) {
        String sql = "UPDATE book SET author = ? WHERE title = ?";
        jdbcTemplate.update(sql, author, title);
    }

    public Book getBookByTitle(String title) {
        String sql = "SELECT * FROM book WHERE title = ?";
        return jdbcTemplate.queryForObject(sql, new BeanPropertyRowMapper<>(Book.class), title);
    }

    public List<Book> getAllBooks() {
        String sql = "SELECT * FROM book";
        return jdbcTemplate.query(sql, new BeanPropertyRowMapper<>(Book.class));
    }

    public int countBooks() {
        String sql = "SELECT COUNT(*) FROM book";
        return jdbcTemplate.queryForObject(sql, Integer.class);
    }
}

JDBC 源码解读

我们其实只关心 JdbcTemplate 的 update 和 query 方法。

update

点击查看 update 源码解读

update 方法封装在 JdbcOperations 中：

@Override
public int update(String sql, @Nullable PreparedStatementSetter pss) throws DataAccessException {
    return update(new SimplePreparedStatementCreator(sql), pss);
}

@Override
public int update(String sql, @Nullable Object... args) throws DataAccessException {
    return update(sql, newArgPreparedStatementSetter(args));
}

跟踪 update 方法：

protected int update(final PreparedStatementCreator psc, @Nullable final PreparedStatementSetter pss)
    throws DataAccessException {

    logger.debug("Executing prepared SQL update");

    return updateCount(execute(psc, ps -> {
        try {
            if (pss != null) {
                // 设置参数
                pss.setValues(ps);
            }
            // 执行更新
            int rows = ps.executeUpdate();
            if (logger.isTraceEnabled()) {
                logger.trace("SQL update affected " + rows + " rows");
            }
            return rows;
        }
        finally {
            if (pss instanceof ParameterDisposer parameterDisposer) {
                parameterDisposer.cleanupParameters();
            }
        }
    }, true));
}

这里，updateCount 方法会返回更新的行数。因此，我们来看 execute 方法：

@Nullable
  private <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action, boolean closeResources)
      throws DataAccessException {

    Assert.notNull(psc, "PreparedStatementCreator must not be null");
    Assert.notNull(action, "Callback object must not be null");
    if (logger.isDebugEnabled()) {
        String sql = getSql(psc);
        logger.debug("Executing prepared SQL statement" + (sql != null ? " [" + sql + "]" : ""));
    }

    // 获取连接
    Connection con = DataSourceUtils.getConnection(obtainDataSource());
    PreparedStatement ps = null;
    try {
        ps = psc.createPreparedStatement(con);
        applyStatementSettings(ps);
        // 执行回调
        T result = action.doInPreparedStatement(ps);
        handleWarnings(ps);
        return result;
    }
    catch (SQLException ex) {
        if (psc instanceof ParameterDisposer parameterDisposer) {
          parameterDisposer.cleanupParameters();
        }
        if (ps != null) {
          handleWarnings(ps, ex);
        }
        String sql = getSql(psc);
        psc = null;
        JdbcUtils.closeStatement(ps);
        ps = null;
        DataSourceUtils.releaseConnection(con, getDataSource());
        con = null;
        throw translateException("PreparedStatementCallback", sql, ex);
    }
    finally {
        if (closeResources) {
            if (psc instanceof ParameterDisposer parameterDisposer) {
                parameterDisposer.cleanupParameters();
            }
            JdbcUtils.closeStatement(ps);
            DataSourceUtils.releaseConnection(con, getDataSource());
        }
    }
}

再看 setValues 方法：

@Override
public void setValues(PreparedStatement ps) throws SQLException {
    if (this.args != null) {
        for (int i = 0; i < this.args.length; i++) {
            Object arg = this.args[i];
            doSetValue(ps, i + 1, arg);
        }
    }
}

跟踪 doSetValue 方法：

protected void doSetValue(PreparedStatement ps, int parameterPosition, @Nullable Object argValue)
    throws SQLException {

    if (argValue instanceof SqlParameterValue paramValue) {
        StatementCreatorUtils.setParameterValue(ps, parameterPosition, paramValue, paramValue.getValue());
    }
    else {
        StatementCreatorUtils.setParameterValue(ps, parameterPosition, SqlTypeValue.TYPE_UNKNOWN, argValue);
    }
}

我们就不继续跟踪下去了。最后实际上就是根据数据类型，设置参数。

executeUpdate 方法实际上是 java.sql 包中的方法，跟框架无关。

Transaction

为什么需要事务

在实际开发中，我们经常会遇到这样的情况：一个业务操作需要执行多个 SQL 语句，如果其中一个 SQL 语句执行失败，那么其他 SQL 语句也应该回滚。例如，购书时，需要先扣除用户的余额，然后再减少书的库存。如果书没了，但是余额却扣除了，那么就会出现问题；同理，如果余额不够，书却减少了，也会出现问题。

在最原始的编程式事务管理中，我们需要先关闭自动提交，然后手动提交或回滚事务：

@Service
public class BookServiceImpl implements BookService {
    @Autowired
    private BookDao bookDao;

    public void buyBook(String title, String username) {
        Book book = bookDao.getBookByTitle(title);
        int price = book.getPrice();

        try {
            DataSourceUtils.getConnection(dataSource).setAutoCommit(false);

            bookDao.updateBookStockByTitle(title, -1);
            bookDao.updateUserBalanceByUsername(username, -price);

            DataSourceUtils.getConnection(dataSource).commit();
        } catch (Exception e) {
            DataSourceUtils.getConnection(dataSource).rollback();
        } finally {
            DataSourceUtils.getConnection(dataSource).setAutoCommit(true);
        }
    }
}

可以看到，这一方法实现了事务管理的 ACID 特性，但是这样做有几个问题：

代码冗余

每个方法都需要写高度雷同事务代码，导致了代码冗余。
耦合度高

业务代码和事务代码耦合在一起，导致了耦合度过高，不利于集中维护。

因此，我们需要事务管理来解决这些问题。

事务管理

Spring 提供了两种事务管理方式：编程式事务管理和声明式事务管理。声明式事务管理可以有效解决上面提到的问题。

使用注解配置：

@Configuration
@EnableTransactionManagement
@ComponentScan(basePackages = "com.example")
public class AppConfig {
    @Bean
    public DataSource dataSource() {
        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        dataSource.setUrl("jdbc:mysql://localhost:3306/bookstore");
        dataSource.setUsername("root");
        dataSource.setPassword("password");
        return dataSource;
    }

    @Bean
    public JdbcTemplate jdbcTemplate() {
        JdbcTemplate jdbcTemplate = new JdbcTemplate();
        jdbcTemplate.setDataSource(dataSource());
        return jdbcTemplate;
    }

    @Bean
    public DataSourceTransactionManager transactionManager() {
        DataSourceTransactionManager transactionManager = new DataSourceTransactionManager();
        transactionManager.setDataSource(dataSource());
        return transactionManager;
    }
}

查看使用 XML 配置

在配置文件中开启事务管理：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns:context="http://www.springframework.org/schema/context"
        xmlns:tx="http://www.springframework.org/schema/tx"
        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
        http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
  
      <context:property-placeholder location="classpath:jdbc.properties"/>

      <bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
          <property name="driverClassName" value="${jdbc.driver}"/>
          <property name="url" value="${jdbc.url}"/>
          <property name="username" value="${jdbc.username}"/>
          <property name="password" value="${jdbc.password}"/>
      </bean>

      <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
          <property name="dataSource" ref="dataSource"/>
      </bean>

      <tx:annotation-driven transaction-manager="transactionManager"/>

      <bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
          <property name="dataSource" ref="dataSource"/>
      </bean>

</beans>

然后，我们可以使用 @Transactional 注解来声明事务：

@Service
public class BookServiceImpl implements BookService {
    @Autowired
    private BookDao bookDao;

    @Transactional
    public void buyBook(String title, String username) {
        Book book = bookDao.getBookByTitle(title);
        int price = book.getPrice();

        bookDao.updateBookStockByTitle(title, -1);
        bookDao.updateUserBalanceByUsername(username, -price);
    }
}

这时，我们就不需要手动提交或回滚事务了，Spring 会自动为我们处理。

@Transactional 注解有以下几个属性：

propagation

事务传播行为，确定了有包含关系的两个事务如何执行，默认值为 Propagation.REQUIRED。它有以下几种取值：
- Propagation.REQUIRED：如果当前没有事务，就新建一个事务；如果当前有事务，就加入到当前事务中
- Propagation.SUPPORTS：如果当前有事务，就加入到当前事务中；如果当前没有事务，就以非事务方式执行
- Propagation.MANDATORY：如果当前有事务，就加入到当前事务中；如果当前没有事务，就抛出异常
- Propagation.REQUIRES_NEW：新建一个事务，如果当前有事务，就将当前事务挂起
- Propagation.NOT_SUPPORTED：以非事务方式执行，如果当前有事务，就将当前事务挂起
- Propagation.NEVER：以非事务方式执行，如果当前有事务，就抛出异常
- Propagation.NESTED：如果当前没有事务，就新建一个事务；如果当前有事务，就在当前事务中嵌套一个事务
isolation

事务隔离级别，确定了不同事务之间如何互相影响，默认值为 Isolation.DEFAULT。它有以下几种取值：
- Isolation.DEFAULT：使用数据库默认的隔离级别
- Isolation.READ_UNCOMMITTED：读未提交，即一个事务可以读取另一个事务已经修改但还未提交的数据
- Isolation.READ_COMMITTED：读已提交，即一个事务只能读取另一个事务已经提交的数据
- Isolation.REPEATABLE_READ：可重复读，即一个事务在多次读取同一数据时，读到的数据是一样的，在此期间，其他事务对该数据的修改是不可见的
- Isolation.SERIALIZABLE：串行化，即一个事务在执行时，另一个事务不能对其进行修改
timeout

事务超时时间，默认值为 -1，单位为秒。如果程序卡住，超过了指定时间，事务会自动回滚
readOnly

是否只读事务，默认值为 false。如果设置为 true，则只能进行查询操作，不能进行增删改操作
rollbackFor、rollbackForClassName、noRollbackFor、noRollbackForClassName

设置哪些异常会回滚事务，哪些异常不会回滚事务。参数为 Class 类型或者 String 类型

基于 XML 配置

查看使用 XML 配置

声明式事务管理同样能够基于 XML 配置：

<tx:advice id="txAdvice" transaction-manager="transactionManager">
    <tx:attributes>
        <tx:method name="buy*" propagation="REQUIRED" isolation="DEFAULT" timeout="-1" read-only="false"/>
        <tx:method name="get*" read-only="true"/>
        <tx:method name="*" propagation="REQUIRED"/>
    </tx:attributes>
</tx:advice>

<aop:config>
    <aop:pointcut id="txPointcut" expression="execution(* com.example.service.*.*(..))"/>
    <aop:advisor advice-ref="txAdvice" pointcut-ref="txPointcut"/>
</aop:config>

这种方式通过 AOP 切面来拦截方法，并应用事务管理。tx:advice 定义了事务的属性，aop:config 定义了切入点和切面。

Transaction 源码解读

Spring 的事务管理是通过 AOP 实现的。当我们在方法上添加 @Transactional 注解时，Spring 会为该方法创建一个代理对象，在方法执行前后进行事务管理。

事务管理器

Spring 提供了多种事务管理器，用于不同的数据访问技术：

DataSourceTransactionManager：用于 JDBC 和 MyBatis
HibernateTransactionManager：用于 Hibernate
JpaTransactionManager：用于 JPA
JtaTransactionManager：用于分布式事务

这些事务管理器都实现了 PlatformTransactionManager 接口：

public interface PlatformTransactionManager extends TransactionManager {
    TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException;
    void commit(TransactionStatus status) throws TransactionException;
    void rollback(TransactionStatus status) throws TransactionException;
}

事务拦截器

当方法被调用时，TransactionInterceptor 会拦截该方法，并在方法执行前后进行事务管理。核心方法是 invoke：

@Override
@Nullable
public Object invoke(MethodInvocation invocation) throws Throwable {
    Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
    
    // 在事务中执行方法
    return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
}

跟踪 invokeWithinTransaction 方法：

@Nullable
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);
    final TransactionManager tm = determineTransactionManager(txAttr);

    PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
    final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);

    if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
        // 创建事务
        TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);

        Object retVal;
        try {
            // 执行目标方法
            retVal = invocation.proceedWithInvocation();
        }
        catch (Throwable ex) {
            // 回滚事务
            completeTransactionAfterThrowing(txInfo, ex);
            throw ex;
        }
        finally {
            cleanupTransactionInfo(txInfo);
        }

        // 提交事务
        commitTransactionAfterReturning(txInfo);
        return retVal;
    }
    // ...其他代码...
}

关键步骤：

获取事务属性：从 @Transactional 注解中获取事务属性
创建事务：根据事务传播行为创建或加入事务
执行目标方法：通过反射调用原始方法
提交或回滚：根据执行结果提交或回滚事务

事务创建

createTransactionIfNecessary 方法负责创建事务：

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

    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) {
            // 获取事务
            status = tm.getTransaction(txAttr);
        }
    }
    return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

getTransaction 方法会根据事务传播行为决定是创建新事务还是加入现有事务：

@Override
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
        throws TransactionException {

    TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());

    Object transaction = doGetTransaction();
    
    // 检查是否存在事务
    if (isExistingTransaction(transaction)) {
        // 处理现有事务
        return handleExistingTransaction(def, transaction, debugEnabled);
    }

    // 检查传播行为
    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;
        }
    }
    // ...其他代码...
}

事务提交与回滚

提交事务的 commitTransactionAfterReturning 方法：

protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) {
    if (txInfo != null && txInfo.getTransactionStatus() != null) {
        txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
    }
}

回滚事务的 completeTransactionAfterThrowing 方法：

protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {
    if (txInfo != null && txInfo.getTransactionStatus() != null) {
        if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
            try {
                txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
            }
            catch (TransactionSystemException ex2) {
                ex2.initApplicationException(ex);
                throw ex2;
            }
            catch (RuntimeException | Error ex2) {
                throw ex2;
            }
        }
        else {
            // 不需要回滚的异常，仍然提交事务
            try {
                txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
            }
            catch (TransactionSystemException ex2) {
                ex2.initApplicationException(ex);
                throw ex2;
            }
            catch (RuntimeException | Error ex2) {
                throw ex2;
            }
        }
    }
}

这里会根据 rollbackFor 属性判断是否需要回滚。默认情况下，只有运行时异常（RuntimeException）和错误（Error）会导致回滚，检查型异常（CheckedException）不会回滚。

深入 Spring 源码 | (3)

JDBC 和事务

JDBC

JDBC 源码解读

update

Transaction

为什么需要事务

事务管理

基于 XML 配置

Transaction 源码解读

事务管理器

事务拦截器

事务创建

事务提交与回滚

Series 深入 Spring 源码

Suggested Posts

Comments

深入 Spring 源码 | (3) JDBC 和事务

JDBC

JDBC 源码解读

update

Transaction

为什么需要事务

事务管理

基于 XML 配置

Transaction 源码解读

事务管理器

事务拦截器

事务创建

事务提交与回滚

Series 深入 Spring 源码

Suggested Posts

Comments

Share This Post

深入 Spring 源码 | (3)

JDBC 和事务