大家好，欢迎来到IT知识分享网。
编程式事务
目录
- 编程式事务
- 一、概述
- 二、代码展示
- 1、引入依赖
- 2、使用xml来进行配置
- 3、代码编写
- 数据库连接什么时候开始创建
- 什么时候需要使用到编程式事务
- 4、编程式事务改造
-
- 注意
-
- 三、Spring提供的编程式事务
- 3.1、execute方法
- 3.2、executeWithoutResult
- 四、传播行为
- 4.1、事务传播
- 4.2、根据条件来创建事务对象
- doGetTransaction
- startTransaction
- doBegin
- prepareSynchronization
- prepareTransactionInfo
- 4.3、回滚
- 4.4、提交
- 同步情况
- 4.5、存在多个传播行为
- 事务失效
- 传播行为调用传播行为
一、概述
对于Spring提供的事务机制来说，如果使用的是注解式事务，往往不好把控业务逻辑中执行数据库操作的时候。
线程的生命周期阶段中(这里的生命周期认为是一次请求之间)相对于数据库连接生命周期(数据库连接执行数据库操作)来说，数据库连接生命周期是短暂的。
所以如何把控业务逻辑使用到数据库连接的时间，这是这个章节所要来讨论的事情。
我觉得分成两种情况：
- 1、数据库连接操作数据库占主要逻辑；
- 2、业务逻辑占据主要逻辑；
对于这两种情况来说，我们都是可以来实现把控的，这也决定了我们究竟来使用编程式事务还是注解式事务。
二、代码展示
1、引入依赖
<dependencies>
<!-- https://mvnrepository.com/artifact/org.Springframework/Spring-context -->
<dependency>
<groupId>org.Springframework</groupId>
<artifactId>Spring-context</artifactId>
<version>5.3.22</version>
</dependency>
<dependency>
<groupId>org.Springframework</groupId>
<artifactId>Spring-jdbc</artifactId>
<version>5.3.22</version>
</dependency>
<dependency>
<groupId>org.Springframework</groupId>
<artifactId>Spring-test</artifactId>
<version>5.3.22</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.mybatis/mybatis-Spring -->
<dependency>
<groupId>org.mybatis</groupId>
<artifactId>mybatis-Spring</artifactId>
<version>2.0.7</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.mybatis/mybatis -->
<dependency>
<groupId>org.mybatis</groupId>
<artifactId>mybatis</artifactId>
<version>3.5.10</version>
</dependency>
<!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.30</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
</dependencies>
2、使用xml来进行配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.Springframework.org/schema/beans"
xmlns:context="http://www.Springframework.org/schema/context"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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:component-scan base-package="com.guang"/>
<!--事务的定义-->
<bean id="txDefinition" class="org.Springframework.transaction.support.DefaultTransactionDefinition">
<!--事务传播特性-->
<property name="propagationBehaviorName" value="PROPAGATION_REQUIRED"/>
<!--事务隔离级别-->
<property name="isolationLevelName" value="ISOLATION_DEFAULT"/>
<!--事务超时时间-->
<property name="timeout" value="-1"/>
<!--事务是否只读-->
<property name="readOnly" value="false"/>
</bean>
<!--定义事务管理器-->
<bean id="txManager" class="org.Springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
<!--定义连接池-->
<bean id="dataSource" class="org.Springframework.jdbc.datasource.DriverManagerDataSource">
<property name="driverClassName" value="com.mysql.cj.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://localhost:3306/ie_user"/>
<property name="username" value="root"/>
<property name="password" value="root"/>
</bean>
<bean id="sqlSessionFactoryBean" class="org.mybatis.Spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dataSource"/>
<property name="configLocation" value="mybatis-config.xml"/>
</bean>
<bean id="mapperScannerConfigurer" class="org.mybatis.Spring.mapper.MapperScannerConfigurer">
<property name="basePackage" value="com.guang.Springmybatistransaction.dao"/>
</bean>
<!--开启事务的注解驱动-->
<tx:annotation-driven transaction-manager="txManager"/>
</beans>
3、代码编写
@Service
public class UserServiceImpl implements UserService {
@Autowired
private UserMapper userMapper;
@Override
@Transactional
public void doHandlTransaction() {
List<User> users = userMapper.selectAllUser();
User user = users.get(0);
user.setEmail("tmdemail111");
userMapper.updateByPrimaryKeySelective(user);
try {
// doSomething
TimeUnit.SECONDS.sleep(5);
int i = 1 / 0;
user.setCreatedBy(666111);
userMapper.updateByPrimaryKeySelective(user);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
如果遇到了上面这种代码操作方式，那么使用的是无论是编程式事务还是注解式事务，从效果上来说，是差不多的。
但是也从这里可以看到，数据库连接占用的时间过长。
数据库连接什么时候开始创建
直接来到TransactionInterceptor中，org.Springframework.jdbc.datasource.DataSourceTransactionManager#doBegin中可以看到一行代码：
Connection newCon = this.obtainDataSource().getConnection();
在这里已经将数据库连接创建好了，在后续执行的时候会拿到数据库连接来进行执行。
什么时候需要使用到编程式事务
所以数据库连接创建过早的情况下，如果是高并发情况，那么这种情况将会造成数据库连接是不够的。
也就是说，将上面的代码调换一下顺序：
@Override
@Transactional
public void doHandlTransaction() {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
List<User> users = userMapper.selectAllUser();
User user = users.get(0);
user.setEmail("tmdemail111");
userMapper.updateByPrimaryKeySelective(user);
user.setCreatedBy(666111);
userMapper.updateByPrimaryKeySelective(user);
}
那么这种情况才是最为致命的！！！
因为没有利用到数据库连接的高效性质，这里相当于是拿到了数据库连接，但是却没有利用其来做事情，而占用了数据库连接池中的数据库连接耗时过长的情况下，最终将会导致数据库连接池消耗光的情况，而其余的线程需要等待数据库连接池中的数据库连接归还之后才能再次进行利用。
4、编程式事务改造
配置类：
@ComponentScan(basePackages = "com.guang.Springmybatistransaction")
@Configuration
// 开启Spring事务
@EnableTransactionManagement
public class AppConfig {
/**
* 事务定义
* 这里都使用默认的，而且是新创建出来的！当然也可以使用已经存在的
* @return
*/
@Bean
public TransactionDefinition transactionDefinition(){
return new DefaultTransactionDefinition();
}
@Bean
public DataSource dataSource(){
DriverManagerDataSource driverManagerDataSource = new DriverManagerDataSource();
driverManagerDataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
driverManagerDataSource.setUrl("jdbc:mysql://localhost:3306/ie_user");
driverManagerDataSource.setUsername("root");
driverManagerDataSource.setPassword("root");
return driverManagerDataSource;
}
@Bean
public PlatformTransactionManager transactionManager(){
PlatformTransactionManager platformTransactionManager = new DataSourceTransactionManager(dataSource());
return platformTransactionManager;
}
@Bean
public SqlSessionFactoryBean sqlSessionFactoryBean(){
SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
sqlSessionFactoryBean.setDataSource(dataSource());
ClassPathResource classPathResource = new ClassPathResource("mybatis-config.xml");
sqlSessionFactoryBean.setConfigLocation(classPathResource);
return sqlSessionFactoryBean;
}
@Bean
public MapperScannerConfigurer mapperScannerConfigurer(){
MapperScannerConfigurer mapperScannerConfigurer = new MapperScannerConfigurer();
mapperScannerConfigurer.setBasePackage("com.guang.Springmybatistransaction.dao");
return mapperScannerConfigurer;
}
}
注意
我在这里使用的时候，爆出了一个错误：
信息: Cannot enhance @Configuration bean definition 'appConfig' since its singleton instance has been created too early. The typical cause is a non-static @Bean method with a BeanDefinitionRegistryPostProcessor return type: Consider declaring such methods as 'static'.
但是又突然想起来了MapperScannerConfigurer是实现了BeanDefinitionRegistryPostProcessor的，这个实例化应该要早于AppConfig。可能就是因为早了一步，就会导致各种各样的问题。
但是官方提示说添加一个static关键字来解决这个问题。
所以先添加一下
@Bean
public static MapperScannerConfigurer mapperScannerConfigurer(){
MapperScannerConfigurer mapperScannerConfigurer = new MapperScannerConfigurer();
mapperScannerConfigurer.setBasePackage("com.guang.springmybatistransaction.dao");
return mapperScannerConfigurer;
}
直接添加static关键字来解决这个问题或者是直接利用@MapperScan来进行添加扫描即可。
对应代码：
@Service
public class UserServiceImpl implements UserService {
@Autowired
private UserMapper userMapper;
@Autowired
private TransactionDefinition transactionDefinition;
@Autowired
private PlatformTransactionManager platformTransactionManager;
@Override
public void doHandlTransaction() {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
TransactionStatus transaction = platformTransactionManager.getTransaction(transactionDefinition);
try {
List<User> users = userMapper.selectAllUser();
User user = users.get(0);
user.setEmail("hushidandan99");
userMapper.updateByPrimaryKeySelective(user);
user.setCreatedBy(777);
userMapper.updateByPrimaryKeySelective(user);
int i = 1 / 0;
platformTransactionManager.commit(transaction);
} catch (Exception e) {
platformTransactionManager.rollback(transaction);
System.out.println("----------------->>>>出现了异常，进行回滚<<<<-----------------");
}finally {
System.out.println("任务执行结束");
}
}
}
但是这种情况Spring肯定也考虑到了，所以我们来确定一下最终的使用方式
三、Spring提供的编程式事务
3.1、execute方法
TransactionTemplate是spring提供出来的能够操作spring中的事务的操作。
使用TransactionTemplate需要配置一个PlatformTransactionManager，但是不需要配置事务定义。因为在TransactionTemplate中，spring已经帮我们配置好了。
直接看下最为核心的方法：execute
public <T> T execute(TransactionCallback<T> action) throws TransactionException {
Assert.state(this.transactionManager != null, "No PlatformTransactionManager set");
if (this.transactionManager instanceof CallbackPreferringPlatformTransactionManager) {
return ((CallbackPreferringPlatformTransactionManager) this.transactionManager).execute(this, action);
}
else {
// 获取得到事务状态信息，然后执行对应的方法
TransactionStatus status = this.transactionManager.getTransaction(this);
T result;
try {
result = action.doInTransaction(status);
}
// 执行失败逻辑
catch (RuntimeException | Error ex) {
// Transactional code threw application exception -> rollback
rollbackOnException(status, ex);
throw ex;
}
catch (Throwable ex) {
// Transactional code threw unexpected exception -> rollback
rollbackOnException(status, ex);
throw new UndeclaredThrowableException(ex, "TransactionCallback threw undeclared checked exception");
}
// 执行成功逻辑
this.transactionManager.commit(status);
return result;
}
}
事务定义已经在下面这行代码中定义好了，可以直接使用lambda表达式来进行操作：
TransactionStatus status = this.transactionManager.getTransaction(this);
事务回滚和提交逻辑也帮我们做好了，我们只需要配置一下TransactionTemplate即可，在逻辑中只需要写对应的逻辑代码即可。
@Service
public class UserServiceImpl implements UserService {
@Autowired
private UserMapper userMapper;
@Autowired
private TransactionTemplate transactionTemplate;
@Override
public void doHandlTransaction() {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
transactionTemplate.execute(status -> {
List<User> users = userMapper.selectAllUser();
User user = users.get(0);
user.setEmail("hushidandan99");
userMapper.updateByPrimaryKeySelective(user);
user.setCreatedBy(777);
userMapper.updateByPrimaryKeySelective(user);
int i = 1/0;
return Boolean.TRUE;
});
}
}
从这里可以看到，使用编程式事务，可以解决掉在方法中书写业务逻辑占用数据库连接时间长的问题。
编程式事务更加细粒度的来进行控制我们的事务，所以也是推荐使用的，配置起来也不麻烦，开箱即用。
3.2、executeWithoutResult
和execute的区别的地方就在于：看返回值是否有对应的结果。如果有返回值，那么就直接使用execute方法；如果没有返回值，就直接使用executeWithoutResult方法即可。那么来看下这里的原理：
default void executeWithoutResult(Consumer<TransactionStatus> action) throws TransactionException {
execute(status -> {
action.accept(status);
return null;
});
}
本质上就是这里使用的一个Consumer接口，只需要执行Consumer接口中的方法即可。这里的返回值对应的值是null。
四、传播行为
回到最开始的位置上，在我们理想情况下，我们在执行SQL的时候，按照：
- 1、获取得到数据库连接；
- 2、设置手动提交(开启事务)；
- 3、执行SQL；
- 4、提交/回滚事务；
从上面的简单例子中就可以看到，就是一次简单的事务开启和提交方式。
但是spring考虑到业务方法可能会调用到业务方法，也就存在一个事务方法调用另外一个事务方法的情况。那么这样子来做的话，就会存在着事务传播情况。
所以的事务传播，在我理解来看，就是方法调用方法，而方法中都存在着事务，spring会来帮助我们来进行处理这种情况。
我从一个线程角度出发：
———–thread———————–>事务A————>事务B（挂起事务A）——–事务C（挂起事务B）——–>……………………………|
———–thread<———————–事务A<————执行事务B（恢复事务A）<——–执行事务C（恢复保存事务B）<—-…………………|
———–>事务执行结束
所谓的挂起和恢复，无非是对象属性中来进行保存的。
如果这里涉及到了多线程的情况，那么首先因为多线程之间的调度是不确定的，其次是无法感知到其他事务，而且在顺序上是无法来进行保证的。所以spring干脆就做了处理，支持在单线程上处理，多线程上的线程不用来进行支持。
无论对于编程式事务还是注解式事务，走的逻辑都是一样的，所以我们重点只需要关注的是一条线路即可。
4.1、事务传播
如果上面的事务A、B、C属性都是一样的，从逻辑上来说，代表的是相同事务；但是从对象上来说，代表的不同的事务，但是因为是相同的，又可以来进行复用而已。但是从本质上来说，都是将挂起的事务进行恢复，然后执行恢复的事务。那么接下来看看spring对传播行为的定义：
Spring定义了7中传播行为：
那么对于其中代码来说，不必每种都看，因为都写在代码中，那么上面做了总结之后，直接看看spring中的代码是怎么实现的即可：
直接来到org.springframework.transaction.interceptor.TransactionInterceptor#invoke方法中来即可：
获取得到运行时候的目标类，然后在事务下调用对应的方法。
那么先总结一下这里的步骤：
- 1、获取得到@Transactional中的属性，封装到TransactionAttributeSource；看注解，如果属性为null，表示的是方法没有事务；
- 2、从容器中获取得到事务管理器；这也就说明了为什么要配置事务管理器；
- 3、判断配置的事务管理器是否是响应式事务管理器还是普通管理器；我们使用的是编程式事务管理器；
- 4、获取得到事务的方法名称；
- 5、根据事务信息来执行业务方法；
- 6、事务提交/回滚；
这样的一番分析下来，整个流程开始变得清晰起来。那么最重要的应该是第5步。
因为只有在第5步骤中涉及到事务传播行为。
4.2、根据条件来创建事务对象
根据配置的事务信息来获取得到事务对象
如果根据属性来创建得到事务的呢？
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
throws TransactionException {
// Use defaults if no transaction definition given.
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
Object transaction = doGetTransaction();
boolean debugEnabled = logger.isDebugEnabled();
if (isExistingTransaction(transaction)) {
// Existing transaction found -> check propagation behavior to find out how to behave.
return handleExistingTransaction(def, transaction, debugEnabled);
}
// Check definition settings for new transaction.
if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
}
// No existing transaction found -> check propagation behavior to find out how to proceed.
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);
if (debugEnabled) {
logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
}
try {
return startTransaction(def, transaction, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error ex) {
resume(null, suspendedResources);
throw ex;
}
}
else {
// Create "empty" transaction: no actual transaction, but potentially synchronization.
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);
}
}
doGetTransaction
protected Object doGetTransaction() {
DataSourceTransactionObject txObject = new DataSourceTransactionObject();
txObject.setSavepointAllowed(isNestedTransactionAllowed());
ConnectionHolder conHolder =
(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
txObject.setConnectionHolder(conHolder, false);
return txObject;
}
创建数据源事务对象，从名称中可以看到应该是有一个数据库连接的的。
这里显示来创建对象，然后从当前线程中，以dataSource为key，来获取得到持有数据库连接的ConnectionHolder，然后将其保存起来。这里分为两种情况：
- 1、当前线程中本来就没有存，所以获取得到的是null；
- 2、当前线程中存了，那么就应该返回存入的值；
在进行返回之后需要来进行判断，是否已经存在了事务以及存在了事务之后，检查传播行为，看看如何来进行执行的？
if (isExistingTransaction(transaction)) {
// Existing transaction found -> check propagation behavior to find out how to behave.
return handleExistingTransaction(def, transaction, debugEnabled);
}
先看下判断：
protected boolean isExistingTransaction(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
}
这个判断很简单，就是判断DataSourceTransactionObject对象中的两个属性：connectionHolder和transactionActive中是否是有值的。
所以这里对应着了上面获取的两个判断。
- 1、当前线程中本来就没有存，所以获取得到的是null；
- 2、当前线程中存了，那么检查一下事务是否是激活的状态；
那么第一次来的时候，默认是都没有的。
下面就开始来判断传播行为，需要根据传播行为来判断如何执行事务。
// No existing transaction found -> check propagation behavior to find out how to proceed.
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);
if (debugEnabled) {
logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
}
try {
return startTransaction(def, transaction, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error ex) {
resume(null, suspendedResources);
throw ex;
}
}
else {
// Create "empty" transaction: no actual transaction, but potentially synchronization.
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);
}
一般来说，对应事务传播行为都是Propagation.REQUIRED，那么看一下这种是如何来进行工作的。
Propagation propagation() default Propagation.REQUIRED;
那么直接看下如何来开启一个事务的。
startTransaction
/**
* Start a new transaction.
*/
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;
}
1、首先来判断是否是一个新的事务；
2、不管是否，都创建一个新的对象默认的事务属性对象出来；
3、然后根据默认事务属性开启事务；
4、准备同步资源；
/**
* Create a TransactionStatus instance for the given arguments.
*/
protected DefaultTransactionStatus newTransactionStatus(
TransactionDefinition definition, @Nullable Object transaction, boolean newTransaction,
boolean newSynchronization, boolean debug, @Nullable Object suspendedResources) {
boolean actualNewSynchronization = newSynchronization &&
!TransactionSynchronizationManager.isSynchronizationActive();
return new DefaultTransactionStatus(
transaction, newTransaction, actualNewSynchronization,
definition.isReadOnly(), debug, suspendedResources);
}
如果是新的创建事务，那么这里将会是true，然后判断当前事务管理器中的同步是否被激活了。默认是没有激活的。
然后看下对应的创建状态：
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;
}
doBegin
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();
if (logger.isDebugEnabled()) {
logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
}
// 设置进来，表示的当前数据源对象是有资源的，因为设置了true，表示的是新的资源
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());
// Switch to manual commit if necessary. This is very expensive in some JDBC drivers,
// so we don't want to do it unnecessarily (for example if we've explicitly
// configured the connection pool to set it already).
// 默认的自动提交，下面改成手动提交。即：开始事务
if (con.getAutoCommit()) {
txObject.setMustRestoreAutoCommit(true);
if (logger.isDebugEnabled()) {
logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
}
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为数据库持有者
// Bind the connection holder to the thread.
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);
}
}
prepareSynchronization
按理来说，将数据库连接设置成了手动提交事务之后，下面应该来执行对应的事务了。但是spring因为支持了传播行为。
故还需要来做些事务同步操作的处理：
protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
// 如果是的新的事务
if (status.isNewSynchronization()) {
// 设置激活状态
TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
// 设置隔离级别
TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(
definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ?
definition.getIsolationLevel() : null);
// 设置是否只读
TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());
// 设置事务名称
TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());
// 这个时候来初始化事务同步器
TransactionSynchronizationManager.initSynchronization();
}
}
看一下事务同步器中的操作：
public static void initSynchronization() throws IllegalStateException {
if (isSynchronizationActive()) {
throw new IllegalStateException("Cannot activate transaction synchronization - already active");
}
synchronizations.set(new LinkedHashSet<>());
}
如果没有激活事务的时候，那么这里只是存入了一个LinkedHashSet集合。那么这个什么时候可以来进行注册呢？
在registerSynchronization方法上：
public static void registerSynchronization(TransactionSynchronization synchronization)
throws IllegalStateException {
Assert.notNull(synchronization, "TransactionSynchronization must not be null");
Set<TransactionSynchronization> synchs = synchronizations.get();
if (synchs == null) {
throw new IllegalStateException("Transaction synchronization is not active");
}
synchs.add(synchronization);
}
这里存入进去了之后会很有帮助的！！方便我们后续来进行操作。
prepareTransactionInfo
准备事务信息。
最重点的是
txInfo.bindToThread();
看下面这行代码：
private void bindToThread() {
// Expose current TransactionStatus, preserving any existing TransactionStatus
// for restoration after this transaction is complete.
this.oldTransactionInfo = transactionInfoHolder.get();
transactionInfoHolder.set(this);
}
将之前存的事务信息存起来，然后将新的事务信息进行返回！
返回之后，就可以根据条件来进行自定义操作了。
4.3、回滚
如果在自定义逻辑之间提交事务失败，那么将会进行事务回滚，看下是如何来进行事务回滚的：
completeTransactionAfterThrowing(txInfo, ex);
看下对应的信息：
protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {
if (txInfo != null && txInfo.getTransactionStatus() != null) {
if (logger.isTraceEnabled()) {
logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() +
"] after exception: " + ex);
}
// 得到事务的属性，不为空而且是指定的回滚类型！那么将会走到这里来
// 默认的回滚类型是RuntimeException或者是Error
if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
try {
// 获取得到事务管理器进行回滚
txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
}
catch (TransactionSystemException ex2) {
logger.error("Application exception overridden by rollback exception", ex);
ex2.initApplicationException(ex);
throw ex2;
}
catch (RuntimeException | Error ex2) {
logger.error("Application exception overridden by rollback exception", ex);
throw ex2;
}
}
else {
// We don't roll back on this exception.
// Will still roll back if TransactionStatus.isRollbackOnly() is true.
try {
txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
}
catch (TransactionSystemException ex2) {
logger.error("Application exception overridden by commit exception", ex);
ex2.initApplicationException(ex);
throw ex2;
}
catch (RuntimeException | Error ex2) {
logger.error("Application exception overridden by commit exception", ex);
throw ex2;
}
}
}
}
一般来说，这里设置的都是默认的。那么来看看默认的是如何来进行实现的：
private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
try {
// unexpected=false，这里表示的是希望回滚
boolean unexpectedRollback = unexpected;
try {
// 回滚前的操作
// 这里就是之前的同步器操作的设置
triggerBeforeCompletion(status);
// 是否有回滚点！如果有回滚点，这里可以设置
if (status.hasSavepoint()) {
if (status.isDebug()) {
logger.debug("Rolling back transaction to savepoint");
}
status.rollbackToHeldSavepoint();
}
// 是否是新的事务。这里的回滚就是数据库连接回滚
else if (status.isNewTransaction()) {
if (status.isDebug()) {
logger.debug("Initiating transaction rollback");
}
doRollback(status);
}
else {
// 如果不是新的数据库连接，表示的是复用的。应该取决于外部的事务决定
// Participating in larger transaction
if (status.hasTransaction()) {
// 是否设置了局部回滚或者是全局回滚(默认为true)
if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
if (status.isDebug()) {
logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
}
// 这里只是设置了一个标记而已！并不是真正的回滚操作
doSetRollbackOnly(status);
}
else {
if (status.isDebug()) {
logger.debug("Participating transaction failed - letting transaction originator decide on rollback");
}
}
}
else {
logger.debug("Should roll back transaction but cannot - no transaction available");
}
// Unexpected rollback only matters here if we're asked to fail early
if (!isFailEarlyOnGlobalRollbackOnly()) {
unexpectedRollback = false;
}
}
}
catch (RuntimeException | Error ex) {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
throw ex;
}
// 开始进行回调
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
// Raise UnexpectedRollbackException if we had a global rollback-only marker
if (unexpectedRollback) {
throw new UnexpectedRollbackException(
"Transaction rolled back because it has been marked as rollback-only");
}
}
finally {
// 释放掉当前事务信息、将数据库连接返回数据库连接池中(手动开始事务true--->false)
// 释放当前线程资源等等信息
cleanupAfterCompletion(status);
}
}
最终回滚之后，还需要将外部的事务恢复执行。
4.4、提交
在正常提交返回之后，开始来提交事务
commitTransactionAfterReturning(txInfo);
但是不要想看了这里的提交过程，因为提交里面还存在回滚操作：
public final void commit(TransactionStatus status) throws TransactionException {
if (status.isCompleted()) {
throw new IllegalTransactionStateException(
"Transaction is already completed - do not call commit or rollback more than once per transaction");
}
DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
// 如果上一次操作中让回滚了，那么这里也需要来进行回滚操作，而不是去真正的提交
if (defStatus.isLocalRollbackOnly()) {
if (defStatus.isDebug()) {
logger.debug("Transactional code has requested rollback");
}
processRollback(defStatus, false);
return;
}
// 全局操作，也不知道这里什么时候会执行到这里来
// 如果为true，也会导致回滚操作
if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
if (defStatus.isDebug()) {
logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");
}
processRollback(defStatus, true);
return;
}
// 开始提交操作
processCommit(defStatus);
}
提交操作：
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
try {
// 结束之前调用
boolean beforeCompletionInvoked = false;
try {
// 希望提交
boolean unexpectedRollback = false;
// 空实现
prepareForCommit(status);
// 在提交结束做的操作
// 用户自定义操作
triggerBeforeCommit(status);
triggerBeforeCompletion(status);
// 提交之前做的操作
beforeCompletionInvoked = true;
if (status.hasSavepoint()) {
if (status.isDebug()) {
logger.debug("Releasing transaction savepoint");
}
unexpectedRollback = status.isGlobalRollbackOnly();
status.releaseHeldSavepoint();
}
else if (status.isNewTransaction()) {
if (status.isDebug()) {
logger.debug("Initiating transaction commit");
}
unexpectedRollback = status.isGlobalRollbackOnly();
// 如果是新事务，那么在这里来进行提交
doCommit(status);
}
else if (isFailEarlyOnGlobalRollbackOnly()) {
unexpectedRollback = status.isGlobalRollbackOnly();
}
// 全局异常回滚点设置
// Throw UnexpectedRollbackException if we have a global rollback-only
// marker but still didn't get a corresponding exception from commit.
if (unexpectedRollback) {
throw new UnexpectedRollbackException(
"Transaction silently rolled back because it has been marked as rollback-only");
}
}
catch (UnexpectedRollbackException ex) {
// can only be caused by doCommit
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
throw ex;
}
catch (TransactionException ex) {
// can only be caused by doCommit
if (isRollbackOnCommitFailure()) {
doRollbackOnCommitException(status, ex);
}
else {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
}
throw ex;
}
catch (RuntimeException | Error ex) {
if (!beforeCompletionInvoked) {
triggerBeforeCompletion(status);
}
doRollbackOnCommitException(status, ex);
throw ex;
}
// 提交事务之后做的事情。用户自定义操作
// Trigger afterCommit callbacks, with an exception thrown there
// propagated to callers but the transaction still considered as committed.
try {
triggerAfterCommit(status);
}
finally {
// 事务完成之后做的事情。用户自定义操作
triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
}
}
finally {
// 开始清理对应的信息！
cleanupAfterCompletion(status);
}
}
同步情况
无非就是利用TransactionSynchronizationManager来做一下操作上的事情
public void doHandlTransaction() {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
transactionTemplate.execute(statis-> {
String currentTransactionName = TransactionSynchronizationManager.getCurrentTransactionName();
System.out.println("获取得到当前事务的名称，默认是方法名称："+currentTransactionName);
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void beforeCommit(boolean readOnly) {
System.out.println("在提交之前需要做的事情");
}
@Override
public void beforeCompletion() {
System.out.println("完成事务之前，需要做的事情");
}
@Override
public void afterCommit() {
System.out.println("在提交之后做的事情");
}
@Override
public void afterCompletion(int status) {
System.out.println("在事务结束之后做的事情");
}
});
List<User> users = userMapper.selectAllUser();
User user = users.get(0);
user.setEmail("hushidandan222");
userMapper.updateByPrimaryKeySelective(user);
user.setCreatedBy(333);
userMapper.updateByPrimaryKeySelective(user);
int i = 1/0;
return Boolean.TRUE;
});
}
控制台打印：
Releasing transactional SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@6492fab5]
Transaction synchronization deregistering SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@6492fab5]
Transaction synchronization closing SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@6492fab5]
完成事务之前，需要做的事情
在事务结束之后做的事情
Exception in thread "main" java.lang.ArithmeticException: / by zero
说明，完成事务之前和之后的，这里是不对是否正常提交事务做一个区分，而是提交和回滚都有的完成事务前和完成时候的这样子的一个操作。
那么看一下常规操作：
public void doHandlTransaction() {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
transactionTemplate.execute(statis-> {
String currentTransactionName = TransactionSynchronizationManager.getCurrentTransactionName();
System.out.println("获取得到当前事务的名称，默认是方法名称："+currentTransactionName);
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void beforeCommit(boolean readOnly) {
System.out.println("在提交之前需要做的事情");
}
@Override
public void beforeCompletion() {
System.out.println("完成事务之前，需要做的事情");
}
@Override
public void afterCommit() {
System.out.println("在提交之后做的事情");
}
@Override
public void afterCompletion(int status) {
System.out.println("在事务结束之后做的事情");
}
});
List<User> users = userMapper.selectAllUser();
User user = users.get(0);
user.setEmail("hushidandan222");
userMapper.updateByPrimaryKeySelective(user);
user.setCreatedBy(333);
userMapper.updateByPrimaryKeySelective(user);
return Boolean.TRUE;
});
}
控制台输出：
在提交之前需要做的事情
Transaction synchronization deregistering SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@6492fab5]
Transaction synchronization closing SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@6492fab5]
完成事务之前，需要做的事情
在提交之后做的事情
在事务结束之后做的事情
很有意思的操作。也可以参考一下SqlSessionTemplate中spring和mybatis整合的时候做的事情。
因为这是spring提供的TransactionSynchronizationManager在执行过程中进行回调的事情。
4.5、存在多个传播行为
存在多个行为的时候，这个时候spring支持的传播行为才会真正上线工作！那么来看下对应的实现原理：
类似如下：
@Transactional
public A a(){
b();
}
@Transactional
public B b(){
}
在执行事务a方法的时候，会来执行事务b方法。那么问题就在于这里来了！
先执行a方法，一切正常；当a方法在调用b方法的时候，会来执行b逻辑。
事务失效
但是需要知道的是：如果加了@Transactional是利用代理对象来调用方法的，才会走这里的事务方法。
所以如果是直接调用的话，那么就会导致事务失效。所以还是需要利用动态代理对象来进行调用的时候，才会重新走调用事务的方法。
这里有两种方式：
- 1、在当前类中注入自己，只不过注入的是代理对象而已
@Autowired
private A a;
@Transactional
public A a(){
a.b();
}
@Transactional
public B b(){
}
- 2、使用AopContext.currentProxy方法获取得到线程中的代理对象
利用这种方式解决的是上面的自己调用自己的情况。
- 修改类，不要出现“自调用”的情况：这是Spring文档中推荐的“最佳”方案；
- 若一定要使用“自调用”，那么this.b()替换为：((A) AopContext.currentProxy()).b()；此时需要修改spring的aop配置：
<aop:aspectj-autoproxy expose-proxy="true" />
如果是Spring配置方式或者是springboot项目，加上注解：@EnableAspectJAutoProxy(exposeProxy = true)
那么利用AopContext.currentProxy()即可获取得到线程中的当前类的代理对象进行操作了。
<tx:annotation-driven transaction-manager="transactionManager"
proxy-target-class="true"/>
注意：proxy-target-class属性值决定是基于接口的还是基于类的代理被创建。如果proxy-target-class 属性值被设置为true，那么基于类的代理将起作用（这时需要cglib库）。如果proxy-target-class属值被设置为false或者这个属性被省略，那么标准的JDK 基于接口的代理将起作用。
即使你未声明 proxy-target-class="true" ，但运行类没有继承接口，spring也会自动使用CGLIB代理。
高版本spring自动根据运行类选择 JDK 或 CGLIB 代理
传播行为调用传播行为
使用代理对象来进行调用的时候，可以看到，还是会走到TransactionInterceptor里面，然后执行一下第一次走过的逻辑
重点看一下下面几行代码：
Object transaction = doGetTransaction();
if (isExistingTransaction(transaction)) {
// Existing transaction found -> check propagation behavior to find out how to behave.
return handleExistingTransaction(def, transaction, debugEnabled);
}
Object transaction = doGetTransaction();
protected Object doGetTransaction() {
DataSourceTransactionObject txObject = new DataSourceTransactionObject();
txObject.setSavepointAllowed(isNestedTransactionAllowed());
ConnectionHolder conHolder =
(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
txObject.setConnectionHolder(conHolder, false);
return txObject;
}
这里会从当前线程中获取得到对应的数据库连接对象，但是因为是复用上一次事务存入留下的数据库连接，所以这里设置成了false。
然后看下检查是否存在事务代码：
protected boolean isExistingTransaction(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
}
这里获取得到的是从上面获取得到的connectionholder且事务处于活跃状态。
下面就开始来处于事务代码
首先做的事情就是来判断传播行为，这里传播行为中的核心代码：
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) {
if (debugEnabled) {
logger.debug("Suspending current transaction");
}
Object suspendedResources = suspend(transaction);
boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
return prepareTransactionStatus(
definition, null, false, newSynchronization, debugEnabled, suspendedResources);
}
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
if (debugEnabled) {
logger.debug("Suspending current transaction, creating new transaction with name [" +
definition.getName() + "]");
}
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 (debugEnabled) {
logger.debug("Creating nested transaction with name [" + definition.getName() + "]");
}
if (useSavepointForNestedTransaction()) {
// Create savepoint within existing Spring-managed transaction,
// through the SavepointManager API implemented by TransactionStatus.
// Usually uses JDBC 3.0 savepoints. Never activates Spring synchronization.
DefaultTransactionStatus status =
prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
status.createAndHoldSavepoint();
return status;
}
else {
// Nested transaction through nested begin and commit/rollback calls.
// Usually only for JTA: Spring synchronization might get activated here
// in case of a pre-existing JTA transaction.
return startTransaction(definition, transaction, debugEnabled, null);
}
}
// Assumably PROPAGATION_SUPPORTS or PROPAGATION_REQUIRED.
if (debugEnabled) {
logger.debug("Participating in existing transaction");
}
if (isValidateExistingTransaction()) {
if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
Integer currentIsolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
if (currentIsolationLevel == null || currentIsolationLevel != definition.getIsolationLevel()) {
Constants isoConstants = DefaultTransactionDefinition.constants;
throw new IllegalTransactionStateException("Participating transaction with definition [" +
definition + "] specifies isolation level which is incompatible with existing transaction: " +
(currentIsolationLevel != null ?
isoConstants.toCode(currentIsolationLevel, DefaultTransactionDefinition.PREFIX_ISOLATION) :
"(unknown)"));
}
}
if (!definition.isReadOnly()) {
if (TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
throw new IllegalTransactionStateException("Participating transaction with definition [" +
definition + "] is not marked as read-only but existing transaction is");
}
}
}
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
}
免责声明：本站所有文章内容,图片，视频等均是来源于用户投稿和互联网及文摘转载整编而成，不代表本站观点，不承担相关法律责任。其著作权各归其原作者或其出版社所有。如发现本站有涉嫌抄袭侵权/违法违规的内容,侵犯到您的权益，请在线联系站长,一经查实,本站将立刻删除。 本文来自网络,若有侵权，请联系删除，如若转载，请注明出处：https://yundeesoft.com/30372.html
