IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		图片批量下载器
↓批量下载图片,美女图库↓ 		图片自动播放器
↓图片自动播放器↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> Java知识库 -> Spring - 声明式事务 - 源码分析 -> 正文阅读

[Java知识库]Spring - 声明式事务 - 源码分析

参考：

  • <<Spring源码深度解析 第2版>>

一、声明式事务

1.1 @Transactional

在分析源码前，我们必须对声明式事务的使用，即@Transactional注解比较熟悉，因为源码中就是根据注解的属性进行解析，创建不同的事务，特别是传播机制。

@Transactional注解可以放在类或方法上：

  • 类上：指定某个类及子类中所有方法
  • 方法上：指定某个方法启用事务

@Transactional属性如下表所示：

属性默认值描述
value/transactionManager“”事务管理器
propagationPropagation.REQUIRED传播机制
isolationIsolation.DEFAULT隔离级别，默认使用数据库的隔离级别，例如MySQL的可重复读
timeoutTransactionDefinition.TIMEOUT_DEFAULT事务超时时间，单位：秒，默认不超时
readOnlyfalse事务只读，默认非只读
rollbackFor指定哪些异常需要回滚事务，默认当异常是RuntimeException、Error会回滚事务
rollbackForClassName回滚事务的异常名称
noRollbackFor指定哪些异常不需要回滚事务
noRollbackForClassName不需要回滚事务的异常名称

1.1.1 propagation：传播机制

传播机制是非常重要的一个属性之一了，参考org.springframework.transaction.annotation.Propagation枚举类，定义了所有的传播行为，可选值：

  • REQUIRED：默认的传播机制，支持当前事务，如果不存在则创建一个新事务
  • SUPPORTS：支持当前事务，如果不存在则以非事务方式执行
  • MANDATORY：支持当前事务，如果不存在则抛出异常
  • REQUIRES_NEW：创建一个新事务，如果存在，则挂起当前事务
  • NOT_SUPPORTED：以非事务方式执行，如果存在则挂起当前事务
  • NEVER：以非事务方式执行，如果存在事务则抛出异常
  • NESTED：如果当前事务存在，则在嵌套事务中执行，其他行为类似于REQUIRED
1.1.1.1 REQUIRED 示例

（1）示例1

testA()、testB()传播机制都是REQUIRED，所以这两个方法在同一个事务中：

  • 不管在什么位置发生了异常，两个方法都会回滚，也就是a、b无法保存至数据库
// AService#testA()
@Transactional(propagation = Propagation.REQUIRED)
public void testA() {
	insert(a);
	bService.testB()
}

// BService#testB()
@Transactional(propagation = Propagation.REQUIRED)
public void testB() {
	insert(b);
	throws Exception();
}

（2）示例2

testA()不使用事务、testB()使用事务且传播机制都是REQUIRED，testA()没有事务，testB()会新建一个事务：

  • testA()中发生异常：a、b都可以正常保存，因为testB()有自己的事务
  • testB()中发生了异常：a可以保存，b无法保存
// AService#testA()
public void testA() {
	insert(a);
	bService.testB()
	throws Exception();
}

// BService#testB()
@Transactional(propagation = Propagation.REQUIRED)
public void testB() {
	insert(b);
}

（3）示例3

testA()、testB()使用事务且传播机制都是REQUIRED，testA()和testB()在同一个事务中：

  • 若testA()中发生异常，并且捕获了该异常：a、b可以保存，因为捕获了异常，所以相当于正常执行
  • 若testB()中发生异常，并且在testA()中捕获了testB()抛出的异常：a、b无法保存
    • 因为testB()中发生了异常，造成当前事务回滚，并标记事务为rollback-only
    • testA()捕获了异常，所以testA()会走正常的提交事务流程，但是发现事务状态已经标记为回滚了，所以抛出异常：UnexpectedRollbackException: Transaction rolled back because it has been marked as rollback-only，所以也就只能回滚了（debug一下会比较容易理解）
