Spring事务源码解析
- 一、基本概念
- 1、ACID属性
- 2、事务的隔离级别
- 3、事务行为
- 4、Spring事务的传播级别
- 5、Spring事务支持方式
- 二、Spring事务的执行源码
- 1、事务AOP
- 2、事务处理拦截器TransactionInterceptor
- 2.1 主要流程
- 2.2 尝试创建事务
- 2.3 清除线程事务信息
- 2.4 事务提交
- 2.5 事务异常处理
- 三、总结
一、基本概念
使用事务的目的是保证数据一致性和操作隔离。
1、ACID属性
- 原子性（Atomicity）：事务作为一个整体被执行，包含在其中的对数据库的操作要么全部被执行，要么都不执行。
- 一致性（Consistency）：事务应确保数据库的状态从一个一致状态转变为另一个一致状态。事务开始前后，数据库的完整性没有被破坏
- 隔离性（Isolation）：多个事务并发执行时，一个事务的执行不应影响其他事务的执行。
- 持久性（Durability）：已被提交的事务对数据库的修改应该永久保存在数据库中。
2、事务的隔离级别
-
读未提交（READ UNCOMMITTED）：另一个事务修改了数据，但尚未提交，而本事务中的SELECT会读到这些未被提交的数据（脏读）。
-
读已提交（READ COMMITTED）：本事务读取到的是最新的数据（其他事务提交后的）。问题是，在同一个事务里，前后两次相同的SELECT会读到不同的结果（不重复读）。
-
可重复读（REPEATABLE READ）：在同一个事务里，SELECT的结果是事务开始时时间点的状态，因此，同样的SELECT操作读到的结果会是一致的。但是，会有幻读现象。
-
串行化（SERIALIZABLE）：通过共享锁和排它锁实现，可以保证不同事务间的互斥，读读操作不会发生阻塞。
四个隔离级别逐渐增强，每一级解决一个问题，MySQL InnoDB的默认隔离级别是可重复读。
脏读和不重复读比较好理解，这里主要解释下什么是幻读？
所谓幻读就是一个事务在前后两次查询（当前读/更新）同一范围数据数据时，后一次查询看到了前一次查询没有看到的行。在下图的例子中，由于事务B的插入，导致事务A第一次查询时只有两条数据，而在第二次查询时出现了三条数据，就像之前读到的数据像幻觉一样。
3、事务行为
事务的开启：可用START TRANSACTION、BEGI命令显式开启事务
事务的提交：默认情况下，单条SQL执行成功后，MySQL会自动提交事务。开始事务后可以通过COMMIT命令提交；部分命令（DDL命令、管理数据库架构命令、管理命令）也会隐式执行COMMIT命令
事务的回滚：ROLLBACK命令
4、Spring事务的传播级别
| 传播级别 | 定义 |
|---|---|
| PROPAGATION_REQUIRED | 支持当前事务，如果当前没有事务，则新建一个事务 |
| PROPAGATION_SUPPORTS | 支持当前事务，如果当前没有事务，则以非事务进行 |
| PROPAGATION_MANDATORY | 支持当前事务，如果当前没有事务，则抛异常 |
| PROPAGATION_REQUIRES_NEW | 新建事务，如果当前存在事务，则把当前事务挂起 |
| PROPAGATION_NESTED | 如果当前存在事务，则在嵌套事务内执行。如果没有，则进行与PROPAGATION_REQUIRED类似操作 |
| PROPAGATION_NOT_SUPPORTED | 以非事务进行，如果当前存在事务，则挂起事务，执行当前逻辑，结束后恢复上下文的事务 |
| PROPAGATION_NEVER | 以非事务进行，如果当前存在事务，则抛异常 |
5、Spring事务支持方式
- 声明式事务
- @Transactional注解
- 编程式事务
- TransactionTemplate
- TransactionManager
编程式事务有两种方式：TransactionTemplate和TransactionManager
// TransactionTemplate
@Autowired
private TransactionTemplate transactionTemplate;
public void testTransaction() {
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
@Override
protected void doInTransactionWithoutResult(TransactionStatus transactionStatus) {
try {
// .... 业务代码
} catch (Exception e){
//回滚
transactionStatus.setRollbackOnly();
}
}
});
}
// TransactionManager
@Autowired
private PlatformTransactionManager transactionManager;
public void testTransaction() {
TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionDefinition());
try {
// .... 业务代码
transactionManager.commit(status);
} catch (Exception e) {
transactionManager.rollback(status);
}
}
我们常用的是声明式事务，Spring事务的核心流程在抽象类AbstractPlatformTransactionManager中，其基于模版方法的设计模式，事务的getTransaction、suspend、startTransaction、rollback、commit等都由各框架的事务管理器实现。下面主要结合**JDBC(DataSourceTransactionManager)**的事务实现进行Spring事务源码的分析。
二、Spring事务的执行源码
1、事务AOP
代码在执行@Transcational所注解的目标方法之前，会首先执行CglibAopProxy.DynamicAdvisedInterceptor#intercept方法，在这个方法中主要是寻找所执行目标方法的拦截器列表，也就是目标方法上对应注解的拦截器，如果有拦截器则先以链式的方式执行拦截器方法，否则执行目标方法。
对于@Transactional注解，其对应的拦截器为TransactionInterceptor。
2、事务处理拦截器TransactionInterceptor
TransactionInterceptor类是Advice实现类，用于对事务性的方法进行拦截，并通过Spring的事务管理器(PlatformTransactionManager)进行事务管理，PlatformTransactionManager有三个方法，通过 PlatformTransactionManager 这个接口，Spring 为各个平台如 JDBC(DataSourceTransactionManager)、Hibernate(HibernateTransactionManager)、JPA(JpaTransactionManager)等都提供了对应的事务管理器。
interface PlatformTransactionManager extends TransactionManager{
// 根据事务定义获取事务状态
TransactionStatus getTransaction(TransactionDefinition definition)
throws TransactionException;
// 提交事务
void commit(TransactionStatus status) throws TransactionException;
// 事务回滚
void rollback(TransactionStatus status) throws TransactionException;
}
2.1 主要流程
TransactionInterceptor的核心方法是invokeWithinTransaction，主要流程
源码如下：
@Override
@Nullable
public Object invoke(MethodInvocation invocation) throws Throwable {
// 获取目标类，即执行方法的类
Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
// 事务处理逻辑
return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
}
@Nullable
protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
final InvocationCallback invocation) throws Throwable {
// 获取事务属性源
// 该类有一个属性publicMethodsOnly，为true，限制了只有public方法，@Transcational注解才能生效
TransactionAttributeSource tas = getTransactionAttributeSource();
// 获取事务属性，包括传播级别、隔离级别、回滚规则等
final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
// 获取事务管理类，一般用的是JDBC的DataSourceTransactionManager
// 这里有本地缓存，只有第一次获取才会加载bean
final PlatformTransactionManager tm = determineTransactionManager(txAttr);
// 获取全路径方法名，用于监控和记录
final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
// 创建TransactionInfo（这里的TransactionInfo是TransactionAspectSupport的内部类）
// 这里会处理事务的传播级别，同时将事务和线程绑定
TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
Object retVal;
try {
// 执行目标方法
retVal = invocation.proceedWithInvocation();
}
catch (Throwable ex) {
// 目标方法执行异常，事务回滚
completeTransactionAfterThrowing(txInfo, ex);
throw ex;
}
finally {
// 重置线程的事务信息
cleanupTransactionInfo(txInfo);
}
// 事务提交
commitTransactionAfterReturning(txInfo);
return retVal;
}
else {
// 省略CallbackPreferringPlatformTransactionManager的处理
// 这部分是编程式事务的处理
......
}
}
2.2 尝试创建事务
根据给定的属性尝试创建一个事务，并将相关的事务信息TransactionStatus绑定到TransactionInfo返回，同时会将当前的TransactionInfo对象绑定到当前线程中，将该线程之前的事务信息记录到TransactionInfo的oldTransactionInfo属性中，也就是记录外层事务，可以形成不同方法之间的事务信息链表。
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
//如果没有名称，将方法全路径作为事务名称
if (txAttr != null && txAttr.getName() == null) {
txAttr = new DelegatingTransactionAttribute(txAttr) {
@Override
public String getName() {
return joinpointIdentification;
}
};
}
TransactionStatus status = null;
if (txAttr != null) {
if (tm != null) {
// 在该方法决定是否开启一个事务，
status = tm.getTransaction(txAttr);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
"] because no transaction manager has been configured");
}
}
}
// 通过各参数生成TransactionInfo，并记录线程中上个方法的TransactionInfo，将新TransactionInfo和线程绑定
return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
// 获取事务的数据源连接
// 设置是否允许保存点
Object transaction = doGetTransaction();
// Cache debug flag to avoid repeated checks.
boolean debugEnabled = logger.isDebugEnabled();
if (definition == null) {
// Use defaults if no transaction definition given.
definition = new DefaultTransactionDefinition();
}
if (isExistingTransaction(transaction)) {
// 如果事务对象的是否存在数据库连接且已经开启过事务，则说明存在外层事务
// 根据不同的事务传播级别进行处理
return handleExistingTransaction(definition, transaction, debugEnabled);
}
// 若不存在外层事务，开始创建新事务
// 判断是否超时
if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
}
// PROPAGATION_MANDATORY传播级别是：如果存在事务，则将当前方法加入到该事务中，如果不存在事务则当前方法抛出异常
// 这里由于不存在外层事务，所以直接抛出异常
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
throw new IllegalTransactionStateException(
"No existing transaction found for transaction marked with propagation 'mandatory'");
}
// 处理PROPAGATION_REQUIRED、PROPAGATION_REQUIRES_NEW、PROPAGATION_NESTED三种传播级别的事务
// 这三种传播级别的共同点是如果当前不存在事务，则创建一个新的事务
else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
// 挂起给定的事务
// 首先挂起事务同步器，清空TransactionSynchronizationManager中保存的当前线程的事务信息，返回被挂起的资源信息
// 核心方法是doSuspend，由不同子类实现，DataSourceTransactionManage的实现就是将当前线程的连接资源解绑
SuspendedResourcesHolder suspendedResources = suspend(null);
if (debugEnabled) {
logger.debug("Creating new transaction with name [" + definition.getName() + "]: " + definition);
}
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 | Error ex) {
resume(null, suspendedResources);
throw ex;
}
}
else {
// 处理PROPAGATION_SUPPORTS、PROPAGATION_NEVER、PROPAGATION_NOT_SUPPORTED三种传播级别的事务
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);
// 这里并没有真正开启一个事务，只创建了一个DefaultTransactionStatus对象用于记录和初始化事务同步器
return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null);
}
}
最核心的方法是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");
}
// 将新的连接信息保存到事务对象中
txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
}
// 将synchronizedWithTransaction设置为true，即资源标记为与事务同步
txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
con = txObject.getConnectionHolder().getConnection();
// 设置数据库连接的隔离级别和只读标志
Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
// 设置事务的隔离级别属性
txObject.setPreviousIsolationLevel(previousIsolationLevel);
// 将自动提交设置为手动提交
if (con.getAutoCommit()) {
// 设置mustRestoreAutoCommit为true，用于事务提交后回复自动提交
txObject.setMustRestoreAutoCommit(true);
if (logger.isDebugEnabled()) {
logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
}
con.setAutoCommit(false);
}
// 到这里事务已经开启，后续的sql需要等待commit命令执行后才会提交
// 对于只读事务，执行"SET TRANSACTION READ ONLY"优化
prepareTransactionalConnection(con, definition);
txObject.getConnectionHolder().setTransactionActive(true);
// 设置超时时间
int timeout = determineTimeout(definition);
if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
}
// 将连接资源和当前线程绑定
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);
}
}
2.3 清除线程事务信息
这里比较简单，本次事务结束后，将本次事务对象中的外层事务oldTransactionInfo重新和线程绑定
2.4 事务提交
目标方法执行完成后，则调用事务管理器的commit方法进行事务提交。但是以下两种情况是不进行提交而进行事务回滚的：
- 当前事务被标记为仅回滚，这里是一般是当前方法抛出异常后进行手动设置
- 当前事务被标记为全局回滚，这里一般是内层方法设置了仅回滚，且该内部方法的事务传播级别是不需要创建新的事务的（即PROPAGATION_REQUIRED、PROPAGATION_SUPPORTS、PROPAGATION_MANDATORY），内部方法在进行内部提交时会将整个连接资源设置为仅回滚，也就是全局回滚。
事务提交的核心方法是processCommit，源码如下：
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
try {
boolean beforeCompletionInvoked = false;
try {
boolean unexpectedRollback = false;
// 事务提交前的操作，空方法，由子类实现
prepareForCommit(status);
// 执行所有已注册的事务同步器的beforeCommit回调
triggerBeforeCommit(status);
// 执行所有已注册的事务同步器的beforeCompletion回调
triggerBeforeCompletion(status);
beforeCompletionInvoked = true;
// 存在保存点，则释放保存点，但是不提交事务，需要等待外层事务提交
// PROPAGATION_NESTED传播级别会开启保存点
if (status.hasSavepoint()) {
if (status.isDebug()) {
logger.debug("Releasing transaction savepoint");
}
unexpectedRollback = status.isGlobalRollbackOnly();
status.releaseHeldSavepoint();
}
// 如果是新的事务，则通过Connection的commit方法提交事务，具体由子类实现
// 对于PROPAGATION_REQUIRES_NEW或者最外层PROPAGATION_REQUIRED都是开启的新事务
else if (status.isNewTransaction()) {
if (status.isDebug()) {
logger.debug("Initiating transaction commit");
}
unexpectedRollback = status.isGlobalRollbackOnly();
doCommit(status);
}
// 剩余情况仅做rollbackOnly检查
else if (isFailEarlyOnGlobalRollbackOnly()) {
unexpectedRollback = status.isGlobalRollbackOnly();
}
// 有rollbackOnly标志，那么抛出UnexpectedRollbackException异常
if (unexpectedRollback) {
throw new UnexpectedRollbackException(
"Transaction silently rolled back because it has been marked as rollback-only");
}
}
catch (UnexpectedRollbackException ex) {
// 有rollback标志
// 执行所有已注册的事务同步器的afterCompletion回调
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
throw ex;
}
catch (TransactionException ex) {
// 执行doCommit发生异常
// 若设置了在doCommit执行异常时进行回滚，则需要进行回滚
if (isRollbackOnCommitFailure()) {
doRollbackOnCommitException(status, ex);
}
else {
// 执行所有已注册的事务同步器的afterCompletion回调
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
}
throw ex;
}
catch (RuntimeException | Error ex) {
// 如果还没有触发beforeCompletion方法回调，则进行回调
if (!beforeCompletionInvoked) {
triggerBeforeCompletion(status);
}
// 执行回滚，并执行所有已注册的事务同步器的afterCompletion回调
doRollbackOnCommitException(status, ex);
throw ex;
}
try {
// 事务成功提交后调用，触发afterCommit回调
triggerAfterCommit(status);
}
finally {
//最终触发afterCompletion方法回调
triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
}
}
finally {
// 事务提交后进行清理工作
//（1）如果开启了新的事务同步器，则清理事务同步器
//（2）如果该事务是新开启的，则恢复事务的自动提交，重置事务使用数据库连接的属性，释放数据库连接
//（3）如果存在被挂起的事务，则恢复挂起的事务
cleanupAfterCompletion(status);
}
}
在事务提交后的清理工作中，会恢复之前挂起的事务，那如何理解挂起和恢复？
事务的挂起实际上是开启内层事务时，将外层事务的连接和当前线程解绑，并存储在内层事务的属性中，然后将新事务所获取的连接资源与当前线程绑定，后续的操作都是基于这个新的连接，也就是一个新的事务，所以之前的事务和数据库连接就是被挂起了，并没有执行最后的事务提交或回滚；
事务的恢复就是内层事务提交或者回滚后，当前线程再次和之前保存的外层事务资源进行绑定，后续的处理都是基于外层事务的连接资源进行。
2.5 事务异常处理
当事务执行抛出异常时，会调用rollbackOn方法进行判断，当前出现的异常是否和配置的回滚异常匹配，若匹配则调用事务管理器的rollback方法进行回滚，否则调用事务管理器的commit方法进行事务提交。
对于rollbackOn的判断逻辑：
- 默认是只有RuntimeException或者Error类型的异常才会回滚
- 基于xml则是rollback-for和no-rollback-for属性
- 基于@Transactional注解则是rollbackFor、rollbackForClassName、noRollbackFor、noRollbackForClassName属性
事务提交在2.4中内容已经涉及，这里不再赘述。下面重点分析rollback方法，其核心方法是processRollback，源码如下：
private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
try {
boolean unexpectedRollback = unexpected;
try {
// 执行所有已注册的事务同步器的beforeCompletion回调
triggerBeforeCompletion(status);
// 如果存在保存点，则回滚到保存点并释放保存点
if (status.hasSavepoint()) {
if (status.isDebug()) {
logger.debug("Rolling back transaction to savepoint");
}
status.rollbackToHeldSavepoint();
}
// 如果是新开启的事务，则通过Connection的rollback进行回滚，具体由子类实现
else if (status.isNewTransaction()) {
if (status.isDebug()) {
logger.debug("Initiating transaction rollback");
}
doRollback(status);
}
// 其他情况
else {
// 若存在事务且事务被标记为仅回滚或者globalRollbackOnParticipationFailure属性为true
// globalRollbackOnParticipationFailure属性默认为true，表示只要你的参与事务失败了，就标记此事务为rollback-only
if (status.hasTransaction()) {
if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
if (status.isDebug()) {
logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
}
// 此处是将该事务的资源设置为仅回滚，这样即使没有抛出异常，外层事务在提交时也会进行回滚
// 具体见2.4节事务提交时进行回滚的情况
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) {
// 执行所有已注册的事务同步器的afterCompletion回调
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
throw ex;
}
// 执行所有已注册的事务同步器的afterCompletion回调
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 {
// 清除事务信息，同processCommit中的cleanupAfterCompletion方法
cleanupAfterCompletion(status);
}
}
三、总结
在我看来，对于spring事务的源码，搞清楚一下几个问题即可：
- 不同的传播级别Spring是怎么处理的？
- 在什么时候真正开启一个事务？
- Spring事务是如何通过“挂起”和“恢复”实现外层事务和内层事务的提交和回滚的？
另外，Spring事务的实现使用了AOP的逻辑实现，以及其采用了模板方法的设计模式，将核心处理流程交给各事务管理器实现，这种设计模式是非常值得我们学习的。
