spring技术内幕15-Spring事务处理的实现

1、编程式事务处理：

Spring提供编程式和声明式两种事务处理方式，我们首先通过一个编程时事务的小列子了解Spring中编程式事务处理的基本实现：

//通过DefaultTransactionDefinition对象来持有事务处理属性

TransactionDefinition td = new DefaultTransactionDefinition();

//获取事务的状态

TransactionStatus status = transactionManager.getTransaction(td);

try{

//调用需要进行事务处理的目标方法

}catch(ApplicationException e){

//调用目标方法过程中产生异常，则对事务进行回滚处理

transactionManager.rollrack(status);

throw e;

}

//成功调用目标方法之后，对事务进行提交处理

transactionManager.commit(status);

Spring的声明式事务处理的即开即用特性为用户提供了很大的方便，在使用Spring时，我们绝大数情况下还是使用其声明式事务处理。声明式事务处理涉及Spring框架对事务处理的统一管理，已经对并发事务和事务属性的处理，是一个比较复杂的过程，下面我们通过源码分析，了解Spring框架声明式事务处理功能的具体实现。

2、事务的创建：

上篇博客对Spring事务拦截器TransactionInterceptor回调方法invoke的源码分析中，我们了解到在进行事务处理前，首先根据是否是CallbackPreferringPlatformTransactionManager类型的事务处理器分包通过下面两个方法创建事务信息对象：

TransactionInfo txInfo = createTransactionIfNecessary(tm,txAttr,joinpointIdentification);

TransactionInfo txInfo = prepareTransactionInfo(tm,txAttr,joinpointIdentification,status);

事务拦截器TransactionInterceptor回调方法invoke通过调用TransactionAspectSupport事务切面支持类中的createTransactionIfNecessary和prepareTransactionInfo方法创建事务对象：

//根据给定的事务属性创建事务对象

protected TransactionIfo createTransactionIfNecessary(PlatformTransactionManager tm,TransactionAttribute txAttr,final String joinpointIdentification){

//读取事务方法调用的事务配置属性

if(txAttr != null && txAttr.getName() == null){

//如果事务名称为null，则使用方法的名称（事务连接标识）作为事务名称，调用一个实现DelegatingTransactionAttribute接口的匿名内部类

txAttr = new DelegatingTransactionAttribute(txAttr){

//使用方法名称作为事务名称

public String getName(){

return joinpointIdentification;

}

};

}

//事务状态封装了事务执行的状态信息

TransactionStatus status = nulll;

if(txAttr != null){

if(tm != null){

//事务处理器创建事务，并且返回当前事务的状态信息

status = tm.getTransaction(txAttr);

}

else {

if(logger.isDebugEnabled()){

logger.debug("Skipping transactional joinpoint [" + joinpointIdentification + "] because no transaction manager has been configured");

}

}

}

//准备事务信息，事务信息TransactionInfo封装了事务配置和状态信息

return prepareTransactionInfo(tm,txAttr,joinpointIdentification,status);

}

//准备事务信息