// AService#testA()
@Transactional(propagation = Propagation.REQUIRED)
public void testA() {
	insert(a);
	bService.testB()
	try {
		throws Exception();
	} catch (Exception e){
	}
	
}

// BService#testB()
@Transactional(propagation = Propagation.REQUIRED)
public void testB() {
	insert(b);
}

1.1.1.2 NESTED示例

（1）示例1

testA()开启一个事务（父事务）、testB()开启一个嵌套事务（子事务）：

  • 若testA()中发生异常：a、b无法保存，即父事务发生了异常，父子事务都会回滚
  • 若testB()中发生异常：a、b也无法保存，即子事务发生了异常，父子事务都会回滚
// AService#testA()
@Transactional(propagation = Propagation.REQUIRED)
public void testA() {
	insert(a);
	bService.testB()
	throws Exception();
}

// BService#testB()
@Transactional(propagation = Propagation.NESTED)
public void testB() {
	insert(b);
}

（2）示例2

testA()开启一个事务（父事务）、testB()开启一个嵌套事务（子事务）：

  • 若testA()中发生异常，并且捕获了该异常：a、b可以保存，因为捕获了异常，所以相当于正常执行
  • 若testB()中发生异常，并且在testA()中捕获了testB()抛出的异常：a可以保存，b无法保存，即子事务发生了异常，子事务会回滚，不影响父事务（对比下1.1.1.1中的示例3，都设置为REQUIRED，但是a、b都无法保存的）
// AService#testA()
@Transactional(propagation = Propagation.REQUIRED)
public void testA() {
	insert(a);
	try {
		bService.testB();
	} catch (Exception e) {
	}
	
	try {
		throws Exception();
	} catch (Exception e) {
	}
}

// BService#testB()
@Transactional(propagation = Propagation.NESTED)
public void testB() {
	insert(b);
}

其余属性暂时省略…

二、源码分析

声明式事务核心类是org.springframework.transaction.interceptor.TransactionInterceptor，在类或方法添加@Transactional注解后，Spring就会基于AOP为该类生成代理类，在调用目标方法时，就会触发TransactionInterceptor#invoke()方法执行事务。

TransactionInterceptor类图如下图所示：
在这里插入图片描述

事务执行的答题流程：

  • 获取事务属性
  • 加载配置中配置的TransactionManager
  • 不同的事务处理方式使用不同的逻辑
  • 在目标方法执行前获取事务并收集事务信息
  • 执行目标方法
  • 出现异常，进行异常处理
  • 提交事务前将事务信息清除
  • 提交事务

下面具体分析下事务的创建、提交、回滚等相关源码。

2.1 事务执行入口

入口方法在TransactionInterceptor#invoke()中，主要流程：

  • 获取目标类、目标方法
  • 调用父类TransactionAspectSupport#invokeWithinTransaction()执行事务

源码如下：

// TransactionInterceptor#invoke()
@Override
public Object invoke(final MethodInvocation invocation) throws Throwable {
	// 获取目标类
	Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);

	// 调用父类TransactionAspectSupport的invokeWithinTransaction()
	return invokeWithinTransaction(invocation.getMethod(), targetClass, new InvocationCallback() {
		@Override
		public Object proceedWithInvocation() throws Throwable {
			return invocation.proceed();
		}
	});
}

2.2 事务执行主流程

事务执行主要发生在TransactionAspectSupport#invokeWithinTransaction()方法中，主要流程：

  • 获取事务属性、事务管理器
  • 判断是声明式事务、编程式事务，走不同的分支：
    • 声明式事务：执行以下流程：
      • 创建事务
      • 执行目标方法
      • 发生异常则回滚
      • 提交事务前将事务信息清除
      • 提交事务
    • 编程式事务：这里不讨论

源码如下：

