spring--事务原理

Posted 未来的那啥

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spring--事务原理相关的知识,希望对你有一定的参考价值。

Spring支持以下7种事务传播行为。 

传播行为

XML文件

propagation

含义

PROPAGATION_REQUIRED

 REQUIRED

表示当前方法必须在一个具有事务的上下文中运行。

如果当前没有事务,就新建一个事务;如果已经存在一个事务,就加入到这个事务中。

如果被调用端发生调用端需要回滚的异常,那么调用端和被调用端事务都将回滚;如果被调用端异常不是调用端需要回滚的,那么调用端终止执行,已执行操作正常提交

PROPAGATION_SUPPORTS

 SUPPORTS

表示当前方法不必需要具有一个事务上下文。

如果当前没有事务,就以非事务方式执行;但是如果有一个事务的话,它就加入到这个事务中运行。

PROPAGATION_MANDATORY

 MANDATORY

表示当前方法必须在一个事务中运行。

如果当前没有事务,就抛出异常;如果有事务,就加入到这个事务中。

PROPAGATION_REQUIRES_NEW REQUIRES_NEW

表示当前方法必须运行在它自己的事务中,新建事务,如果当前存在事务,把当前事务挂起,直到新的事务提交或者回滚才恢复执行。

(如果被调用端抛出rollback异常,则被调用端回滚,如果同时是调用端的rollback异常,调用端同时回滚)

PROPAGATION_NOT_SUPPORTED

NOT_SUPPORTED

表示该方法不应该在一个事务中运行,应该以非事务方式执行,每句sql马上提交。

如果当前存在事务,就把当前事务挂起。

(如果被调用端抛出调用端rollback异常,则调用端回滚)

PROPAGATION_NEVER

 NEVER

表示当前方法不应该在一个事务中运行,应该以非事务方式执行。

如果存在一个事务,则抛出异常

PROPAGATION_NESTED

 NESTED

如果当前没有事务,则执行与PROPAGATION_REQUIRED类似的操作;如果当前有一个事务,则该方法应该运行在一个嵌套事务中,被嵌套的事务可以独立于被封装的事务中进行提交或者回滚。

(如果封装事务存在,并且外层事务抛出异常回滚,那么内层事务必须回滚;反之,内层事务并不影响外层事务,如果内层抛出异常不需rollback,则忽略,如果内层抛出异常需要rollback,则回滚到外层的savepoint)

 

spring中配置事务时,往往这样

<tx:advice id="txAdvice" transaction-manager="transactionManager">
        <tx:attributes>
            <tx:method name="do*" read-only="false" rollback-for="java.lang.Exception"/>
            <tx:method name="*" propagation="SUPPORTS" read-only="true"/>
        </tx:attributes>
</tx:advice>

<aop:config>
        <aop:pointcut id="pc" expression="execution(* ffm.web.service.*.*(..))"/>
        <aop:advisor pointcut-ref="pc" advice-ref="txAdvice"/>
</aop:config>

但是是怎么起作用的呢?

tx是TransactionNameSpace,对应的是handler是TxNamespaceHandler,<tx:advice />最终解析出一个以TransactionInerceptor为classname的beandefinition并且注册这个bean,拦截方法如下:

public Object invoke(final MethodInvocation invocation) throws Throwable {
        Class targetClass = invocation.getThis() != null?AopUtils.getTargetClass(invocation.getThis()):null;
        return this.invokeWithinTransaction(invocation.getMethod(), targetClass, new InvocationCallback() {
            public Object proceedWithInvocation() throws Throwable {
                return invocation.proceed();
            }
        });
    }

1. 调用父类 TransactionAspectSupport的invokeWithinTransaction方法:

protected Object invokeWithinTransaction(Method method, Class<?> targetClass, final TransactionAspectSupport.InvocationCallback invocation) throws Throwable {
//事务隔离级别、事务传播策略、只读、回滚等属性信息
final TransactionAttribute txAttr = this.getTransactionAttributeSource().getTransactionAttribute(method, targetClass); final PlatformTransactionManager tm = this.determineTransactionManager(txAttr); final String joinpointIdentification = this.methodIdentification(method, targetClass); if(txAttr != null && tm instanceof CallbackPreferringPlatformTransactionManager) { try { Object ex1 = ((CallbackPreferringPlatformTransactionManager)tm).execute(txAttr, new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { TransactionAspectSupport.TransactionInfo txInfo = TransactionAspectSupport.this.prepareTransactionInfo(tm, txAttr, joinpointIdentification, status); TransactionAspectSupport.ThrowableHolder var4; try { Object ex = invocation.proceedWithInvocation(); return ex; } catch (Throwable var8) { if(txAttr.rollbackOn(var8)) { if(var8 instanceof RuntimeException) { throw (RuntimeException)var8; } throw new TransactionAspectSupport.ThrowableHolderException(var8); } var4 = new TransactionAspectSupport.ThrowableHolder(var8); } finally { TransactionAspectSupport.this.cleanupTransactionInfo(txInfo); } return var4; } }); if(ex1 instanceof TransactionAspectSupport.ThrowableHolder) { throw ((TransactionAspectSupport.ThrowableHolder)ex1).getThrowable(); } else { return ex1; } } catch (TransactionAspectSupport.ThrowableHolderException var14) { throw var14.getCause(); } } else {
       // step 1.1 TransactionAspectSupport.TransactionInfo ex
= this.createTransactionIfNecessary(tm, txAttr, joinpointIdentification); Object retVal = null; try { retVal = invocation.proceedWithInvocation(); } catch (Throwable var15) { this.completeTransactionAfterThrowing(ex, var15); throw var15; } finally {
          // step 1.2
this.cleanupTransactionInfo(ex); } this.commitTransactionAfterReturning(ex); return retVal; } }