protected TransactionInfo prepareTransactionInfo(PlatformTransactionManager tm,TransactionAttribute txAttr,String joinpointIdentification,TransactionStatus status){

//创建事务信息对象

TransactionInfo txInfo = new TransactionInfo(tm,txAttr,joinpointIdentification);

//如果事务属性不为null，需要为方法使用事务

if(txAttr != null){

if(logger.isTraceEnabled()){

logger.trace("Getting transaction for [" + txInfo.getJoinpointIdentification() + "]");

}

//为事务信息对象设置事务状态

txInfo.newTransactionStatus(status);

}

//如果事务属性为null，不需要为方法使用事务

else{

if(logger.isTraceEnabled()){

logger.trace("Don't need to create transaction for [" + joinpointIdentification + "]: This method isn't transactional.");

}

//把当前创建的事务信息对象和线程绑定

txInfor.bindToThread();

return txInfo;

}

通过上面对TransactionAspectSupport事务切面支持类创建事务信息对象的源码分析，我们了解了在创建事务信息对象过程中创建事务状态，将创建的事务信息对象和当前线程资源绑定等基本的处理流程。在创建事务信息对象的方法中，事务处理器的tm.getTransaction(txAttr);是真正底层创建事务对象的方法，下面我们继续分析事务处理器创建事务对象的过程。

3、抽象事务管理器AbstractPlatformTransactionManager获取事务：

抽象事务管理器AbstractPlatformTransactionManager提供了创建事务的模板，这个模板会被具体的事务处理器所使用，抽象事务管理器根据事务属性配置和当前线程绑定信息对事务是否需要创建以及如何创建进行一些通用的处理，然后把事务创建的底层细节交给具体的事务处理器实现。抽象事务管理器创建事务的模板方法如下：

public final TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {

//doGetTransaction()方法是抽象方法，具体的实现由具体的事务处理器提供

Object transaction = doGetTransaction();

boolean debugEnabled = logger.isDebugEnabled();

//如果没有配置事务属性，则使用默认的事务属性

if(definition == null){

definition = new DefaultTransactionDefinition();

}

//检查当前线程是否存在事务

if(isExistingTransaction(transaction)){

//处理已经存在的事务

return handleExistingTransaction(definition,transaction,debugEnabled);

}

//检查事务属性中timeout超时属性设置是否合理

if(definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT){

throw new InvalidTimeoutException("Invalid transaction timeout",definition.getTimeout());

}

//对事务属性中配置的事务传播特性处理，如果事务传播特性配置的是mandatory，当前没有事务存在，抛出异常

if(definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY){

throw new IllegalTransactionStateException("No existing transaction found for transaction marked with propagation 'mandatory'");

}

//如果事务传播特性为required、required_new 或nested

else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||

definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED_NEW ||

definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED){

SuspendedResourceHolder suspendedResources = suspend(null);

if(debugEnabled){

logger.debug("Creating new transaction with name [ " + definition.getName() + "]: " + definition);

}

//创建事务

try{

//不激活和当前线程绑定的事务，因为事务传播特性配置要求创建新的事务

boolean newSynchronization = (getTranactionSynchronization() != SYNCHRONIZATION_NEVER);

//创建一个新的事务状态

DefaultTransactionStatus status = newTransactionStatus(definition,transaction,true,newSynchronization,debugEnabled,suspendedResources);

//创建事务的调用，具体实现由具体的事务处理器提供

doBegin(transaction,definition);

//初始化和同步事务状态

prepareSynchronization(status,definition);

return status;

}

catch(RuntimeException ex){

resume(null,suspendResource);

throw ex;

}

catch(Error err){

resume(null,suspendedResources);

throw err;

}

}

else{

//创建空事务，针对supported类型的事务传播特性，激活和当前线程绑定的事务

boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);

//准备事务状态

return prepareTransactionStatus(definition,null,true,newSynchronization,debugEnabled,null);

}

}

//准备事务状态

protected final DefaultTransactionStatus prepareTransactionStatus(TransactionDefinition definition,Object transaction,boolean newTransaction,

boolean newSynchronization,boolean debug,Object suspendedResources){

//创建事务状态

DefaultTransactionStatus status = newTransactionStatus(definition,transaction,newTransaction,newSynchronization,debug,suspendedResource);

//准备事务状态

prepareSynchronization(status,definition);

return status;

}

//创建事务状态

protected DefaultTransactionStatus newTransactionStatus(TransactionDefinition definition,Object transaction,boolean newTransaction,

boolean newSynchronization,boolean debug,Object suspendedResources){

//判断是否是新事务，如果是新事务，则需要把事务属性存放到当前线程中

boolean actualNewSynchronization = newSynchronization && !TransactionSynchronizationManager.isSynchronizationActive();

return new DefaultTransactionStatus(transaction,newTransaction,actualNewSynchronization,definition.isReadOnly(),debug,suspendedResources);

}

//初始化事务属性

protected void prepareSynchronization(DefaultTransactionStatus status,TransactionDefinition definition){

if(status.isNewSynchronization()){

//设置当前是否有活跃事务

TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());

//设置当前事务隔离级别

TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(definition.getIsolationLevel()!= TransactionDefinition.ISOLATION_DEFAULT) ?

definition.getIsolationLevel() : null);

TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());

//设置当前事务名称

TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());

}

}

通过对上面创建事务对象的源码分析，我们看到这抽象事务管理器获取事务对象的这个模板方法主要功能是处理事务属性中配置的事务传播特性，对于判断是否存在事务的

isExistingTransaction方法和创建事务对象的doBegin方法，均委派给具体的事务处理器实现。

4、抽象事务管理器AbstractPlatformTransactionManager处理已经存在的事务：

对于新事务的处理相对比较简单，只需根据事务属性配置创建，同时将事务隔离级别等属性保存到事务绑定的线程资源中。而对于已经存在的事务处理相对比较复杂一些，在抽象事务管理器AbstractPlatformTransactionManager中通过handleExistingTransaction方法来处理已经存在的事务：