// TransactionAspectSupport#invokeWithinTransaction()
protected Object invokeWithinTransaction(Method method, Class<?> targetClass, final InvocationCallback invocation) throws Throwable {
	// 获取事务属性
	final TransactionAttribute txAttr = getTransactionAttributeSource().getTransactionAttribute(method, targetClass);
	// 获取beanFactory中的事务管理器，通常是DataSourceTransactionManager
	final PlatformTransactionManager tm = determineTransactionManager(txAttr);
	// 构造方法唯一标识（类完全限定名.方法，如：com.xxx.service.UserService.save）
	final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
	// 声明式事务处理
	if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
		// 假如必要的话，创建事务
		TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
		Object retVal = null;
		try {
			// 环绕通知：执行下一个拦截器
			// 通常是执行被增强的方法
			retVal = invocation.proceedWithInvocation();
		} catch (Throwable ex) {
			// 目标执行异常，回滚
			completeTransactionAfterThrowing(txInfo, ex);
			throw ex;
		} finally {
			// 清理事务信息：恢复之前的事务信息
			cleanupTransactionInfo(txInfo);
		}
		// 返回前提交事务
		commitTransactionAfterReturning(txInfo);
		return retVal;
	} else {
		// 编程式事务处理，此处省略...
	}
}

从代码可以发现Spring事务源码整体风格：

  • 搭建骨架，也就是通过调用方法，具体逻辑都在各个方法里
  • 方法返回值，为下一步骤（方法）做准备
  • 通过继承或者说大量利用设计模式，调用父类方法实现，再根据配置选择具体的实现

所以乍一看源码比较清晰，深入看就很复杂，各种抽象类，实现类也很多（代表扩展性很好，很标准的模板设计模式），非常值得学习。。


2.2.1 创建事务

创建事务逻辑在TransactionAspectSupport#createTransactionIfNecessary()方法中，主要流程：

  • 使用 DelegatingTransactionAttribute 封装 事务属性
  • 获取事务
  • 构造事务信息
// TransactionAspectSupport#createTransactionIfNecessary()
protected TransactionInfo createTransactionIfNecessary(PlatformTransactionManager tm, TransactionAttribute txAttr, final String joinpointIdentification) {
	// 如果没有名称指定则使用方法唯一标识，并使用 DelegatingTransactionAttribute 封装 txAttr
	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) {
			// 根据事务属性获取事务TransactionStatus，调用PlatformTransactionManager.getTransaction()。
			status = tm.getTransaction(txAttr);
		} else {
			// log
		}
	}
	// 构造一个TransactionInfo事务信息对象，绑定当前线程：ThreadLocal<TransactionInfo>
	return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

2.1.1.1获取事务

AbstractPlatformTransactionManager#getTransaction()来处理事务的准备工作，包括事务获取以及信息的构建，主要流程：

  • 1.获取事务
  • 2.如果当前线程存在事务则转向嵌套事务的处理
  • 3.事务超时设置验证
  • 4.事务传播机制校验
  • 5.构建 DefaultTransactionStatus
  • 6.完善 transaction，包括设置 ConnectionHolder、 隔离级别、 timeout，如果是新连接，则绑定到当前线程
// AbstractPlatformTransactionManager#getTransaction
/**
 * 获取事务，处理事务传播机制，分派给doGetTransaction()、isExistingTransaction()、doBegin()三个方法处理
 */
