【Spring从成神到升仙系列四】从源码分析 Spring 事务的来龙去脉

2023-09-08 08:34

短信预约 -IT技能 免费直播动态提醒

👏作者简介：大家好，我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，阿里云专家博主
📕系列专栏：Java设计模式、数据结构和算法、Kafka从入门到成神、Kafka从成神到升仙、Spring从成神到升仙系列
🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
🍂博主正在努力完成2023计划中：以梦为马，扬帆起航，2023追梦人
📝联系方式：hls1793929520，加我进群，大家一起学习，一起进步👀

在这里插入图片描述

文章目录

Spring 事务源码解析

Spring 事务源码解析

一、引言

对于Java开发者而言，关于 Spring ，我们一般当做黑盒来进行使用，不需要去打开这个黑盒。

但随着目前程序员行业的发展，我们有必要打开这个黑盒，去探索其中的奥妙。

本期 Spring 源码解析系列文章，将带你领略 Spring 源码的奥秘

本期源码文章吸收了之前 Kafka 源码文章的错误，将不再一行一行的带大家分析源码，我们将一些不重要的部分当做黑盒处理，以便我们更快、更有效的阅读源码。

废话不多说，发车！

本篇目录如下：
请添加图片描述

本文流程图可关注公众号：爱敲代码的小黄，回复：事务获取
贴心的小黄为大家准备的文件格式为 POS文件，方便大家直接导入 ProcessOn 修改使用

二、事务的本质

数据库事务(Database Transaction) ，是指作为单个逻辑工作单元执行的一系列操作，要么完全地执行，要么完全地不执行。

事务处理可以确保除非事务性单元内的所有操作都成功完成，否则不会永久更新面向数据的资源。通过将一组相关操作组合为一个要么全部成功要么全部失败的单元，可以简化错误恢复并使应用程序更加可靠。

一个逻辑工作单元要成为事务，必须满足所谓的 ACID（原子性、一致性、隔离性和持久性）属性。事务是数据库运行中的逻辑工作单位，由DBMS中的事务管理子系统负责事务的处理。

1、JDBC的事务

我们来看一下在 JDBC 中对事务的操作处理：