private TransactionStatus handleExistingTransaction(TransactionDefinition definition,Object transaction,boolean debugEnabled) throws TransactionException {

//如果事务传播特性为：never，则抛出异常

if(definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {

throw new IllegalTransactionStateException("Existing transaction found for transaction marked with propagation 'never'");

}

//如果事务传播特性是not_supported，同时当前线程存在事务，则将事务挂起

if(definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED){

if(debugEnabled){

logger.debug("Suspending current transaction");

}

//挂起事务

Object suspendedResource = suspend(transaction);

boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);

//创建非事务的事务状态，让方法非事务地执行

return prepareTransactionStatus(definition,null,false,newSynchronization,debugEnabled,suspendResources);

}

//如果事务传播特性是requrired_new，则创建新事务，同时把当前线程中存在的事务挂起

if(definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW){

if(debugEnabed){

logger.debug("Suspending current transaction,creating new transaction with name [ " + defintino.getName() + "]");

}

//挂起已存在的事务

SuspendedResourcesHolder suspendedResources = suspend(transaction);

try{

boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);

//将挂起的事务状态保存起来

DefaultTransactionStatus status = newTransactionStatus(definition,transaction,true,newSynchronization,debugEnabled,suspendedResources);

//创建新事务

doBegin(transaction,definition);

prepareSynchronization(status,definition);

return status;

}

catch(RuntimeException beginEx){

resumeAfterBeginException(transaction,suspendedResources,beginEx);

throw beginEx;

}

catch(Error beginErr){

resumeAfterBeginException(transaction,suspendedResources,beginErr);

throw beginErr;

}

}

//如果事务传播特性是nested嵌套事务

if(definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED){

//如果不允许事务嵌套，则抛出异常

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() + "]");

}

//如果允许使用savepoint保存点保存嵌套事务

if(useSavepointForNestedTransaction()){

//为当前事务创建一个保存点

DefaultTransactionStatus status = prepareTransactionStauts(definition,transaction,false,false,debugEnabled,null);

status.createAndHoldSavepoint();

return status;

}

//如果不允许使用savepoint保存点保存嵌套事务

else{

//使用JTA的嵌套commit/rollback调用

boolean newSynchronization = (getTrasactionSynchronization() != SYNCHRONIZATION_NEVER);

DefaultTransactionStatus status = newTransactionStatus(definition,transaction,true,newSynchronization,debugEnabled,null);

doBegin(transaction,definition);

prepareSynchronization(status,definition);

return status;

}

}

//对于事务传播特性为supported和required的处理

if(debugEnabled){

logger.debug("Participating in existing transaction");

}

//校验已存在的事务，如果已有事务与事务属性配置不一致，则抛出异常

if(isValidateExistingTransaction()){

//如果事务隔离级别不是默认隔离级别

if(definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT){

//获取当前事务的隔离级别

Integer currentIsolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();

//如果获取到的当前事务隔离级别为null或者不等于事务属性配置的隔离级别

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 exist = 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");

}

}

}

boolean newSynchronization = (getTransactionSynchronization()!= SYNCHRONIZATION_NEVER);

//返回当前事务的执行状态

return prepareTransactionStatus(definition,transaction,false,newSynchronization,debugEnabled,null);

}

5、事务的挂起：

如果当前线程存在事务，但事务传播特性又要求开启新事务时，需要将已有的事务进行挂起，事务的挂起涉及线程与事务信息的保存，实现源码如下：

protected final SuspendedResourcesHolder suspend(Object transaction) throws TransactionException {

//如果事务是激活的，且当前线程事务同步机制也是激活状态

if(TransactionSynchronizationManager.isSynchronizationActive()){

//挂起当前线程中所有同步的事务

List<TransactionSynchronization> suspendedSynchronizations = doSuspendedSynchronization();

try{

Object suspendedResources = null;

//把挂起事务的操作交由具体的事务处理器处理

if(transaction != null){

suspendedResource = doSuspend(transaction);

}

//在线程中保存与事务处理有关的信息，并将线程里有关的线程局部变量重置

String name = TransactionSynchronizationManager.getCurrentTransactionName();

//重置当前线程中事务相关的线程局部变量

TransactionSynchronizationManager.setCurrentTransactionName(null);

boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();

TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);

Integer isolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();

TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(null);

boolean wasActive = TransactionSynchronizationManager.isActualTransactionActive();

TransactionSynchronizationManager.setActualTransactionActive(false);

//将当前线程中事务相关信息保存

return new SuspendedResourcesHolder(suspendedResources,suspendedSynchronization,name,readOnly,isolationLevel,wasActive);

}

//对事务挂起操作中产生异常和错误的处理

catch(RuntimeException ex){

doResumeSynchronization(suspendedSynchronizations);

throw ex;

}

catch(Error err){

doResumeSynchronization(suspendedSynchronizations);

throw err;

}

}