@Override
public final TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {
	// 1.获取事务
	// 调用DataSourceTransactionManager#doGetTransaction()
	// 创建基于JDBC的事务实例：DataSourceTransactionObject
	Object transaction = doGetTransaction();

	// 假如没有事务定义，使用默认的
	if (definition == null) {
		definition = new DefaultTransactionDefinition();
	}

	// 2.当先线程存在事务，处理嵌套事务 
	// 2.1判断当前线程是否存在事务，判断依据为当前线程记录的连接不为空且连接中(connectionHolder) 中的transactionActive属性不为空
	if (isExistingTransaction(transaction)) {
		// 2.2当前线程已经存在事务，检查传播机制如何传播
		return handleExistingTransaction(definition, transaction, debugEnabled);
	}

	// 3.事务超时设置验证
	if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
		throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
	}

	// 4.事务传播机制校验
	// 当前线程不存在事务，但是传播机制设置为MANDATORY，则会抛出异常
	// MANDATORY：当前存在事务，则加入当前事务，如果当前事务不存在，则抛出异常
	if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
		throw new IllegalTransactionStateException("No existing transaction found for transaction marked with propagation 'mandatory'");
	} else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
			definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
			definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
		// REQUIRED、REQUIRES_NEW、NESTED这3种传播机制都需要新建事务

		// 挂起，由于传入的是null，该方法返回值就是null，相当于没做任何处理
		SuspendedResourcesHolder suspendedResources = suspend(null);
		
		try {
		    // 构造事务状态，包括事务定义、事务信息、新事务的标识等
			boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
			DefaultTransactionStatus status = newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
			// 构造transaction，包括设置ConnectionHolder、隔离级别、timeout
			// 如果是新连接，绑定到当前线程
			doBegin(transaction, definition);
			// 新同步事务的设置，针对当前线程的设置
			prepareSynchronization(status, definition);
			return status;
		} catch (RuntimeException ex) {
			resume(null, suspendedResources);
			throw ex;
		} catch (Error err) {
			resume(null, suspendedResources);
			throw err;
		}
	} else {
		// 其他传播机制SUPPORTS、NOT_SUPPORTED、NEVER：不使用事务
		if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
			logger.warn("Custom isolation level specified but no actual transaction initiated; isolation level will effectively be ignored: " + definition);
		}
		boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
		return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null);
	}
}
（1）doGetTransaction()：获取事务

主要流程：

  • 如果当前线程中存在dataSource的连接，那么直接使用
  • 开启允许保存点：是否设置了允许嵌套事务
// DataSourceTransactionManager#doGetTransaction()
protected Object doGetTransaction() {
	DataSourceTransactionObject txObject = new DataSourceTransactionObject();
	// 
	txObject.setSavepointAllowed(isNestedTransactionAllowed());
	// 如果当前线程已经记录数据库连接，则使用原有连接
	ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(this.dataSource);
	// false 表示非新创建连接
	txObject.setConnectionHolder(conHolder, false);
	return txObject;
}

（2）doBegin()

当前线程没有事务时，且目标方法的传播机制要求新建事务，那么就会执行该方法。

主要流程如下：

  • 尝试获取连接
  • 设置隔离级别以及只读标识
  • 更改默认的提交设置
  • 设置标志位，标识当前连接已经被事务激活
  • 设置过期时间
  • 将 connectionHolder绑定到当前线程
  • 将事务信息记录在当前线程中
// DataSourceTransactionManager#doBegin()
protected void doBegin(Object transaction, TransactionDefinition definition) {
	DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
	Connection con = null;
	try {
		// 若当前线程没有数据库连接 或 资源与事务同步，则从数据源中获取连接，并绑定到当前线程上
		if (!txObject.hasConnectionHolder() || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
			Connection newCon = this.dataSource.getConnection();
			txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
		}
		// 设置资源与事务同步
		txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
		con = txObject.getConnectionHolder().getConnection();
		// 设置连接属性：是否需要设置连接只读、设置隔离级别，不指定就是默认的隔离级别，取决于数据库
		Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
		txObject.setPreviousIsolationLevel(previousIsolationLevel);

		// 事务提交由自动提交修改为手动提交
		if (con.getAutoCommit()) {
			txObject.setMustRestoreAutoCommit(true);
			con.setAutoCommit(false);
		}
		// 在事务开始前，设置事务只读：enforceReadOnly=true 且 设置了事务只读
		// 假如满足的条件的话，就会执行sql：`SET TRANSACTION READ ONLY`
		prepareTransactionalConnection(con, definition);
		
		// 设置判断当前线程是否存在事务的依据：设置transactionActive=ture
		txObject.getConnectionHolder().setTransactionActive(true);

		// 事务超时时间：没有设置则默认不超时，-1
		int timeout = determineTimeout(definition);
		if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
			txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
		}

		// 将当前获取到的连接绑定到当前线程
		if (txObject.isNewConnectionHolder()) {
			TransactionSynchronizationManager.bindResource(getDataSource(), txObject.getConnectionHolder());
		}
	} catch (Throwable ex) {
		if (txObject.isNewConnectionHolder()) {
			DataSourceUtils.releaseConnection(con, this.dataSource);
			txObject.setConnectionHolder(null, false);
		}
		throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
	}
}

