spring--事务原理

Posted 2020-07-26 未来的那啥

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事务传播机制实例讲解