亲宝软件园·资讯

展开

Spring事务捕获异常后回滚

Lester_Tai 人气:0

前沿

一段生产事故发人深省,在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所拦截到

加载全部内容

相关教程
猜你喜欢
用户评论