（3）prepareSynchronization()

将事务信息记录在当前线程中：新连接、隔离级别、是否只读、事务名称等

// AbstractPlatformTransactionManager#prepareSynchronization()
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();
	}
}

（4）handleExistingTransaction()

当前线程存在事务，嵌套事务的处理，就会执行该方法。
// TODO

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) {
		Object suspendedResources = suspend(transaction);
		boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
		return prepareTransactionStatus(definition, null, false, newSynchronization, debugEnabled, suspendedResources);
	}

	if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
		SuspendedResourcesHolder suspendedResources = suspend(transaction);
		try {
			boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
			DefaultTransactionStatus status = newTransactionStatus(
					definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
			doBegin(transaction, definition);
			prepareSynchronization(status, definition);
			return status;
		}catch (RuntimeException beginEx) {
			resumeAfterBeginException(transaction, suspendedResources, beginEx);
			throw beginEx;
		}catch (Error beginErr) {
			resumeAfterBeginException(transaction, suspendedResources, beginErr);
			throw beginErr;
		}
	}

	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 (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.
			boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
			DefaultTransactionStatus status = newTransactionStatus(
					definition, transaction, true, newSynchronization, debugEnabled, null);
			doBegin(transaction, definition);
			prepareSynchronization(status, definition);
			return status;
		}
	}

	// Assumably PROPAGATION_SUPPORTS or PROPAGATION_REQUIRED.
	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);
}

2.1.2 准备事务信息

当已经建立事务连接并完成了事务信息的提取后，将所有的事务信息记录在Transactionlnfo 类型的实例中，这个实例包含了目标方法开始前的所有状态信息，一旦事务执行失败， Spring会通过 Transactionlnfo类型的实例中的信息来进行回滚等后续工作。

// TransactionAspectSupport#prepareTransactionInfo()
protected TransactionInfo prepareTransactionInfo(PlatformTransactionManager tm, TransactionAttribute txAttr, String joinpointIdentification, TransactionStatus status) {
	// 事务信息：事务管理器、事务属性、事务名称
	TransactionInfo txInfo = new TransactionInfo(tm, txAttr, joinpointIdentification);
	if (txAttr != null) {
		// 记录事务状态
		txInfo.newTransactionStatus(status);
	} else {
		// 省略日志...
	}
	// 将事务信息绑定到当前线程
	txInfo.bindToThread();
	return txInfo;
}

2.2.2 回滚

当目标方法执行出现异常时，就需要执行回滚操作了，回滚逻辑在TransactionAspectSupport#completeTransactionAfterThrowing()方法中，主要流程：

  • 判断是否有事务：有事务才可能需要回滚
  • 是否需要回滚：默认只有时RuntimeException或Error类型才要回滚

代码：

protected void completeTransactionAfterThrowing(TransactionInfo txInfo, Throwable ex) {
	// 事务存在才要回滚
	if (txInfo != null && txInfo.hasTransaction()) {
		// RuntimeException或Error的类型的异常才需要回滚
		if (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 ex2) {
				logger.error("Application exception overridden by rollback exception", ex);
				throw ex2;
			} catch (Error err) {
				logger.error("Application exception overridden by rollback error", ex);
				throw err;
			}
		} else {
			// 如果不满足回滚条件即使抛出异常也同样会提交
			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 ex2) {
				logger.error("Application exception overridden by commit exception", ex);
				throw ex2;
			} catch (Error err) {
				logger.error("Application exception overridden by commit error", ex);
				throw err;
			}
		}
	}
}

2.2.2.1 回滚的条件

默认的回滚条件在DefaultTransactionAttribute#rollbackOn()方法中，很明显可以看出只有RuntimeException或Error类型的异常才需要回滚，否则就算抛出异常也会正常提交。

