前沿
一段生产事故发人深省，在Spring的声明式事务中手动捕获异常，居然判定回滚了，这是什么操作？话不多说直接上代码
@Service
public class A {
@Autowired
private B b;
@Autowired
private C c;
@Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.DEFAULT)
public void operate() {
try {
b.insertB();
c.insertC();
}catch (Exception e) {
e.printStackTrace();
}
}
}
@Service
public class B {
@Autowired
private BM bm;
@Transactional(propagation = Propagation.REQUIRED)
public int insertB() {
return bm.insert("B");
}
}
@Service
public class C {
@Autowired
private CM cm;
@Transactional(propagation = Propagation.REQUIRED)
public int insertC() {
return cm.insert("C");
}
}
问题阐述
好了大家都看到上面这段代码了，在正常的情况的我们会往B表和C表中各插入一条数据，那么当代码出现异常时又会怎么样呢？
我们现在假设B插入数据成功，但是C插入数据失败了，此时异常会上抛到A，被A中operate方法的try - cache所捕获，正常来说此时数据库中B能插入一条记录，而C表插入失败，这是我们期望的情况，但事实却不是，实际情况是B表没有插入数据，C表也没有插入数据，也就是说整个操作被Spring给回滚了
注意点
如果代码稍稍变动一下，将try - cache放在
insertC的代码块中，在同样的场景下，B中会成功插入一条记录
知识点前置条件
了解Spring的传播机制的可以直接跳过
我们先要搞清楚Spring中的
REQUIRED的作用REQUIRED：如果当前没有事务就创建一个新的事务，如果当前已经存在事务就加入到当前事务
也就是说当我们的传播机制同时为
REQUIRED时，A、B、C三者的事务是共用一个的，只有当A的流程全部走完时才会做一次commit或者rollback操作，不会在执行B或者C的过程中进行commit和rollback
问题追踪
好，有了一定的知识储备，我们一起来看源码
我们首先找到Spring事务的代理入口TransactionInterceptor, 当我们通过调用A类中的operate方法时会调用TransactionInterceptor的invoke方法，这是整个事务的入口，我们直接看重点invoke中的invokeWithinTransaction方法
//获取事务属性类 AnnotationTransactionAttributeSource
TransactionAttributeSource tas = getTransactionAttributeSource();
//获取事务属性
final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
//获取事务管理器
final TransactionManager tm = determineTransactionManager(txAttr);
PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
//获取joinpoint
final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
//注解事务会走这里
if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
// Standard transaction demarcation with getTransaction and commit/rollback calls.
//开启事务
TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
Object retVal;
try {
// This is an around advice: Invoke the next interceptor in the chain.
// This will normally result in a target object being invoked.
retVal = invocation.proceedWithInvocation();
} catch (Throwable ex) {
// target invocation exception
//事务回滚
completeTransactionAfterThrowing(txInfo, ex);
throw ex;
} finally {
cleanupTransactionInfo(txInfo);
}
if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
// Set rollback-only in case of Vavr failure matching our rollback rules...
TransactionStatus status = txInfo.getTransactionStatus();
if (status != null && txAttr != null) {
retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
}
}
//事务提交
commitTransactionAfterReturning(txInfo);
return retVal;
}
不重要的代码我已经省略了，好我们来看这个流程，上面这段代码很明显反应出了，当我们程序执行过程中抛出了异常时会调用到completeTransactionAfterThrowing的回滚操作，如果没有发生异常最终会调用事务提交commitTransactionAfterReturning， 我们来分析一下
正常情况是C发生异常，并且执行到了completeTransactionAfterThrowing事务回滚，但是因为不是新创建的事务，而是加入的事务所以并不会触发回滚操作，而在A中捕获了该异常，并且最终走到commitTransactionAfterReturning事务提交，事实是这样的吗？
事实上就是这样的，那就奇怪了，我明明提交了，怎么反而回滚了，我们继续往下看
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
// Use defaults if no transaction definition given.
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
//重点看.. DataSourceTransactionObject拿到对象
Object transaction = doGetTransaction();
boolean debugEnabled = logger.isDebugEnabled();
//第一次进来connectionHolder为空的, 所以不存在事务
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'");
}
//第一次进来大部分会走这里（传播属性是 Required ｜ Requested New ｜ 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 {
// 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);
}
}
这段代码是开启事务的代码，我们来看，当我们A第一次走进来的时候，此时是没有事务的，所以isExistingTransaction方法不成立，往下走，因为我们的传播机制是REQUIRED，所以我们会走到startTransaction方法中
private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction, boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
//创建一个新的事务状态, 注意这里的newTransaction 属性为true
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
//开启事务
doBegin(transaction, definition);
//开启事务后, 改变事务状态
prepareSynchronization(status, definition);
return status;
}
好这里我们只需要关注一个点那就是newTransactionStatus的第三个参数newTransaction，只有当我们新创建一个事务的时候才会为true， 这个属性很重要 ，我们后续还会用到它
好了，到这里第一次的事务开启就已经完成了，然后我们会调用业务逻辑，当调用insertB时，又会走到getTransaction，我们继续来看它，此时isExistingTransaction就可以拿到值了，因为A已经帮我们创建好了事务，此时会调用到handleExistingTransaction方法
//如果第二次进来还是PROPAFGATION_REQUIRED， 走这里, newTransation为false
return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
针对REQUIRED有用的代码就这一句，其他全部不用看，同样的我们看到第三个参数newTransaction，这里是false了，说明是加入了之前的事务，而不是自己新创建的，然后执行业务代码，最后走到commit，我们来看看commit中做了什么
//如果有回滚点
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();
}
它什么事情都没有做，为什么？因为我们的newTransaction不为true，所以当我们的代码在operate方法全部执行完以后才会走到这里
好接下来我们来看insertC，前面的流程都一模一样，我们直接看到回滚代码
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 {
// Participating in larger transaction
if (status.hasTransaction()) {
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 {
cleanupAfterCompletion(status);
}
}
我们的insertC方法同样它的newTransaction不是true，所以最终会走到doSetRollbackOnly，这个方法 重中之重 ，最后会调用这样一段代码
public void setRollbackOnly() {
this.rollbackOnly = true;
}
然后我们就要执行到我们的关键代码A中的operate的提交代码了
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;
}
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);
}
好了，看到这大家都明白了吧，在commit中，Spring会去判断defStatus.isGlobalRollbackOnly有没有抛出过异常被Spring所拦截，如果有，那么就不会执行commit操作，转而执行processRollback回滚操作
总结
在Spring的REQUIRED中，只要异常被Spring捕获到过，那么Spring最终就会回滚整个事务，即使自己在业务中已经捕获
所以我们回到最初的代码，如果我们希望Spring不进行回滚，那么我们只用将try-cache方法insertC方法中就可以，因为此时抛出的异常并不会被Spring所拦截到