spring 事务原理

关注
发布于: 2020 年 05 月 20 日
事务有原子性、一致性、隔离性、持久性的特点，在开发中我们将一组不可分割的操作放在事务中，要么一起成功，要么一起失败，例如最简单的转账，我们看一下spring是如何实现事务管理的。
spring事务的关键对象spring的事务实现中有几个关键对象，我们先来认识一下:
PlatformTransactionManager事务管理器接口只有三个方法：
﻿
getTransaction：获取当前激活的事务或者创建一个事务
commit：提交当前事务
rollback：回滚当前事务
﻿
PlatformTransactionManager有很多具体的实现：
DataSourceTransactionManager，使用JDBC与ibatis进行持久化
JtaTransactionManager：使用分布式事务
JpaTransactionManager：使用jpa
HibernateTransactionManager：使用hibernate
﻿
TransactionDefiition﻿
TransactionDefiition是事务属性的定义，其中包含事务的隔离级别、事务传播类型、事务过期时间、事务是否可读等事务信息。
﻿
﻿
spring的隔离级别和数据库的是对应的，只是多了一个ISOLATION_DEFAULT类型，表示使用mysql默认隔离级别。
﻿
spring的事务传播类型如下：
﻿
事务的具体实现我们用一个简单的例子看一下spring的事务是如何实现的
新建一个账户表，包含几个基本字段，金额、姓名等
﻿
CREATE TABLE `account` (
  `ID` int(11) NOT NULL AUTO_INCREMENT,
  `count` int(11) NOT NULL DEFAULT '0',
  `name` varchar(400) DEFAULT NULL COMMENT '姓名',
  `update` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`ID`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;
新建一个springboot工程，引入jdbc与mysql的依赖
新建一个service，在用一个controller调用此service的转账方法
@Service
public class AccountService {
    private JdbcTemplate jdbctemplate;
    public AccountService(DataSource dataSource){
        jdbctemplate = new JdbcTemplate(dataSource);
    }
    @Transactional
    public int update(int count,int fromId,int toId){
        int update = jdbctemplate.update("update account set count = count-? where id=?", count, fromId);
//        int i = 0 / 0;
        jdbctemplate.update("update account set count = count+? where id=?", count, toId);
        return update;
    }
}
为了探究spring是如何实现事务的，我们通过tcpdump抓mysql的包来看执行的语句
 //楼主是mac环境
 sudo tcpdump -i lo0  -s 0 -l -w -  port 3306 | strings
我们执行用户1给用户2转账5元，看到抓包结果为
﻿
﻿
可以看到，在第一个语句执行前设置为手动提交，在所有语句执行完后执行commit，最后再设置成自动提交。
如果中间执行失败呢，我们将 int i = 0 / 0;这行的注释打开，制造一个RuntimeException，看看执行结果
﻿
事务被执行回滚了。
分析事务代码﻿
spring中可以用aop注入TransactionInterceptor，或者像前面的例子一样，在需要事务管理的方法上加上注解@Transactional，spring在此方法执行时会执行个方法的interceptor集合，事务的interceptor是org.springframework.transaction.interceptor.TransactionInterceptor，它继承自TransactionAspectSupport，事务的具体实现在TransactionAspectSupport中。
﻿
TransactionInterceptor在执行时会调用方法invoke
  @Override
	@Nullable
	public Object invoke(MethodInvocation invocation) throws Throwable {
		Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
		return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
	}
在invokeWithinTransaction中会调用createTransactionIfNecessary方法
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
			@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
		// 设置事务名称
		if (txAttr != null && txAttr.getName() == null) {
			txAttr = new DelegatingTransactionAttribute(txAttr) {
				@Override
				public String getName() {
					return joinpointIdentification;
				}
			};
		}
		TransactionStatus status = null;
		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");
				}
			}
		}
  	//填充事务信息
		return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
	}
