• TransactionSynchronizationManager的使用场景以及原理分析

    目录

    一、是什么

    二、如何使用

    三、原理分析

    前段时间，在开发一个功能时，遇到这么一个场景：有一个同步操作A需要在另一个aop的功能B（涉及事务）完成之后，才去调用。简而言之就是A操作需要在一个B事务成功提交之后才去执行，A和B，这时候用到了TransactionSynchronizationManager（这里当然同样可以使用AOP去处理，事务的本质就是aop）,这里我使用到了TransactionSynchronizationManager！

     阅读本文前提需要读者了解spring事物的基本操作和事务的传播机制：

    Spring注解开发（十四）——事务基本原理分析_advicemode_cj_eryue的博客-CSDN博客

    Spring事务的隔离级别以及传播_spring事务隔离级别_cj_eryue的博客-CSDN博客

    一、是什么

    翻译过来就是事务同步管理器，我们可以自定义实现TransactionSynchronization类，来监听Spring的事务操作。可以在事务提交之后，回调TransactionSynchronization类的方法。
     

    二、如何使用

    我们通过一个简单的demo来看下如何使用：

    依赖如下：

    1. <dependency>
    2.             <groupId>org.springframeworkgroupId>
    3.             <artifactId>spring-contextartifactId>
    4.             <version>5.2.9.RELEASEversion>
    5.         dependency>
    6.         <dependency>
    7.             <groupId>commons-logginggroupId>
    8.             <artifactId>commons-loggingartifactId>
    9.             <version>1.2version>
    10.         dependency>
    11.  
    12.         <dependency>
    13.             <groupId>org.springframeworkgroupId>
    14.             <artifactId>spring-aspectsartifactId>
    15.             <version>5.2.9.RELEASEversion>
    16.         dependency>
    17.         <dependency>
    18.             <groupId>org.springframeworkgroupId>
    19.             <artifactId>spring-jdbcartifactId>
    20.             <version>4.2.2.RELEASEversion>
    21.         dependency>
    22.         <dependency>
    23.             <groupId>mysqlgroupId>
    24.             <artifactId>mysql-connector-javaartifactId>
    25.             <version>8.0.28version>
    26.         dependency>
    27.         
    28.         <dependency>
    29.             <groupId>c3p0groupId>
    30.             <artifactId>c3p0artifactId>
    31.             <version>0.9.1.2version>
    32.         dependency>
    33.     dependencies>

    配置类：

    1. package com.cjian.config;
    2.  
    3. import com.mchange.v2.c3p0.ComboPooledDataSource;
    4. import org.springframework.context.annotation.Bean;
    5. import org.springframework.context.annotation.ComponentScan;
    6. import org.springframework.context.annotation.Configuration;
    7. import org.springframework.jdbc.core.JdbcTemplate;
    8. import org.springframework.jdbc.datasource.DataSourceTransactionManager;
    9. import org.springframework.transaction.PlatformTransactionManager;
    10. import org.springframework.transaction.annotation.EnableTransactionManagement;
    11.  
    12. import javax.sql.DataSource;
    13. import java.beans.PropertyVetoException;
    14.  
    15. /**
    16.  * @Author: cjian
    17.  * @Date: 2023/9/25 10:58
    18.  * @Des:
    19.  */
    20. @Configuration
    21. @ComponentScan("com.cjian.tx")
    22. @EnableTransactionManagement
    23. public class TxConfig {
    24.     @Bean
    25.     public DataSource dataSource() throws PropertyVetoException {
    26.         ComboPooledDataSource dataSource = new ComboPooledDataSource();
    27.         dataSource.setUser("root");
    28.         dataSource.setPassword("111111");
    29.         dataSource.setDriverClass("com.mysql.cj.jdbc.Driver");
    30.         dataSource.setJdbcUrl("jdbc:mysql://localhost:3306/test");
    31.         return dataSource;
    32.     }
    33.  
    34.     @Bean
    35.     public JdbcTemplate jdbcTemplate() throws PropertyVetoException {
    36.         return new JdbcTemplate(dataSource());
    37.     }
    38.  
    39.     @Bean
    40.     public PlatformTransactionManager platformTransactionManager() throws PropertyVetoException {
    41.         return new DataSourceTransactionManager(dataSource());
    42.     }
    43. }

    Dao

    1. package com.cjian.tx;
    2.  
    3. import org.springframework.beans.factory.annotation.Autowired;
    4. import org.springframework.jdbc.core.JdbcTemplate;
    5. import org.springframework.stereotype.Repository;
    6. import org.springframework.transaction.annotation.Propagation;
    7. import org.springframework.transaction.annotation.Transactional;
    8.  
    9. import java.util.List;
    10. import java.util.Map;
    11. import java.util.UUID;
    12.  
    13. /**
    14.  * @Author: cjian
    15.  * @Date: 2023/9/25 11:05
    16.  * @Des:
    17.  */
    18. @Repository
    19. public class UserDao {
    20.     @Autowired
    21.     private JdbcTemplate jdbcTemplate;
    22.  
    23.     public void insertUser() {
    24.         String sql = "insert into person(name,age) values(?,?)";
    25.         String name = UUID.randomUUID().toString().substring(0, 5);
    26.         jdbcTemplate.update(sql, name, 30);
    27.     }
    28.  
    29.     // 这里我通过开启一个新的事务，来模拟两个不同事务
    30.     @Transactional(propagation = Propagation.REQUIRES_NEW)
    31.     public void findUser() {
    32.         String sql = "select * from person where age = 30";
    33.         List> maps = jdbcTemplate.queryForList(sql);
    34.         System.out.println(maps.size());
    35.     }
    36.  
    37. }

    Service

    1. package com.cjian.tx;
    2.  
    3. import org.springframework.beans.factory.annotation.Autowired;
    4. import org.springframework.stereotype.Service;
    5. import org.springframework.transaction.annotation.Transactional;
    6. import org.springframework.transaction.support.TransactionSynchronizationAdapter;
    7. import org.springframework.transaction.support.TransactionSynchronizationManager;
    8.  
    9. /**
    10.  * @Author: cjian
    11.  * @Date: 2023/9/25 11:58
    12.  * @Des:
    13.  */
    14. @Service("userService")
    15. public class UserService {
    16.     @Autowired
    17.     private UserDao userDao;
    18.  
    19.     @Transactional
    20.     public void inertUser() {
    21.         userDao.insertUser();
    22.         System.out.println("插入成功！");
    23.         if (TransactionSynchronizationManager.isSynchronizationActive()) {
    24.             TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
    25.                 @Override
    26.                 public void beforeCommit(boolean readOnly) {
    27.                     userDao.findUser();
    28.                 }
    29.  
    30.                 @Override
    31.                 public void afterCommit() {
    32.                     userDao.findUser();
    33.                 }
    34.             });
    35.         }
    36.     }
    37. }

    测试类 

    1.    private static AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(
    2.             TxConfig.class);
    3.  
    4.     public static void main(String[] args) {
    5.         UserService userService = applicationContext.getBean(UserService.class);
    6.         userService.inertUser();
    7.         applicationContext.close();
    8.  
    9.     }

    输入结果为：

    1. 插入成功！
    2. 0
    3. 1

    事务提交前输出0：是因为findUser开启了一个新的事务（如果不加上新的事务，由spring的事务传播机制可知insertUser和findUser处在同一个事务里，同一个事务里的新增对后续的查是可见的）。

    事务提交后输出1：在insertUser的事务提交后才去查询

    三、原理分析

    这里顺带分析下jdbcTemplate的事务原理，以及TransactionSynchronizationManager在源码中的使用。

    先看下TransactionSynchronizationManager 的一些属性：

    1. public abstract class TransactionSynchronizationManager {
    2.  
    3.      //线程上下文中保存着【线程池对象：ConnectionHolder】的Map对象。线程可以通过该属性获取到同一个Connection对象。
    4.     private static final ThreadLocal> resources = new NamedThreadLocal<>("Transactional resources");
    5.  
    6.     //事务同步器，是Spring交由程序员进行扩展的代码，每个线程可以注册N个事务同步器。
    7.     private static final ThreadLocal> synchronizations = new NamedThreadLocal<>("Transaction synchronizations");
    8.     
    9.     // 事务的名称  
    10.     private static final ThreadLocal currentTransactionName = new NamedThreadLocal<>("Current transaction name");
    11.     // 事务是否是只读  
    12.     private static final ThreadLocal currentTransactionReadOnly = new NamedThreadLocal<>("Current transaction read-only status");
    13.     // 事务的隔离级别
    14.     private static final ThreadLocal currentTransactionIsolationLevel = new NamedThreadLocal<>("Current transaction isolation level");
    15.     // 事务是否开启   actual：真实的
    16.     private static final ThreadLocal actualTransactionActive = new NamedThreadLocal<>("Actual transaction active");
    17.  
    18. public static void bindResource(Object key, Object value) throws IllegalStateException {
    19.         Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
    20.         Assert.notNull(value, "Value must not be null");
    21.         Map map = (Map)resources.get();
    22.         if (map == null) {
    23.             map = new HashMap();
    24.             resources.set(map);
    25.         }
    26.  
    27.         //将Connection对象绑定到resources 上。
    28.         Object oldValue = ((Map)map).put(actualKey, value);
    29.         if (oldValue instanceof ResourceHolder && ((ResourceHolder)oldValue).isVoid()) {
    30.             oldValue = null;
    31.         }
    32.  
    33.         if (oldValue != null) {
    34.             throw new IllegalStateException("Already value [" + oldValue + "] for key [" + actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
    35.         } else {
    36.             if (logger.isTraceEnabled()) {
    37.                 logger.trace("Bound value [" + value + "] for key [" + actualKey + "] to thread [" + Thread.currentThread().getName() + "]");
    38.             }
    39.  
    40.         }
    41.     }
    42. }

    resources 用来存储数据库连接，真的吗？来跟下DataSourceTransactionManager链路

    1.  @Bean
    2.  public PlatformTransactionManager platformTransactionManager() throws
    3.  PropertyVetoException {
    4.         return new DataSourceTransactionManager(dataSource());
    5.     }
    1. public class DataSourceTransactionManager extends AbstractPlatformTransactionManager implements ResourceTransactionManager, InitializingBean {
    2. //...
    3. protected void doBegin(Object transaction, TransactionDefinition definition) {
    4.         DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
    5.         Connection con = null;
    6.  
    7.         try {
    8.             if (!txObject.hasConnectionHolder() || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
    9.                 Connection newCon = this.obtainDataSource().getConnection();
    10.                 if (this.logger.isDebugEnabled()) {
    11.                     this.logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
    12.                 }
    13.  
    14.                 txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
    15.             }
    16.  
    17.             txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
    18.             con = txObject.getConnectionHolder().getConnection();
    19.             Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
    20.             txObject.setPreviousIsolationLevel(previousIsolationLevel);
    21.             if (con.getAutoCommit()) {
    22.                 txObject.setMustRestoreAutoCommit(true);
    23.                 if (this.logger.isDebugEnabled()) {
    24.                     this.logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
    25.                 }
    26.                 //关闭自动提交，转交由事务管理器控制，也就是交给spring的aop来控制
    27.                 con.setAutoCommit(false);
    28.             }
    29.  
    30.             this.prepareTransactionalConnection(con, definition);
    31.             txObject.getConnectionHolder().setTransactionActive(true);
    32.             int timeout = this.determineTimeout(definition);
    33.             if (timeout != -1) {
    34.                 txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
    35.             }
    36.  
    37.             //将Connection对象绑定到事务管理器中的Thread中
    38.             if (txObject.isNewConnectionHolder()) {
    39.                 TransactionSynchronizationManager.bindResource(this.obtainDataSource(), txObject.getConnectionHolder());
    40.             }
    41.  
    42.         } catch (Throwable var7) {
    43.             if (txObject.isNewConnectionHolder()) {
    44.                 DataSourceUtils.releaseConnection(con, this.obtainDataSource());
    45.                 txObject.setConnectionHolder((ConnectionHolder)null, false);
    46.             }
    47.  
    48.             throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", var7);
    49.         }
    50.     }
    51. //...
    52. }
    53.  
    54. public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
    55. //...
    56. public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
    57.         Object transaction = this.doGetTransaction();
    58.         boolean debugEnabled = this.logger.isDebugEnabled();
    59.         if (definition == null) {
    60.             definition = new DefaultTransactionDefinition();
    61.         }
    62.  
    63.         if (this.isExistingTransaction(transaction)) {
    64.             return this.handleExistingTransaction((TransactionDefinition)definition, transaction, debugEnabled);
    65.         } else if (((TransactionDefinition)definition).getTimeout() < -1) {
    66.             throw new InvalidTimeoutException("Invalid transaction timeout", ((TransactionDefinition)definition).getTimeout());
    67.         } else if (((TransactionDefinition)definition).getPropagationBehavior() == 2) {
    68.             throw new IllegalTransactionStateException("No existing transaction found for transaction marked with propagation 'mandatory'");
    69.         } else if (((TransactionDefinition)definition).getPropagationBehavior() != 0 && ((TransactionDefinition)definition).getPropagationBehavior() != 3 && ((TransactionDefinition)definition).getPropagationBehavior() != 6) {
    70.             if (((TransactionDefinition)definition).getIsolationLevel() != -1 && this.logger.isWarnEnabled()) {
    71.                 this.logger.warn("Custom isolation level specified but no actual transaction initiated; isolation level will effectively be ignored: " + definition);
    72.             }
    73.  
    74.             boolean newSynchronization = this.getTransactionSynchronization() == 0;
    75.             return this.prepareTransactionStatus((TransactionDefinition)definition, (Object)null, true, newSynchronization, debugEnabled, (Object)null);
    76.         } else {
    77.             AbstractPlatformTransactionManager.SuspendedResourcesHolder suspendedResources = this.suspend((Object)null);
    78.             if (debugEnabled) {
    79.                 this.logger.debug("Creating new transaction with name [" + ((TransactionDefinition)definition).getName() + "]: " + definition);
    80.             }
    81.  
    82.             try {
    83.                 boolean newSynchronization = this.getTransactionSynchronization() != 2;
    84.                 DefaultTransactionStatus status = this.newTransactionStatus((TransactionDefinition)definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
    85.                 //重点方法
    86.                 this.doBegin(transaction, (TransactionDefinition)definition);
    87.                  //重点方法
    88.                 this.prepareSynchronization(status, (TransactionDefinition)definition);
    89.                 return status;
    90.             } catch (Error | RuntimeException var7) {
    91.                 this.resume((Object)null, suspendedResources);
    92.                 throw var7;
    93.             }
    94.         }
    95.     }
    96.  
    97.  protected abstract void doBegin(Object var1, TransactionDefinition var2) throws TransactionException;
    98.  
    99. protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
    100.         if (status.isNewSynchronization()) {
    101.             TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
    102.             TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(definition.getIsolationLevel() != -1 ? definition.getIsolationLevel() : null);
    103.             TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());
    104.             TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());
    105.             TransactionSynchronizationManager.initSynchronization();
    106.         }
    107.  
    108.     }
    109. //...
    110. }
    111.  
    112. //PlatformTransactionManager是事务的核心接口
    113. public interface PlatformTransactionManager {
    114.     TransactionStatus getTransaction(@Nullable TransactionDefinition var1) throws TransactionException;
    115.  
    116.     void commit(TransactionStatus var1) throws TransactionException;
    117.  
    118.     void rollback(TransactionStatus var1) throws TransactionException;
    119. }
    可以看到，数据连接都会绑定到TransactionSynchronizationManager的resources属性上。

    我们通过调用TransactionSynchronizationManager的如下方法，将自定义的事务同步方法添加到synchronizations

    1. public static void registerSynchronization(TransactionSynchronization synchronization) throws IllegalStateException {
    2.         Assert.notNull(synchronization, "TransactionSynchronization must not be null");
    3.         if (!isSynchronizationActive()) {
    4.             throw new IllegalStateException("Transaction synchronization is not active");
    5.         } else {
    6.             ((Set)synchronizations.get()).add(synchronization);
    7.         }
    8.     }

    至于synchronizations的执行时机，可将断点打在getSynchronizations()方法上，一看便知，是在事务提交前/后，遍历执行synchronizations中对应的事务同步方法。

    类似于bean的后置处理器的涉及思想。

    1. public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
    2. // ...
    3.  
    4. private void processCommit(DefaultTransactionStatus status) throws TransactionException {
    5.         try {
    6.            
    7.             try {
    8.                
    9.                 this.prepareForCommit(status);
    10.                 this.triggerBeforeCommit(status);
    11.                 this.triggerBeforeCompletion(status);
    12.                 //...
    13.                 this.doCommit(status);
    14.                 // ...
    15.  
    16.             try {
    17.                 this.triggerAfterCommit(status);
    18.             } finally {
    19.                 this.triggerAfterCompletion(status, 0);
    20.             }
    21.         } finally {
    22.             this.cleanupAfterCompletion(status);
    23.         }
    24.  
    25.     }
    26. // ...
    27. }

    总结：

    TransactionSynchronizationManager.registerSynchronization()的原理就是通过threadLocal获取到当前线程所持有的的dataResource，注册一个乃至多个事务同步方法，在事务提交前后分别执行对应的事务同步方法。

  • 相关阅读:
    [暑假]Js对象部分的学习
    1-31 正则表达式 String Buffer String Builder
    从0开始用C写贪吃蛇（基于链表）
    python中的if ... is None else ...含义与用法
    mysql字符串函数大全（更新完成）
    【Spring IOC容器加载过程】
    《ElementUI 基础知识》el-tree 之“我的电脑”目录结构效果
    antd4 icon使用svg
    滑动窗口练习（一）— 固定窗口最大值问题
    【ORACLE】Oracle里有“time”数据类型吗？--关于对Oracle数据类型的一点研究
  • 原文地址：https://blog.csdn.net/cj_eryue/article/details/133271991