1.1 关键方法createTransactionIfNecessary

protected TransactionAspectSupport.TransactionInfo createTransactionIfNecessary(PlatformTransactionManager tm, final TransactionAttribute txAttr, final String joinpointIdentification) {
        if(txAttr != null && ((TransactionAttribute)txAttr).getName() == null) {
            txAttr = new DelegatingTransactionAttribute((TransactionAttribute)txAttr) {
                public String getName() {
                    return joinpointIdentification;
                }
            };
        }

        TransactionStatus status = null;
        if(txAttr != null) {
            if(tm != null) {
          //step 1.1.1 调用org.springframework.jdbc.datasource.DataSourceTransactionManager status
= tm.getTransaction((TransactionDefinition)txAttr); } else if(this.logger.isDebugEnabled()) { this.logger.debug("Skipping transactional joinpoint [" + joinpointIdentification + "] because no transaction manager has been configured"); } }      //step 1.1.2绑定到当前线程 return this.prepareTransactionInfo(tm, (TransactionAttribute)txAttr, joinpointIdentification, status); }

1.1.1 调用AbstractPlatformTransactionManager的getTransaction

public final TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {
//step 1.1.1.1 Object transaction
= this.doGetTransaction(); boolean debugEnabled = this.logger.isDebugEnabled(); if(definition == null) { definition = new DefaultTransactionDefinition(); }
//step 1.1.1.2
if(this.isExistingTransaction(transaction)) { return this.handleExistingTransaction((TransactionDefinition)definition, transaction, debugEnabled); } else if(((TransactionDefinition)definition).getTimeout() < -1) { throw new InvalidTimeoutException("Invalid transaction timeout", ((TransactionDefinition)definition).getTimeout()); } else if(((TransactionDefinition)definition).getPropagationBehavior() == 2) { throw new IllegalTransactionStateException("No existing transaction found for transaction marked with propagation \\\'mandatory\\\'"); } else if(((TransactionDefinition)definition).getPropagationBehavior() != 0 && ((TransactionDefinition)definition).getPropagationBehavior() != 3 && ((TransactionDefinition)definition).getPropagationBehavior() != 6) { boolean newSynchronization1 = this.getTransactionSynchronization() == 0; return this.prepareTransactionStatus((TransactionDefinition)definition, (Object)null, true, newSynchronization1, debugEnabled, (Object)null); } else { AbstractPlatformTransactionManager.SuspendedResourcesHolder newSynchronization = this.suspend((Object)null); if(debugEnabled) { this.logger.debug("Creating new transaction with name [" + ((TransactionDefinition)definition).getName() + "]: " + definition); } try { boolean err = this.getTransactionSynchronization() != 2; DefaultTransactionStatus status = this.newTransactionStatus((TransactionDefinition)definition, transaction, true, err, debugEnabled, newSynchronization); this.doBegin(transaction, (TransactionDefinition)definition); this.prepareSynchronization(status, (TransactionDefinition)definition); return status; } catch (RuntimeException var7) { this.resume((Object)null, newSynchronization); throw var7; } catch (Error var8) { this.resume((Object)null, newSynchronization); throw var8; } } }

1.1.1.1调用 DataSourceTransactionManager doGetTransaction方法

protected Object doGetTransaction() {
        DataSourceTransactionManager.DataSourceTransactionObject txObject = new DataSourceTransactionManager.DataSourceTransactionObject(null);
        txObject.setSavepointAllowed(this.isNestedTransactionAllowed());
        ConnectionHolder conHolder = (ConnectionHolder)TransactionSynchronizationManager.getResource(this.dataSource);
        txObject.setConnectionHolder(conHolder, false);
        return txObject;
    }

生成一个DataSourceTransactionObject 事物对象,设置它的ConnectionHolder为当前线程上绑定的key=this.dataSource的ConnectionHolder(当然有可能为空)

1.1.1.2开始处理这个新的 DataSourceTransactionObject 事物对象

 DataSourceTransactionManager 的isExistingTransaction方法:

protected boolean isExistingTransaction(Object transaction) {
        DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
        return txObject.getConnectionHolder() != null && txObject.getConnectionHolder().isTransactionActive();
    }

如果返回true,

private TransactionStatus handleExistingTransaction(TransactionDefinition definition, Object transaction, boolean debugEnabled) throws TransactionException {
        if(definition.getPropagationBehavior() == 5) {
            throw new IllegalTransactionStateException("Existing transaction found for transaction marked with propagation \\\'never\\\'");
        } else {
            AbstractPlatformTransactionManager.SuspendedResourcesHolder newSynchronization3;
            boolean isoConstants2;
            if(definition.getPropagationBehavior() == 4) {
                if(debugEnabled) {
                    this.logger.debug("Suspending current transaction");
                }

                newSynchronization3 = this.suspend(transaction);
                isoConstants2 = this.getTransactionSynchronization() == 0;
                return this.prepareTransactionStatus(definition, (Object)null, false, isoConstants2, debugEnabled, newSynchronization3);
            } else if(definition.getPropagationBehavior() == 3) {
                if(debugEnabled) {
                    this.logger.debug("Suspending current transaction, creating new transaction with name [" + definition.getName() + "]");
                }

                newSynchronization3 = this.suspend(transaction);

                try {
                    isoConstants2 = this.getTransactionSynchronization() != 2;
                    DefaultTransactionStatus status = this.newTransactionStatus(definition, transaction, true, isoConstants2, debugEnabled, newSynchronization3);
                    this.doBegin(transaction, definition);
                    this.prepareSynchronization(status, definition);
                    return status;
                } catch (RuntimeException var7) {
                    this.resumeAfterBeginException(transaction, newSynchronization3, var7);
                    throw var7;
                } catch (Error var8) {
                    this.resumeAfterBeginException(transaction, newSynchronization3, var8);
                    throw var8;
                }
            } else {
                boolean newSynchronization1;
                if(definition.getPropagationBehavior() == 6) {
                    if(!this.isNestedTransactionAllowed()) {
                        throw new NestedTransactionNotSupportedException("Transaction manager does not allow nested transactions by default - specify \\\'nestedTransactionAllowed\\\' property with value \\\'true\\\'");
                    } else {
                        if(debugEnabled) {
                            this.logger.debug("Creating nested transaction with name [" + definition.getName() + "]");
                        }

                        if(this.useSavepointForNestedTransaction()) {
                            DefaultTransactionStatus newSynchronization2 = this.prepareTransactionStatus(definition, transaction, false, false, debugEnabled, (Object)null);
                            newSynchronization2.createAndHoldSavepoint();
                            return newSynchronization2;
                        } else {
                            newSynchronization1 = this.getTransactionSynchronization() != 2;
                            DefaultTransactionStatus isoConstants1 = this.newTransactionStatus(definition, transaction, true, newSynchronization1, debugEnabled, (Object)null);
                            this.doBegin(transaction, definition);
                            this.prepareSynchronization(isoConstants1, definition);
                            return isoConstants1;
                        }
                    }
                } else {
                    if(debugEnabled) {
                        this.logger.debug("Participating in existing transaction");
                    }

                    if(this.isValidateExistingTransaction()) {
                        if(definition.getIsolationLevel() != -1) {
                            Integer newSynchronization = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
                            if(newSynchronization == null || newSynchronization.intValue() != definition.getIsolationLevel()) {
                                Constants isoConstants = DefaultTransactionDefinition.constants;
                                throw new IllegalTransactionStateException("Participating transaction with definition [" + definition + "] specifies isolation level which is incompatible with existing transaction: " + (newSynchronization != null?isoConstants.toCode(newSynchronization, "ISOLATION_"):"(unknown)"));
                            }
                        }

                        if(!definition.isReadOnly() && TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
                            throw new IllegalTransactionStateException("Participating transaction with definition [" + definition + "] is not marked as read-only but existing transaction is");
                        }
                    }

                    newSynchronization1 = this.getTransactionSynchronization() != 2;
                    return this.prepareTransactionStatus(definition, transaction, false, newSynchronization1, debugEnabled, (Object)null);
                }
            }
        }
    }

初始下这个返回false,走下面的判断,根据传播机制

最后一个else有个关键方法doBegin,DataSourceTransactionManager 实现了这个抽象方法

protected void doBegin(Object transaction, TransactionDefinition definition) {
        DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
        Connection con = null;

        try {
            if(txObject.getConnectionHolder() == null || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
                Connection ex = this.dataSource.getConnection();
                if(this.logger.isDebugEnabled()) {
                    this.logger.debug("Acquired Connection [" + ex + "] for JDBC transaction");
                }

                txObject.setConnectionHolder(new ConnectionHolder(ex), true);
            }

            txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
            con = txObject.getConnectionHolder().getConnection();
            Integer ex1 = DataSourceUtils.prepareConnectionForTransaction(con, definition);
            txObject.setPreviousIsolationLevel(ex1);
            if(con.getAutoCommit()) {
                txObject.setMustRestoreAutoCommit(true);
                if(this.logger.isDebugEnabled()) {
                    this.logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
                }

                con.setAutoCommit(false);
            }

            txObject.getConnectionHolder().setTransactionActive(true);
            int timeout = this.determineTimeout(definition);
            if(timeout != -1) {
                txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
            }

            if(txObject.isNewConnectionHolder()) {
                TransactionSynchronizationManager.bindResource(this.getDataSource(), txObject.getConnectionHolder());
            }

        } catch (Throwable var7) {
            if(txObject.isNewConnectionHolder()) {
                DataSourceUtils.releaseConnection(con, this.dataSource);
                txObject.setConnectionHolder((ConnectionHolder)null, false);
            }

            throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", var7);
        }
    }

可以看出,这里给DataSourceTransactionObject 事物对象设置它的ConnectionHolder,真正打开连接激活事务,并把这个ConnectionHolder绑定在当前线程上(key=this.dataSource)

1.1.2 生成一个TransactionInfo,设置TransactionStatus为上一步1.1.1得到的TransactionStatus,并把自己绑定 到当前线程

protected TransactionAspectSupport.TransactionInfo prepareTransactionInfo(PlatformTransactionManager tm, TransactionAttribute txAttr, String joinpointIdentification, TransactionStatus status) {
        TransactionAspectSupport.TransactionInfo txInfo = new TransactionAspectSupport.TransactionInfo(tm, txAttr, joinpointIdentification);
        if(txAttr != null) {
            if(this.logger.isTraceEnabled()) {
                this.logger.trace("Getting transaction for [" + txInfo.getJoinpointIdentification() + "]");
            }

            txInfo.newTransactionStatus(status);
        } else if(this.logger.isTraceEnabled()) {
            this.logger.trace("Don\\\'t need to create transaction for [" + joinpointIdentification + "]: This method isn\\\'t transactional.");
        }

        txInfo.bindToThread();
        return txInfo;
    }
TransactionInfo的bindToThread方法:
private void bindToThread() {
            this.oldTransactionInfo = (TransactionAspectSupport.TransactionInfo)TransactionAspectSupport.transactionInfoHolder.get();
            TransactionAspectSupport.transactionInfoHolder.set(this);
        }

1.2 重置当前线程的TransactionInfo为oldTransactionInfo 

private void restoreThreadLocalStatus() {
TransactionAspectSupport.transactionInfoHolder.set(this.oldTransactionInfo);
}

以常用的propagation="SUPPORTS" / "REQUIRED"为例,分析跟踪事务的过程

  • SUPPORTS->REQUIRED,`一路到1.1.1.2,isExistingTransaction返回false,进入
if(((TransactionDefinition)definition).getPropagationBehavior() != 0 && ((TransactionDefinition)definition).getPropagationBehavior() != 3 && ((TransactionDefinition)definition).getPropagationBehavior() != 6) {
            boolean newSynchronization1 = this.getTransactionSynchronization() == 0;
            return this.prepareTransactionStatus((TransactionDefinition)definition, (Object)null, true, newSynchronization1, debugEnabled, (Object)null);

以非事务方式运行,接着调用REQUIRED的方法,直到1.1.1.2,isExistingTransaction返回false,进入最后一个else,打开新事务

  • REQUIRED->SUPPORTS,到1.1.1.2时,isExistingTransaction返回false,进入最后一个else,打开新事务。接着调用SUPPORTS的方法,到达1.1.1.2时发现有事务,进入handleExistingTransaction处理,最终加入已有事务

 

 

参考文章:

1. 揭开Spring事务处理

2. Spring中的事务控制学习中(转)

3. Spring 事务管理高级应用难点剖析--转

4. spring源码分析之——spring 事务管理实现方式

5. Spring @Transactional 如何开启事务

6. 详解spring事务属性

7. Spring事务处理的实现

8. spring事务传播机制实例讲解

 

以上是关于spring--事务原理的主要内容,如果未能解决你的问题,请参考以下文章

Spring事务传播行为

Spring事务的传播行为

Spring事务传播行为详解

18个示例详解 Spring 事务传播机制

spring 事务管理之事务传播行为之实践NOT_SUPPORTED(五)

Spring中事务传播行为