﻿
tm.getTransaction(txAttr);这里会使用具体持久化方式对应的那个TransactionManager，楼主使用的Jdbc，实现是DataSourceTransactionManager，getTransaction方法调用的是父类中的方法
﻿
org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction
﻿
@Override
	public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
			throws TransactionException {
		//如果TransactionDefinition为空则使用默认配置
		TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
		Object transaction = doGetTransaction();
		boolean debugEnabled = logger.isDebugEnabled();
		if (isExistingTransaction(transaction)) {
			//如果有存在的事务，则处理存在的事物
			return handleExistingTransaction(def, transaction, debugEnabled);
		}
		// 判断事务是否超时
		if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
			throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
		}
		// 使用PROPAGATION_MANDATORY这个级别，没有可用的事务则抛异常
		if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
			throw new IllegalTransactionStateException(
					"No existing transaction found for transaction marked with propagation 'mandatory'");
		}
		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);
		}
	}
﻿
startTransaction会调用DataSourceTransactionManager的doBegin方法设置当前连接为手动提交事务
﻿
@Override
	protected void doBegin(Object transaction, TransactionDefinition definition) {
		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
		Connection con = null;
		try {
      //获取数据库连接
			if (!txObject.hasConnectionHolder() ||
					txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
				Connection newCon = obtainDataSource().getConnection();
				if (logger.isDebugEnabled()) {
					logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
				}
				txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
			}
			txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
			con = txObject.getConnectionHolder().getConnection();
			Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
			txObject.setPreviousIsolationLevel(previousIsolationLevel);
			txObject.setReadOnly(definition.isReadOnly());
			//如果连接是自动提交，则设置成手动
			if (con.getAutoCommit()) {
				txObject.setMustRestoreAutoCommit(true);
				if (logger.isDebugEnabled()) {
					logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
				}
        //这里会和数据库通信
				con.setAutoCommit(false);
			}
			prepareTransactionalConnection(con, definition);
			txObject.getConnectionHolder().setTransactionActive(true);
			int timeout = determineTimeout(definition);
			if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
				txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
			}
			// Bind the connection holder to the thread.
			if (txObject.isNewConnectionHolder()) {
				TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
			}
		}
		catch (Throwable ex) {
			if (txObject.isNewConnectionHolder()) {
				DataSourceUtils.releaseConnection(con, obtainDataSource());
				txObject.setConnectionHolder(null, false);
			}
			throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
		}
	}
﻿
再看回org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction方法，这里显示了aop如何处理事务
﻿
if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
		
			//获取事务
			TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
			Object retVal;
		 	try {
				//触发后面的拦截器和方法本身
				retVal = invocation.proceedWithInvocation();
			}
			catch (Throwable ex) {
				// 捕获异常处理回滚
				completeTransactionAfterThrowing(txInfo, ex);
				throw ex;
			}
			finally {
        //重置threadLocal中事务状态
				cleanupTransactionInfo(txInfo);
			}
			//省略
			...
			return retVal;
		}
﻿
再看一下completeTransactionAfterThrowing方法，如果是需要回滚的异常则执行回滚，否则执行提交
﻿
protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {
		if (txInfo != null && txInfo.getTransactionStatus() != null) {
			if (logger.isTraceEnabled()) {
				logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() +
						"] after exception: " + ex);
			}
			//需要回滚的异常
			if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
				try {
					//执行回滚
					txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
				}
				catch (TransactionSystemException ex2) {
					logger.error("Application exception overridden by rollback exception", ex);
					ex2.initApplicationException(ex);
					throw ex2;
				}
				catch (RuntimeException | Error ex2) {
					logger.error("Application exception overridden by rollback exception", ex);
					throw ex2;
				}
			}
			else {
				try {
					//提交
					txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
				}
				catch (TransactionSystemException ex2) {
					logger.error("Application exception overridden by commit exception", ex);
					ex2.initApplicationException(ex);
					throw ex2;
				}
				catch (RuntimeException | Error ex2) {
					logger.error("Application exception overridden by commit exception", ex);
					throw ex2;
				}
			}
		}
	}
﻿
总结﻿
spring中的事务实现从原理上说比较简单，通过aop在方法执行前后增加数据库事务的操作。
1、在方法开始时判断是否开启新事务，需要开启事务则设置事务手动提交 set autocommit=0;
2、在方法执行完成后手动提交事务 commit;
3、在方法抛出指定异常后调用rollback回滚事务
spring事务管理有很多细节的处理，如传播方式的实现，感兴趣的同学自己去看看源码喽~
﻿