// DefaultTransactionAttribute#rollbackOn()
public boolean rollbackOn(Throwable ex) {
	return (ex instanceof RuntimeException || ex instanceof Error);
}

假如有需要，可以通过@Transactional注解的rollbackFor就可以更改：
@Transactl。nal(propagation=Propagation.REQUIRED, rollbackFor=Exception class)

2.2.2.2 回滚处理

回滚逻辑在rollback()方法里，最终调用了processRollback()方法完成实际的回滚操作。主要流程：

  • 如果事务已经完成，再次回滚就会抛出异常
  • 事务没有完成，处理回滚：
    • 当前事务有保存点时，使用保存点信息进行回滚（常用于嵌套事务，内嵌的事务异常不会引起外部事务的回滚）
    • 当前事务是新事物，直接回滚
    • 其他情况，表示有事务但是又不属于上面两种情况，一般是JTA，那么只做回滚标识，等到提交的时候统一不提交
  • 自定义触发器的调用：
    • 回滚前、完成回滚、回滚发生异常时都需要调用，自定义的触发器会根据这些信息作进一步处理
  • 清空记录的资源并将挂起的资源恢复
// AbstractPlatformTransactionManager#rollback()
public final void rollback(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;
	// 处理回滚
	processRollback(defStatus);
}

// AbstractPlatformTransactionManager#processRollback()
private void processRollback(DefaultTransactionStatus status) {
	try {
		try {
			// 回滚前触发的一些操作：激活所有TransactionSynchronization 中对应的方法
			// 例如：关闭MyBatis会话
			triggerBeforeCompletion(status);
			if (status.hasSavepoint()) {
				// 如果有保存点，也就是当前事务为单独的线程则会退到保存点
				status.rollbackToHeldSavepoint();
			} else if (status.isNewTransaction()) {
				// 如果当前事务为独立的新事务，则直接回滚
				// 调用数据库连接的rollback()方法
				doRollback(status);
			} else if (status.hasTransaction()) {
				// 如果当前事务不是独立的事务，那么只能标记状态，等到事务链执行完毕后统一回滚
				if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {	
					doSetRollbackOnly(status);
				} else {
					// 日志...
				}
			} else {
				// 日志...
			}
		} catch (RuntimeException ex) {
			triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
			throw ex;
		} catch (Error err) {
			triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
			throw err;
		}
		// 回滚完成后触发的操作：激活所有 TransactionSynchronization 中对应的方法
		// 重置一些事务状态，如事务已关闭等
		triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
	} finally {
		// 清空记录的资源并将挂起的资源恢复
		cleanupAfterCompletion(status);
	}
}

2.2.2.3 回滚后的信息清除

回滚逻辑执行结束后，无论回滚是否成功，都必须要做的事情就是事务结束后的收尾工作。主要流程：

  • 设置状态是对事务信息作完成标识以避免重复调用
  • 如果当前事务是新的同步状态，需要将绑定到当前线程的事务信息清除
  • 如果是新事务需要做些清除资源的工作
// AbstractPlatformTransactionManager#cleanupAfterCompletion()
private void cleanupAfterCompletion(DefaultTransactionStatus status) {
	// 设置完成状态
	status.setCompleted();
	// 清除绑定在线程上的事务信息，可以参考2.1.1.1节的（3）prepareSynchronization()方法
	if (status.isNewSynchronization()) {
		TransactionSynchronizationManager.clear();
	}
	// 独立的事务：清除绑定在线程上的数据源、恢复数据库连接为自动提交、隔离级别、释放数据库连接
	if (status.isNewTransaction()) {
		doCleanupAfterCompletion(status.getTransaction());
	}
	if (status.getSuspendedResources() != null) {
		// 如果在事务执行前有事务挂起，那么当前事务执行结束后需要将挂起事务恢复 
		resume(status.getTransaction(), (SuspendedResourcesHolder) status.getSuspendedResources());
	}
}