//如果事务是激活的，但是事务同步机制不是激活的，则只需要保存事务状态，不需要重置事务相关的线程局部变量

else if(transaction != null){

Object suspendedResources = doSuspend(transaction);

return new SuspendedResourcesHolder(suspendedResources);

}

//事务和事务同步机制都不是激活的，则不需要处理

else{

return null;

}

}

6、事务的提交：

当事务方法处理成功之后，需要将当前事务提交，将更改同步到数据库中，事务提交的实现源码如下：

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);

return;

}

//如果事务没有被标记为回滚时提交，且事务状态是全局回滚

if(!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()){

if(defStatus.isDebug()){

logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");

}

//回滚处理

processRollback(defStatus);

//如果事务状态是新事务，或者在全局回滚时失败

if(status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()){

throw new UnexpectedRollbackException("Transaction rolled back because it has been marked as rollback-only");

}

return;

}

//处理提交

processCommit(defStatus);

}

//提交处理操作

private void processCommit(DefaultTransactionStatus status) throws TransactionException {

try{

boolean beforeCompletionInvoked = false;

try{

//事务提交的准备工作，由具体的事务处理器完成

prepareForCommit(status);

triggerBeforeCommit(status);

triggerBeforeCompletion(status);

beforeCompletionInvoked = true;

boolean globalRollbackOnly = false;

//如果事务状态是新事务，或者全局回滚失败

if(status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()){

//设置事务全局回滚

globalRollbackOnly = status.isGlobalRollbackOnly();

}

//嵌套事务处理

if(status.hasSavepoint()){

if(status.isDebug()){

logger.debug("Releasing transaction savepoint");

}

//释放挂起事务保存点

status.releaseHeldSavepoint();

}

//如果当前事务是新事务

else if(status.isNewTransaction()){

if(status.isDebug()){

logger.debug("Initiating transaction commit");

}

//调用具体事务处理器提交事务

doCommit(status);

}

//如果事务被标记为全局回滚

if(globalRollbackOnly){

throw new UnexpectedRollbackException("Transaction silently rolled back because it has been marked as rollback-only");

}

}

//提交过程中产生未预期的回滚异常，则回滚处理

catch(UnexpectedRollbackException ex){

triggerAfterCompletion(status,TransactionSynchronization.STATUS_ROLLED_BACK);

throw ex;

}

//对提交过程中产生的事务异常处理

catch(TransactionException ex){

//如果回滚失败，则进行回滚异常处理

if(isRollbackOnCommitFailure()){

doRollbackOnCommitException(status,ex);

}

else{

triggerAfterCompletion(status,TransactionSynchronization.STATUS_UNKNOWN);

}

throw ex;

}

//对提交过程产生的异常处理

catch(RuntimeException ex){

//如果不是在完成前调用的

if(!beforeCompletionInvoked){

//触发完成前的回调方法

triggerBeforeCompletion(status);

}

//进行回滚异常处理

doRollbackOnCommitException(status,ex);

throw ex;

}

//对提交过程中产生的错误处理

catch(Error err){

if(!beforeCompletionInvoked) {

triggerBeforeCompletion(status);

}

doRollbackOnCommitException(status,err);

throw err;

}

//触发提交之后的回调操作

try{

triggerAfterCommit(status);

}

finally{

triggerAfterCompletion(status,TransactionSynchronization.STATUS_COMMITTED);

}

}

//提交完成之后清除事务相关状态

finally{

cleanupAfterCompletion(status);

}

}

7、事务的回滚：

当在事务处理过程中产生异常，或者提交失败时，往往需要对数据库中已有的更改做回滚操作，即恢复到操作之前的状态，回滚的实现代码如下：

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);

}

//回滚操作

private void processRollback(DefaultTransactionStatus status) {

try{

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 if(status.hasTransaction()){

//如果当前事务状态为本地回滚，或全局回滚失败

if(status.isLocalRollbackOnly() || isGlobalbackOnParticipationFailure()){

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 origination decide on rollback");

}

}

}

//如果当前线程没有事务

else{

logger.debug("Should roll back transaction but cannot - no transaction available");

}

}

//对回滚操作过程中的运行时异常和错误的处理

catch(RuntimeException ex){

triggerAfterCompletion(status,TransactionSynchronization.STATUS_UNKNOWN);

throw ex;

}

catch(Error err){

triggerAfterCompletion(status,TransactionSynchronization.STATUS_UNKNOWN);

throw err;

}

//回滚操作完成后，触发回滚之后回调操作

triggerAfterCompletion(status,TransactionSynchronization.STATUS_ROLLED_BACK);

}

//清除回滚之后事务状态信息

finally{

cleanupAfterCompletion(status);

}

}