public class JDBCTransactionExample {    public static void main(String[] args) {        Connection conn = null;        PreparedStatement pstmt1 = null;        PreparedStatement pstmt2 = null;        try {            // 加载驱动            Class.forName("com.mysql.jdbc.Driver");            // 获取连接            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "password");            // 关闭自动提交，开启事务            conn.setAutoCommit(false);            // 创建SQL语句            String sql1 = "UPDATE account SET balance = balance - ? WHERE id = ?";            String sql2 = "UPDATE account SET balance = balance + ? WHERE id = ?";            // 创建PreparedStatement对象            pstmt1 = conn.prepareStatement(sql1);            pstmt2 = conn.prepareStatement(sql2);            // 设置参数            pstmt1.setDouble(1, 1000);            pstmt1.setInt(2, 1);            pstmt2.setDouble(1, 1000);            pstmt2.setInt(2, 2);            // 执行更新操作            int count1 = pstmt1.executeUpdate();            int count2 = pstmt2.executeUpdate();            if (count1 > 0 && count2 > 0) {                System.out.println("转账成功");                // 提交事务                conn.commit();            } else {                System.out.println("转账失败");                // 回滚事务                conn.rollback();            }        } catch (ClassNotFoundException e) {            e.printStackTrace();        } catch (SQLException e) {            try {                if (conn != null) {                    // 回滚事务                    conn.rollback();                }            } catch (SQLException e1) {                e1.printStackTrace();            }            e.printStackTrace();        } finally {            try {                if (pstmt1 != null) {                    pstmt1.close();                }                if (pstmt2 != null) {                    pstmt2.close();                }                if (conn != null) {                    conn.close();                }            } catch (SQLException e) {                e.printStackTrace();            }        }    }}

上面的代码，我相信大部分的人都应该接触过，这里也就不多说了

主要我们看几个重点步骤：

获取连接：Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "password")
关闭自动提交，开启事务：conn.setAutoCommit(false)
提交事务：conn.commit()
回滚事务：conn.rollback()

2、Spring的事务

我们在日常生产项目中，项目由 Controller、Serivce、Dao 三层进行构建。

我们从上图中可以了解到：

对于 addUser 方法实际对于数据调用来说，分别调用了 insertUser()、insertLog 方法，对数据库的操作为两次

我们要保证 addUser 方法是符合事务定义的。

2.1 xml配置

<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:p="http://www.springframework.org/schema/p"xmlns:context="http://www.springframework.org/schema/context"xmlns:aop="http://www.springframework.org/schema/aop"xmlns:tx="http://www.springframework.org/schema/tx"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.3.xsdhttp://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.3.xsdhttp://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd"><context:component-scan base-package="com.dpb.*">context:component-scan><bean class="org.springframework.jdbc.datasource.DriverManagerDataSource" id="dataSource"><property name="url" value="jdbc:oracle:thin:@localhost:1521:orcl"/><property name="driverClassName" value="oracle.jdbc.driver.OracleDriver"/><property name="username" value="pms"/><property name="password" value="pms"/>bean><bean class="org.springframework.jdbc.core.JdbcTemplate" ><constructor-arg name="dataSource" ref="dataSource"/>bean> <bean class="org.springframework.jdbc.datasource.DataSourceTransactionManager" id="transactionManager"> <property name="dataSource" ref="dataSource"/> bean> <tx:advice id="advice" transaction-manager="transactionManager"> <tx:attributes> <tx:method name="fun*" propagation="REQUIRED"/> tx:attributes> tx:advice>  <aop:config> <aop:pointcut expression="execution(* *..service.*.*(..))" id="tx"/>  <aop:advisor advice-ref="advice" pointcut-ref="tx"/> aop:config>beans>

2.2 注解配置

首先必须要添加 @EnableTransactionManagement 注解，保证事务注解生效

@EnableTransactionManagementpublic class AnnotationMain {    public static void main(String[] args) {    }}

其次，在方法上添加 @Transactional 代表注解生效

@Transactionalpublic int insertUser(User user) {    userDao.insertUser();    userDao.insertLog();    return 1;}

上面的操作涉及两个重点：

事务的传播属性
事务的隔离级别

三、Spring事务源码剖析

本次剖析源码我们会尽量挑重点来讲，因为事务源码本身就是依靠 AOP 实现的，我们之前已经很详细的讲过 IOC 和 AOP 的源码实现了，这次带大家一起过一遍事务即可。

因为从博主本身而言，我感觉 Spring 事务其实没有那么的重要，面试也不常考，所以不会花大量的时间去剖析细节源码。

1、TransactionManager

首先我们看一下这个接口的一些组成配置：

****

这里我们重点看 PlatformTransactionManager 的实现，其实现一共三个方法：

获取事务：TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
提交事务：void commit(TransactionStatus status)
回滚事务：void rollback(TransactionStatus status)

我们分别看一下其如何实现的

1.1 获取事务

我们想一下，在获取事务这一阶段，我们会做什么功能呢？

参考上述我们 JDBC 的步骤，这个阶段应该会 创建连接并且开启事务

public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition){       // PROPAGATION_REQUIRED，PROPAGATION_REQUIRES_NEW，PROPAGATION_NESTED都需要新建事务   if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||         def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||         def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {      //没有当前事务的话，REQUIRED，REQUIRES_NEW，NESTED挂起的是空事务，然后创建一个新事务      SuspendedResourcesHolder suspendedResources = suspend(null);      try {         // 看这里重点：开始事务         return startTransaction(def, transaction, debugEnabled, suspendedResources);      }      catch (RuntimeException | Error ex) {         // 恢复挂起的事务         resume(null, suspendedResources);         throw ex;      }   }}private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction, boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {    // 是否需要新同步 boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER); // 创建新的事务 DefaultTransactionStatus status = newTransactionStatus( definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);// 【重点】开启事务和连接doBegin(transaction, definition);// 新同步事务的设置，针对于当前线程的设置prepareSynchronization(status, definition);return status;}protected void doBegin(Object transaction, TransactionDefinition definition) {    // 判断事务对象没有数据库连接持有器if (!txObject.hasConnectionHolder() ||txObject.getConnectionHolder().isSynchronizedWithTransaction()) {// 【重点】通过数据源获取一个数据库连接对象Connection newCon = obtainDataSource().getConnection();// 把我们的数据库连接包装成一个ConnectionHolder对象 然后设置到我们的txObject对象中去             // 再次进来时，该 txObject 就已经有事务配置了txObject.setConnectionHolder(new ConnectionHolder(newCon), 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);}// 绑定我们的数据源和连接到我们的同步管理器上，把数据源作为key,数据库连接作为value 设置到线程变量中if (txObject.isNewConnectionHolder()) {// 将当前获取到的连接绑定到当前线程TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());}}}

到这里，我们的 获取事务 接口完成了 数据库连接的创建 和 关闭自动提交（开启事务），将 Connection 注册到了缓存（resources）当中，便于获取。

1.2 提交事务

public final void commit(TransactionStatus status) throws TransactionException {DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;// 如果在事务链中已经被标记回滚，那么不会尝试提交事务，直接回滚if (defStatus.isLocalRollbackOnly()) {// 不可预期的回滚processRollback(defStatus, false);return;}// 设置了全局回滚if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {// 可预期的回滚，可能会报异常processRollback(defStatus, true);return;}// 【重点】处理事务提交processCommit(defStatus);}// 处理提交，先处理保存点，然后处理新事务，如果不是新事务不会真正提交，要等外层是新事务的才提交，// 最后根据条件执行数据清除,线程的私有资源解绑，重置连接自动提交，隔离级别，是否只读，释放连接，恢复挂起事务等private void processCommit(DefaultTransactionStatus status) throws TransactionException {;     // 如果是独立的事务则直接提交  doCommit(status);                     //根据条件，完成后数据清除,和线程的私有资源解绑，重置连接自动提交，隔离级别，是否只读，释放连接，恢复挂起事务等  cleanupAfterCompletion(status);}

这里比较重要的有两个步骤：

doCommit：提交事务（直接使用 JDBC 提交即可）

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

cleanupAfterCompletion：数据清除，与线程中的私有资源解绑，方便释放

// 线程同步状态清除TransactionSynchronizationManager.clear();// 清除同步状态【这些都是线程的缓存，使用ThreadLocal的】public static void clear() {    synchronizations.remove();    currentTransactionName.remove();    currentTransactionReadOnly.remove();    currentTransactionIsolationLevel.remove();    actualTransactionActive.remove();}// 如果是新事务的话，进行数据清除，线程的私有资源解绑，重置连接自动提交，隔离级别，是否只读，释放连接等doCleanupAfterCompletion(status.getTransaction());// 此方法做清除连接相关操作，比如重置自动提交啊，只读属性啊，解绑数据源啊，释放连接啊，清除链接持有器属性protected void doCleanupAfterCompletion(Object transaction) {DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;// 将数据库连接从当前线程中解除绑定TransactionSynchronizationManager.unbindResource(obtainDataSource());// 释放连接Connection con = txObject.getConnectionHolder().getConnection();    // 恢复数据库连接的自动提交属性con.setAutoCommit(true);      // 重置数据库连接   DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel(), txObject.isReadOnly());// 如果当前事务是独立的新创建的事务则在事务完成时释放数据库连接DataSourceUtils.releaseConnection(con, this.dataSource);// 连接持有器属性清除txObject.getConnectionHolder().clear();}

这就是我们提交事务的操作了，总之来说，主要就是 调用JDBC的commit提交 和 清除一系列的线程内部数据和配置

1.3 回滚事务

public final void rollback(TransactionStatus status) throws TransactionException {   DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;   processRollback(defStatus, false);}private void processRollback(DefaultTransactionStatus status, boolean unexpected) { // 回滚的擦欧洲哦    doRollback(status);        // 回滚完成后回调    triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK); // 根据事务状态信息，完成后数据清除，和线程的私有资源解绑，重置连接自动提交，隔离级别，是否只读，释放连接，恢复挂起事务等cleanupAfterCompletion(status);}protected void doRollback(DefaultTransactionStatus status) {DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();Connection con = txObject.getConnectionHolder().getConnection();// jdbc的回滚con.rollback();}

回滚事务，简单来说 调用JDBC的rollback 和 清除数据

2、事务AOP的实现

我们来思考一下，利用 TransactionManager 重写一下我们第一步的 JDBC 的功能

@Autowiredprivate UserDao userDao;@Autowiredprivate PlatformTransactionManager txManager;@Autowiredprivate LogService logService;@Transactionalpublic void insertUser(User u) {// 1、创建事务定义DefaultTransactionDefinition definition = new DefaultTransactionDefinition();// 2、根据定义开启事务TransactionStatus status = txManager.getTransaction(definition);try {this.userDao.insert(u);Log log = new Log(System.currentTimeMillis() + "", System.currentTimeMillis() + "-" + u.getUserName());// this.doAddUser(u);this.logService.insertLog(log);// 3、提交事务txManager.commit(status);} catch (Exception e) {// 4、异常了，回滚事务txManager.rollback(status);throw e;}}

大家看到上述代码及思考一下 AOP 的作用和源码，有没有一丝丝想法

比如我现在是面试官，问你一个问题：你怎么去设计 Spring 的事务？

如果一点点想法都没有的话，也不用着急，我们来慢慢的分析

2.1 为什么使用AOP？

我们想，如果我们要实现事务，在每一个方法里面都需要进行以下三个步骤：

获取事务
提交事务
回滚事务

是不是显得我们的代码很臃肿，那么我们能不能把这三个步骤搞成一个通用化的东西

一些代码在方法前执行，一些代码方法后执行

这个时候，你是不是就想到了 AOP（切面编程）了

当然，Spring 中也是如此做的，利用 AOP 来对事务进行了统一的包装实现

2.2 @EnableTransactionManagement

我们知道了使用 AOP 技术实现，那到底是如何实现的呢？

我们从 @EnableTransactionManagement 注解聊起，我们点进该注解：

@Import(TransactionManagementConfigurationSelector.class)public @interface EnableTransactionManagement {

很明显，TransactionManagementConfigurationSelector 类是我们主要关注的内容

public class TransactionManagementConfigurationSelector extends AdviceModeImportSelector<EnableTransactionManagement> {      @Override   protected String[] selectImports(AdviceMode adviceMode) {      switch (adviceMode) {         case PROXY:            return new String[] {AutoProxyRegistrar.class.getName(),                  ProxyTransactionManagementConfiguration.class.getName()};         case ASPECTJ:            return new String[] {determineTransactionAspectClass()};         default:            return null;      }   }}

一共注册了两个：

AutoProxyRegistrar.class：注册AOP处理器

ProxyTransactionManagementConfiguration.class：代理事务配置，注册事务需要用的一些类，而且Role=ROLE_INFRASTRUCTURE都是属于内部级别的

@Configuration(proxyBeanMethods = false)@Role(BeanDefinition.ROLE_INFRASTRUCTURE)public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {   @Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)   @Role(BeanDefinition.ROLE_INFRASTRUCTURE)   public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(         TransactionAttributeSource transactionAttributeSource, TransactionInterceptor transactionInterceptor) {// 【重点】注册了 BeanFactoryTransactionAttributeSourceAdvisor 的 advisor       // 其 advice 为 transactionInterceptor      BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();      advisor.setTransactionAttributeSource(transactionAttributeSource);      advisor.setAdvice(transactionInterceptor);      if (this.enableTx != null) {         advisor.setOrder(this.enableTx.<Integer>getNumber("order"));      }      return advisor;   }   @Bean   @Role(BeanDefinition.ROLE_INFRASTRUCTURE)   public TransactionAttributeSource transactionAttributeSource() {      return new AnnotationTransactionAttributeSource();   }   @Bean   @Role(BeanDefinition.ROLE_INFRASTRUCTURE)   public TransactionInterceptor transactionInterceptor(TransactionAttributeSource transactionAttributeSource) {      TransactionInterceptor interceptor = new TransactionInterceptor();      interceptor.setTransactionAttributeSource(transactionAttributeSource);      if (this.txManager != null) {         interceptor.setTransactionManager(this.txManager);      }      return interceptor;   }}

到这里，看到 BeanFactoryTransactionAttributeSourceAdvisor 以 advisor 结尾的类，AOP 的 DNA 应该动了，其 advice 为 transactionInterceptor

到这里，我们思考一下，利用我们之前学习到的 AOP 的源码，猜测其运行逻辑：

我们在方法上写上 @EnableTransactionManagement 注解，Spring 会注册一个 BeanFactoryTransactionAttributeSourceAdvisor 的类
创建对应的方法 Bean 时，会和 AOP 一样，利用该 Advisor 类生成对应的代理对象
最终调用方法时，会调用代理对象，并通过环绕增强来达到事务的功能

2.3 TransactionInterceptor

我们从上面看到其 advice 正是 TransactionInterceptor，那自然不用多说

从上篇 AOP 得知，代理对象运行时，会拿到所有的 advice 并进行排序，责任链递归运行

所以，我们直接看 TransactionInterceptor 这个类即可

这里先看一下 TransactionInterceptor 类图
在这里插入图片描述

然后看其源码内容：

这里的 invoke 方法怎么执行到的，我就不多介绍了，看过上篇 AOP 的文章应该都懂，不熟悉的小伙伴可以去看一下

public class TransactionInterceptor extends TransactionAspectSupport implements MethodInterceptor {   @Override@Nullablepublic Object invoke(MethodInvocation invocation) throws Throwable {// 获取我们的代理对象的class属性Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);// Adapt to TransactionAspectSupport's invokeWithinTransaction...return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);}}@Nullableprotected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, final InvocationCallback invocation){   // 获取我们的事务属性源对象TransactionAttributeSource tas = getTransactionAttributeSource();// 通过事务属性源对象获取到当前方法的事务属性信息final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);// 获取我们配置的事务管理器对象final TransactionManager tm = determineTransactionManager(txAttr);        if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {        // 【重点】创建TransactionInfo  TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);        try {// 执行被增强方法,调用具体的处理逻辑【我们实际的方法】retVal = invocation.proceedWithInvocation();}catch (Throwable ex) {// 异常回滚completeTransactionAfterThrowing(txInfo, ex);throw ex;}finally {//清除事务信息，恢复线程私有的老的事务信息cleanupTransactionInfo(txInfo);}        //成功后提交，会进行资源储量，连接释放，恢复挂起事务等操作  commitTransactionAfterReturning(txInfo);return retVal;    }}// 创建连接 + 开启事务protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {    // 获取TransactionStatus事务状态信息    status = tm.getTransaction(txAttr);    // 根据指定的属性与status准备一个TransactionInfo，return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);}// 存在异常时回滚事务protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {    // 进行回滚    txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());}// 调用事务管理器的提交方法protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo){    txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());}