// AbstractPlatformTransactionManager#doCleanupAfterCompletion()
protected void doCleanupAfterCompletion(Object transaction) {
	DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
	// 将数据库连接从当前线程中解除绑定
	if (txObject.isNewConnectionHolder()) {
		TransactionSynchronizationManager.unbindResource(this.dataSource);
	}
	// 释放连接
	Connection con = txObject.getConnectionHolder().getConnection();
	try {
		// 恢复数据库连接的自动提交属性
		if (txObject.isMustRestoreAutoCommit()) {
			con.setAutoCommit(true);
		}
		// 重置数据库连接
		DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel());
	} catch (Throwable ex) {
		logger.debug("Could not reset JDBC Connection after transaction", ex);
	}

	if (txObject.isNewConnectionHolder()) {
		// 如果当前事务是独立的新创建的事务则在事务完成时释放数据库连接
		DataSourceUtils.releaseConnection(con, this.dataSource);
	}
	txObject.getConnectionHolder().clear();
}

// AbstractPlatformTransactionManager#resume()
// 如果在事务执行前有事务挂起，那么当前事务执行结束后需要将挂起事务恢复 
protected final void resume(Object transaction, SuspendedResourcesHolder resourcesHolder) throws TransactionException {

	if (resourcesHolder != null) {
		Object suspendedResources = resourcesHolder.suspendedResources;
		if (suspendedResources != null) {
			doResume(transaction, suspendedResources);
		}
		List<TransactionSynchronization> suspendedSynchronizations = resourcesHolder.suspendedSynchronizations;
		if (suspendedSynchronizations != null) {
			TransactionSynchronizationManager.setActualTransactionActive(resourcesHolder.wasActive);
			TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(resourcesHolder.isolationLevel);
			TransactionSynchronizationManager.setCurrentTransactionReadOnly(resourcesHolder.readOnly);
			TransactionSynchronizationManager.setCurrentTransactionName(resourcesHolder.name);
			doResumeSynchronization(suspendedSynchronizations);
		}
	}
}

2.2.3 提交事务

当目标方法正常执行后，就需要执行提交操作了，提交逻辑在TransactionAspectSupport#commitTransactionAfterReturning()方法中，主要流程：

  • 判断是否有事务：有事务才需要提交事务
  • 获取事务管理器，执行事务提交操作
    • 事务状态是已完成，不能再执行提交操作，抛出异常

代码：

// TransactionAspectSupport#commitTransactionAfterReturning()
protected void commitTransactionAfterReturning(TransactionInfo txInfo) {
	if (txInfo != null && txInfo.hasTransaction()) {
		// 获取事务管理器执行提交操作
		txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
	}
}

// AbstractPlatformTransactionManager#commit()
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()) {
		processRollback(defStatus);
		return;
	}
	// 全局事务被标记为回滚，则执行回滚操作
	if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
		processRollback(defStatus);
		// 仅在最外层事务边界或明确要求时抛出异常
		if (status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()) {
			throw new UnexpectedRollbackException("Transaction rolled back because it has been marked as rollback-only");
		}
		return;
	}
	// 处理事务提交
	processCommit(defStatus);
}

2.2.3.1 处理事务提交

处理事务提交逻辑在AbstractPlatformTransactionManager#processCommit()方法中，主要流程如下：

  • TransactionSynchronization 中的对应方法的调用
  • 有保存点、不是新事务，则不会执行提交事务操作
  • 假如有全局回滚标识，但是没有在事务提交中获取到相关异常，则抛出异常
  • 处理提交事务过程中出现异常，则进行回滚
  • 事务提交完成后执行一些清理

在提交过程中事务也不是直接提交的：

  • 当事务状态中有保存点信息，不会去提交事务
  • 当事务非新事务的时，不会去提交事务

不能直接提交事务是主要考虑：嵌套事务

  • 对于嵌套事务，正常的处理方式是将嵌套事务开始之前设置保存点
  • 如果嵌套事务出现异常，根据保存点信息进行回滚
  • 如果没有出现异常，嵌套事务并不会单独提交， 而是根据事务由最外层事务负责提交
  • 所以如果存在保存点就表示不是最外层事务，不做保存操作（如果不是新事务则不提交也是类似考虑）
