在上一篇回顾了一下传统的JDBC事务，并分析了Spring解析事务标签，创建事务代理对象的源码，本篇我们来分析在目标方法执行的前后，事务增强器是如何工作的。
1.相关接口
在分析事务增强器的源码之前，先来过一些接口，明确接口的作用和接口之间的关系。
- java jdbc规范接口：DataSource、 Connection
- 事务属性承载对象：TransactionAttribute
- 事务管理器：TransactionManager、PlatformTransactionManager
- 数据库链接holder：ConnectionHolder
- Spring抽象出来的事务对象：JdbcTransactionObjectSupport、DataSourceTransactionObject
- Spring抽象出来的事务状态（包装“事务对象”，做了一些增强）：AbstractTransactionStatus、DefaultTransactionStatus
- Spring抽象出来的事务信息（集大成者“txMgr”、“txStatus”、“txAttr”…）：TransactionInfo
2.事务增强器
接下来我们来分析事务增强器源码。
事务增强器的入口自然是**invoke()**。
/*** 事务增强器的入口* invocation：后续事务增强器向后驱动事务拦截器的时候还需要使用它* @param invocation the method invocation joinpoint* @return* @throws Throwable*/@Override@Nullablepublic Object invoke(MethodInvocation invocation) throws Throwable {// Work out the target class: may be {@code null}.// The TransactionAttributeSource should be passed the target class// as well as the method, which may be from an interface./*需要被事务增强器增强的目标类型* invocation.getThis() 拿到目标对象* */Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);// 参数一：目标方法//参数二：目标对象类型//参数三：return invokeWithinTransaction(invocation.getMethod(), targetClass, new CoroutinesInvocationCallback() {@Override@Nullablepublic Object proceedWithInvocation() throws Throwable {return invocation.proceed();}@Overridepublic Object getTarget() {return invocation.getThis();}@Overridepublic Object[] getArguments() {return invocation.getArguments();}});}
我们直接往下看**invokeWithinTransaction()**。
{
Object result;
final ThrowableHolder throwableHolder = new ThrowableHolder();
// It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
try {
result = ((CallbackPreferringPlatformTransactionManager) ptm).execute(txAttr, status -> {
TransactionInfo txInfo = prepareTransactionInfo(ptm, txAttr, joinpointIdentification, status);
try {
Object retVal = invocation.proceedWithInvocation();
if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
// Set rollback-only in case of Vavr failure matching our rollback rules...
retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
}
return retVal;
}
catch (Throwable ex) {
if (txAttr.rollbackOn(ex)) {
// A RuntimeException: will lead to a rollback.
if (ex instanceof RuntimeException) {
throw (RuntimeException) ex;
}
else {
throw new ThrowableHolderException(ex);
}
}
else {
// A normal return value: will lead to a commit.
throwableHolder.throwable = ex;
return null;
}
}
finally {
cleanupTransactionInfo(txInfo);
}
});
}
catch (ThrowableHolderException ex) {
throw ex.getCause();
}
catch (TransactionSystemException ex2) {
if (throwableHolder.throwable != null) {
logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
ex2.initApplicationException(throwableHolder.throwable);
}
throw ex2;
}
catch (Throwable ex2) {
if (throwableHolder.throwable != null) {
logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
}
throw ex2;
}
// Check result state: It might indicate a Throwable to rethrow.
if (throwableHolder.throwable != null) {
throw throwableHolder.throwable;
}
return result;
}
}
这个方法的内容很多：
- 获取事务注解解析器，解析事务注解
- 获取事务名称
判断如果是声明式事务
**createTransactionIfNecessary()**创建一个最上层的事务上下文，包含所有的事务资源- 驱动方法增强逻辑继续往下执行
**completeTransactionAfterThrowing(txInfo, ex)**执行业务代码出现异常时的逻辑**cleanupTransactionInfo(txInfo)**还原现场逻辑**commitTransactionAfterReturning(txInfo)**提交事务逻辑
判断如果是编程式事务，走编程式事务的逻辑….
接下来我们来分析这几个核心逻辑。
3.创建最上层的事务上下文
这里是事务的核心逻辑，涉及到事务嵌套和传播行为，隔离级别。
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
// If no name specified, apply method identification as transaction name.
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.Transaction 有可能为空
//方法上的注解为 @Transactional上的注解 传播行为设置为了 NOT_SUPPORTED || NEVER
//具体看一下这个方法
status = tm.getTransaction(txAttr);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
"] because no transaction manager has been configured");
}
}
}
//包装成一个上层的事务上下文对象
return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}
- 获取事务状态对象
- 包装成事务上下文对象
主要的逻辑在这里**getTransaction()**。
�
@Override
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
throws TransactionException {
// 事务属性信息
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
//获取事务对象，非常关键
Object transaction = doGetTransaction();
boolean debugEnabled = logger.isDebugEnabled();
//条件成立说明当前是重入的事务的情况
//a开启事务 a调用b ，b也加了事务注解 ，开启了事务的情况
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'");
}
//PROPAGATION_REQUIRED PROPAGATION_REQUIRES_NEW PROPAGATION_NESTED 则进入
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 {
//走到这里是什么情况？ support || not support || never
//没有使用新的事务
// 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_MANDATORY，有就使用当前事务，没有就抛异常
- PROPAGATION_REQUIRED PROPAGATION_REQUIRES_NEW PROPAGATION_NESTED 开启一个新的事物
- support || not support || never 没有使用新的事务，不会主动开启
- 最终构建事务状态对象
**doGetTransaction() 获取事务对象。**
**handleExistingTransaction()处理事务重入的逻辑。**
**startTransaction()开启新事物的逻辑。**
3.1 获取事务对象
@Override
protected Object doGetTransaction() {
//先创建事务对象
DataSourceTransactionObject txObject = new DataSourceTransactionObject();
//事务的保存点，这个由事务管理器控制
txObject.setSavepointAllowed(isNestedTransactionAllowed());
//TransactionSynchronizationManager 事务同步管理器
//从tl中获取连接资源，有可能拿到null，也有可能不是null
//什么时候是null，什么时候不是null？
//==null:事务方法a调用了非事务方法b
//!=null:事务方法a调用了事务方法b
ConnectionHolder conHolder =
(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
//为事务对象赋能，参数二传递的false， 表示当前事务是否新分配了连接资源，而不是和上层事务共享，默认是false，表示共享。
txObject.setConnectionHolder(conHolder, false);
return txObject;
}
3.2处理事务重入
private TransactionStatus handleExistingTransaction(
TransactionDefinition definition, /*事务属性*/Object transaction/*事务对象*/, boolean debugEnabled)
throws TransactionException {
/*
进入这个方法的时候说明当前线程已经持有一个事务了，需要根据新方法的事务注解传播行为，走不同的逻辑
*/
//PROPAGATION_NEVER：需要抛异常
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
throw new IllegalTransactionStateException(
"Existing transaction found for transaction marked with propagation 'never'");
}
//PROPAGATION_NOT_SUPPORTED：有事务就挂起当前事务
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
if (debugEnabled) {
logger.debug("Suspending current transaction");
}
//看一下挂起的逻辑
Object suspendedResources = suspend(transaction);
boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
//将挂起事务返回的 挂起持有者对象(这里面持有上一个事务对象的连接资源和线程上下文参数) 给事务状态对象赋能
//step into 创建一个新的 事务状态对象 ，这里的第二个参数 事务 为null ；表示线程执行到当前方法执行到事务增强的后置处理逻辑的时候
//提交事务的时候会检查事务状态的事务是否有值，如果没有值，Spring就不会做提交的操作。
//参数6：suspendedResources 线程执行到后置处理的逻辑的时候，执行到恢复现场的时候会检查这个参数是否有值，如果有值会进行恢复现场的操作。
return prepareTransactionStatus(
definition, null/*说明当前线程未手动开启事务，连接是直接从数据源拿的，不需要手动提交事务了*/, false, newSynchronization, debugEnabled, suspendedResources);
}
//PROPAGATION_REQUIRES_NEW：挂起当前事务，开启一个新的事务
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 {
//开启一个专属于当前方法的新事务，因为当前方法被挂起的事务执行完当前方法后还要回到上层继续执行，所以
//suspendedResources用来恢复现场
return startTransaction(definition, transaction, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error beginEx) {
resumeAfterBeginException(transaction, suspendedResources, beginEx);
throw beginEx;
}
}
//嵌套事务的逻辑，如果当前存在事务，则在嵌套事务内执行 ，
//如果当前没有事务，则与required的操作类似
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
//默认情况下，spring是不开启这种传播行为的，除非手动开启，所以默认是false。
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() + "]");
}
//一般成立，默认是true
if (useSavepointForNestedTransaction()) {
// 为当前方法创建一个 事务状态对象，共享的上层事务，执行的扩展点是外层的，也不需要挂起事务。
DefaultTransactionStatus status =
prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
//创建一个保存点 ，重要！！！
status.createAndHoldSavepoint();
return status;
}
else {
// 开启新事物
return startTransaction(definition, transaction, debugEnabled, null);
}
}
if (debugEnabled) {
logger.debug("Participating in existing transaction");
}
//是否需要验证，默认是false
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");
}
}
}
/*
执行到这里就剩下 required & supports
*/
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
//prepareTransactionStatus（）上面已经分析过了，属于构造器重载逻辑
//参数二：事务对象 -> connHodler 是 doGetTransaction()的时候从西安城上下文内获取的上层事务的连接资源
//参数六：是空，因为我们没有挂起任何事务
return prepareTransactionStatus(definition, transaction/*事务也是使用上层的*/, false/*表示并不是一个为自己创建的事务，与上层方法共享*/, newSynchronization, debugEnabled, null);
}
- PROPAGATION_NEVER：需要抛异常
- PROPAGATION_NOT_SUPPORTED：有事务就挂起当前事务
- PROPAGATION_REQUIRES_NEW：挂起当前事务，开启一个新的事务
- 嵌套事务的逻辑，如果当前存在事务，则在嵌套事务内执行 ，如果当前没有事务，则与required的操作类似
- required & supports，不会创建事务对象
- 最终创建事务状态对象并返回
看一下创建事务保存点的逻辑，**createAndHoldSavepoint()**。
public void createAndHoldSavepoint() throws TransactionException {
/*
getSavepointManager()：这里实际上就执行到了jdbc的方法，创建保存点，然后保存。
*/
setSavepoint(getSavepointManager().createSavepoint());
}
3.3 开启新事物
private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
//一般情况下是true ，这个值控制是否执行事务的扩展逻辑，这东西有点类似ioc的后置处理器
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
//创建默认的事务状态对象，第三个参数为true，会为当前事务分配 连接资源，就是事务是专门为了当前方法开启的
//suspendedResources:表示挂起的事务 ，从上面过来这里实际上是null。
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
//开启事务 核心逻辑 ，参数一：事务对象；参数二：事务属性
doBegin(transaction, definition);
//处理TransactionSynchronization
prepareSynchronization(status, definition);
return status;
}
核心逻辑在这里**doBegin()**。
@Override
protected void doBegin(Object transaction, TransactionDefinition definition) {
//事务对象
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
Connection con = null;
try {/*判断当前事务对象是不是有线程资源，没有就会走if的逻辑，有就说明需要为当前方法分为连接资源*/
if (!txObject.hasConnectionHolder() ||
txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
//通过数据源拿到真实的数据库连接
Connection newCon = obtainDataSource().getConnection();
if (logger.isDebugEnabled()) {
logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
}
//重点：将上一步创建的数据库连接包装为ConnectionHolder，并且为事务对象赋能
//参数二很关键，给事务新申请的而连接资源，那么就将事务对象的 newConnectionHolder 设置为true。表示当前目标方法开启了一个自己的事务。
txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
}
txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
//获取事务对象商的数据库连接
con = txObject.getConnectionHolder().getConnection();
//修改数据库连接上的一些属性 step into
Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
//将连接原来的隔离级别保存到事务对象，方便释放连接的时候，设置回原来的状态
txObject.setPreviousIsolationLevel(previousIsolationLevel);
txObject.setReadOnly(definition.isReadOnly());
//如果连接的自动提交是true，一般会成立
if (con.getAutoCommit()) {
//因为接下来就是设置自动提交为false，这里设置 must ，表示回头释放的时候要设置回去
txObject.setMustRestoreAutoCommit(true);
if (logger.isDebugEnabled()) {
logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
}
//源头，相当于在数据库开启了事务
con.setAutoCommit(false);
}
//没啥实际的东西
prepareTransactionalConnection(con, definition);
//激活holder的事务状态
txObject.getConnectionHolder().setTransactionActive(true);
//获取超时时间
int timeout = determineTimeout(definition);
//如果时间不相等，就设置
if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
}
// 如果是新开启的事务，分配了新的连接就会成立，这个时候需要将线程和连接进行一个绑定 tl
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);
}
}
- 判断当前事务对象是不是有线程资源，没有就会走if的逻辑，有就说明需要为当前方法分为连接资源
- 通过数据源拿到真实的数据库连接
- 将上一步创建的数据库连接包装为ConnectionHolder，并且为事务对象赋能
- 获取事务对象商的数据库连接
- 修改数据库连接上的一些属性
- 将连接原来的隔离级别保存到事务对象，方便释放连接的时候，设置回原来的状态
- 如果连接的自动提交是true，改成false
- 激活holder的事务状态
- 设置超时时间
如果是新开启的事务，分配了新的连接就会成立，这个时候需要将线程和连接进行一个绑定 tl
4.异常回滚逻辑
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); } //条件一：一般都是成立的 //条件二：transactionAttribute.rollbackOn(ex) 判断目标方法抛出的异常是否需要回滚，条件成立，说明需要回滚。 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; } } } }判断目标方法的异常是否需要回滚？
需要，
**rollback()**。- 不需要，
**commit()**。
来看一下**rollback()**的逻辑。
@Override
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, false);
}
继续往下看**processRollback()**。
private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
try {
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 {
//执行到这里说明，这个事务不是当前方法开启的 （共享上层事务）|| 当前方法压根没开启事务（not_supports，never, supports）
// Participating in larger transaction
//说的是第一种情况：当前方法共享上层事务
if (status.hasTransaction()) {
//条件一：什么时候成立？当前方法共享上层事务，业务代码强制设置当前整个事务 需要回滚的话，可以通过 设置 status.isLocalRollbackOnly() = true
//条件二：默认是true
if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
if (status.isDebug()) {
logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
}
/*
一个共享上层事务的方法可以直接回滚嘛？不行的，需要将回滚的操作交给上层方法来做。
如何交给？设置status.isLocalRollbackOnly()=true，这样的话，线程回到上层事务提交逻辑的时候，会检查该字段，发现是true，就会执行回滚逻辑
*/
//这里其实就是设置回滚字段为true 先拿到事务状态对象-> 事务对象 -> 连接持有者 -> 设置rollbackOnly =true。
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 {
//看这里
cleanupAfterCompletion(status);
}
}
关注两个核心的方法**doRollback() & rollbackToHeldSavepoint()**。
public void rollbackToHeldSavepoint() throws TransactionException {
//获取事务的保存点
Object savepoint = getSavepoint();
if (savepoint == null) {
throw new TransactionUsageException(
"Cannot roll back to savepoint - no savepoint associated with current transaction");
}
//回滚到保存点
getSavepointManager().rollbackToSavepoint(savepoint);
//删除保存点
getSavepointManager().releaseSavepoint(savepoint);
//清空保存点
setSavepoint(null);
}
@Override
protected void doRollback(DefaultTransactionStatus status) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
Connection con = txObject.getConnectionHolder().getConnection();
if (status.isDebug()) {
logger.debug("Rolling back JDBC transaction on Connection [" + con + "]");
}
try {
con.rollback();
}
catch (SQLException ex) {
throw translateException("JDBC rollback", ex);
}
}
5.还原现场逻辑
protected void cleanupTransactionInfo(@Nullable TransactionInfo txInfo) {
if (txInfo != null) {
//往下看
txInfo.restoreThreadLocalStatus();
}
}
我们往下看
private void restoreThreadLocalStatus() {
// Use stack to restore old transaction TransactionInfo.
// Will be null if none was set.
//相当于事务出栈的逻辑
transactionInfoHolder.set(this.oldTransactionInfo);
}
这里很简单，就是让当前事务出栈。将栈内的事务设置成上一层事务。
6.提交事务逻辑
protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) {
if (txInfo != null && txInfo.getTransactionStatus() != null) {
if (logger.isTraceEnabled()) {
logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "]");
}
//调用事务管理器的提交事务方法
txInfo.getTransactionManager().commit(txInfo.getTransactionStatus()/*事务状态*/);
}
}
调用了事务管理器的提交事务方法。
@Override
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;
}
//shouldCommitOnGlobalRollbackOnly() 默认是true
//defStatus.isGlobalRollbackOnly() 其实就是defStatus ->txObject->connHolder->rollbackOnly 字段
//下层事务使用上层事务的时候，想回滚，就会设置这个标记
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);
}
核心的逻辑在**processCommit()**。
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);
}
}
逻辑很简单，清理保存点，提交事务，执行扩展点逻辑，清理资源。
我们来看一下资源的清理**cleanupAfterCompletion(status)**。
private void cleanupAfterCompletion(DefaultTransactionStatus status) {
//设置当前方法的事务状态为完成状态
status.setCompleted();
//清理逻辑 线程上下文变量 & 扩展点注册的东西
if (status.isNewSynchronization()) {
TransactionSynchronizationManager.clear();
}
//如果是开启的新事物，还原现场操作 最重要的就是解绑线程持有的连接
if (status.isNewTransaction()) {
doCleanupAfterCompletion(status.getTransaction());
}
//说明当前事务执行的时候，挂起了一个上层的事务
if (status.getSuspendedResources() != null) {
if (status.isDebug()) {
logger.debug("Resuming suspended transaction after completion of inner transaction");
}
Object transaction = (status.hasTransaction() ? status.getTransaction() : null);
//唤醒上层的事务
resume(transaction, (SuspendedResourcesHolder) status.getSuspendedResources());
}
}
�看几个核心方法**doCleanupAfterCompletion()** & **resume()**。
@Override
protected void doCleanupAfterCompletion(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
// Remove the connection holder from the thread, if exposed.
if (txObject.isNewConnectionHolder()) {
TransactionSynchronizationManager.unbindResource(obtainDataSource());
}
// Reset connection.
Connection con = txObject.getConnectionHolder().getConnection();
try {
if (txObject.isMustRestoreAutoCommit()) {
con.setAutoCommit(true);
}
DataSourceUtils.resetConnectionAfterTransaction(
con, txObject.getPreviousIsolationLevel(), txObject.isReadOnly());
}
catch (Throwable ex) {
logger.debug("Could not reset JDBC Connection after transaction", ex);
}
if (txObject.isNewConnectionHolder()) {
if (logger.isDebugEnabled()) {
logger.debug("Releasing JDBC Connection [" + con + "] after transaction");
}
DataSourceUtils.releaseConnection(con, this.dataSource);
}
txObject.getConnectionHolder().clear();
}
protected final void resume(@Nullable Object transaction, @Nullable 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);
}
}
}
**doResume()**
@Override
protected void doResume(@Nullable Object transaction, Object suspendedResources) {
TransactionSynchronizationManager.bindResource(obtainDataSource(), suspendedResources);
}
补充一个逻辑**doSuspend()**。挂起逻辑
@Override
protected Object doSuspend(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
//将事务对象的 连接持有者设置为null，不想和上层事务共享连接资源...
//当前方法有可能是不开启事务 || 要开启一个独立的事务
txObject.setConnectionHolder(null);
//解绑线程上的事务，将连接持有者从tl移除掉，这样业务就不会再拿上层事务的连接资源了
return TransactionSynchronizationManager.unbindResource(obtainDataSource());
}
7.思考与沉淀
事务的源码有一些琐碎，至此整个事务的源码其实已经分析完了，为了形成一个清晰，完整的链路，我画了一张事务源码的流程图。
我们再来回顾一下事务的流程。 @EnableTransactionManagement
利用TransactionManagementConfigurationSelector给容器中会导入组件
给容器中导入了两个类：
1.AutoProxyRegistrar
往容器中导入了一个组件：InfrastructureAdvisorAutoProxyCreator
这个类是什么？利用后置处理器，包装对象，返回一个代理对象（增强器），代理对象执行方法的时候利用拦截器进行调用
2.ProxyTransactionManagementConfiguration
给容器中注册了三个bean：
1.事务增强器 ：advisor切面
2.事务注解解析器：解析事务注解，获取注解信息
3.事务增强器的拦截器：拦截事务相关的advisor切面
ctrl + H 查看类的继承关系：这个类是MethodInterceptor的子类
在目标方法执行的时候：
执行拦截器链条
事务拦截器：
1.先获取事务相关的属性
2.在获取事务管理器，如果事先没有添加和执行任何事务管理器，最终会从容器中拿出来一个默认的
3.执行目标方法
1.如果是异常，获取到事务管理器，利用事务管理器进行回滚
2.如果是正常执行，利用事务管理器，提交事务。
事务注解解析器和事务增强器的拦截器都包含在事务增强器中，为事务增强器赋能。 �
至此，事务源码分析完成，后续的篇章，我们会去分析web请求的相关流程，我是二十，熟读Java生态圈源码，精通Java高并发编程，欢迎点赞关注。
若有收获，就点个赞吧
0 人点赞