// AbstractPlatformTransactionManager#processCommit()
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
	try {
		boolean beforeCompletionInvoked = false;
		try {
			// 预留，空实现
			prepareForCommit(status);
			// 事务提交前触发的回调：添加的TransactionSynchronization 中的对应方法的调用
			// 如：MyBatis会话的提交
			triggerBeforeCommit(status);
			// 事务完成前触发的回调：添加的 TransactionSynchronization 中的对应方法的调用
			// 如：清除绑定在当前线程的MyBatis的sessionFactory、holderActive=false、关闭会话等
			triggerBeforeCompletion(status);
			beforeCompletionInvoked = true;
			boolean globalRollbackOnly = false;
			if (status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()) {
				globalRollbackOnly = status.isGlobalRollbackOnly();
			}
			// 如果存在保存点则清除保存点信息
			if (status.hasSavepoint()) {
				status.releaseHeldSavepoint();
			} else if (status.isNewTransaction()) {
				// 如果是独立的事务则直接提交，获取绑定在当前线程上的连接，执行commit()方法
				doCommit(status);
			}
			// 假如有全局回滚标识，但是没有在事务提交中获取到相关异常，则抛出UnexpectedRollbackException
			if (globalRollbackOnly) {
				throw new UnexpectedRollbackException("Transaction silently rolled back because it has been marked as rollback-only");
			}
		} catch (UnexpectedRollbackException ex) {
			// 捕获上面抛出的UnexpectedRollbackException异常
			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 ex) {
			if (!beforeCompletionInvoked) {
				triggerBeforeCompletion(status);
			}
			// 事务提交出现异常，执行回滚操作
			doRollbackOnCommitException(status, ex);
			throw ex;
		} catch (Error err) {
			// 添加的 TransactionSynchronization 中的对应方法的调用
			if (!beforeCompletionInvoked) {
				triggerBeforeCompletion(status);
			}
			// 提交过程中出现异常回滚
			doRollbackOnCommitException(status, err);
			throw err;
		}

		// Trigger afterCommit callbacks, with an exception thrown there
		// propagated to callers but the transaction still considered as committed.
		try {
			// 事务提交后触发的回调：添加的 TransactionSynchronization 中的对应方法的调用
			triggerAfterCommit(status);
		} finally {
			triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
		}
	} finally {
		// 事务提交完成后执行一些清理：
		// 设置事务状态已完成、清理绑定在线程上的事务信息、数据源、设置连接自动提交、恢复隔离级别、释放数据库连接
		cleanupAfterCompletion(status);
	}
}
2.2.3.2 执行事务提交

这部分逻辑比较简单：

  • 获取当前事务使用的连接，执行提交操作
  • 捕获SQLException，直接抛出事务异常
protected void doCommit(DefaultTransactionStatus status) {
	DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
	Connection con = txObject.getConnectionHolder().getConnection();
	try {
		con.commit();
	} catch (SQLException ex) {
		throw new TransactionSystemException("Could not commit JDBC transaction", ex);
	}
}

2.3 总结

整体上大概过了一遍事务创建、提交、回滚等流程，以及对于各种隔离级别的事务的处理，实际上还有很多细节也没整明白，先放放吧。

建议通过在不同隔离级别下，设置抛出异常的位置，对照实际的结果进行debug，这样比较容易理解源码各个if else分支的含义。
  Java知识库 最新文章
计算距离春节还有多长时间
系统开发系列 之WebService（spring框架+ma
springBoot+Cache(自定义有效时间配置)
SpringBoot整合mybatis实现增删改查、分页查
spring教程
SpringBoot+Vue实现美食交流网站的设计与实
虚拟机内存结构以及虚拟机中销毁和新建对象
SpringMVC---原理
小李同学: Java如何按多个字段分组
打印票据--java
上一篇文章      下一篇文章      查看所有文章
加:2021-08-22 13:25:07  更:2021-08-22 13:26:41 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年11日历 -2024/11/23 8:55:10-

图片自动播放器
↓图片自动播放器↓ 		TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓ 		图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码