前言
业务系统的数据,一般最后都会落入到数据库中,例如 MySQL、Oracle 等主流数据库,不可避免的,在数据更新时,有可能会遇到错误,这时需要将之前的数据更新操作撤回,避免错误数据。
Spring 的声明式事务能帮我们处理回滚操作,让我们不需要去关注数据库底层的事务操作,可以不用在出现异常情况下,在 try / catch / finaly 中手写回滚操作。
Spring 的事务保证程度比行业中其它技术(例如 TCC / 2PC / 3PC 等)稍弱一些,但使用 Spring 事务已经满足大部分场景,所以它的使用和配置规则也是值得学习的。
事务属性和行为
ACID属性
提到事务,不可避免需要涉及到事务的ACID属性:
- 原子性(Atomicity):事务作为一个整体被执行,包含在其中的对数据库的操作要么全部被执行,要么都不执行。
- 一致性(Consistency):事务应确保数据库的状态从一个一致状态转变为另一个一致状态。一致状态的含义是数据库中的数据应满足完整性约束。
- 隔离性(Isolation):多个事务并发执行时,一个事务的执行不应影响其他事务的执行。
- 持久性(Durability):已被提交的事务对数据库的修改应该永久保存在数据库中。
我们将严格遵循ACID属性的事务称为刚性事务。与之相对,期望最终一致性,在事务执行的中间状态允许暂时不遵循ACID属性的事务称为柔性事务,柔性事务的使用涉及到分布式事务方案,这里我们先将注意集中在事务实现原理上。
隔离级别
根据SQL92标准,MySQL的InnoDB引擎提供四种隔离级别(即ACID中的I):读未提交(READ UNCOMMITTED)、读已提交(READ COMMITTED)、可重复读(REPEATABLE READ)和串行化(SERIALIZABLE),InnoDB默认的隔离级别是REPEATABLE READ,其可避免脏读和不可重复读,但不能避免幻读,需要指出的是,InnoDB引擎的多版本并发控制机制(MVCC)并没有完全避免幻读。
事物传播行为
支持当前事务
支持当前事务的传播机制有三种,分别是
-
REQUIRED (必须有)
含义:如果当前方法没有事务,新建一个事务,如果已经存在一个事务中,则加入到这个事务中。 -
SUPPORTS (可有可无)
含义:支持当前事务,如果当前没有事务,就以非事务方式执行 -
MANDATORY (强制)
含义:使用当前的事务,如果当前没有事务,就抛出异常。
不支持当前事务
-
REQUIRES_NEW
含义:新建事务,如果当前存在事务,把当前事务挂起。 -
NOT_SUPPORTED
含义:以非事务方式执行操作,如果当前存在事务,就把当前事务挂起。 -
NEVER
含义: 以非事务方式执行,如果当前存在事务,则抛出异常。
NESTED
含义: 如果当前存在事务,则在嵌套事务内执行。如果当前没有事务,则执行与PROPAGATION_REQUIRED类似的操作。
事务行为
事务的行为包括事务开启、事务提交和事务回滚。InnoDB所有的用户SQL执行都在事务控制之内,在默认情况下,autocommit设置为true,单条SQL执行成功后,MySQL会自动提交事务,或者如果SQL执行出错,则根据异常类型执行事务提交或者回滚。可以使用START TRANSACTION(SQL标准)或者BEGIN开启事务,使用COMMIT和ROLLBACK提交和回滚事务;也可以通过设置autocommit属性来控制事务行为,当设置autocommit为false时,其后执行的多条SQL语句将在一个事务内,直到执行COMMIT或者ROLLBACK事务才会提交或者回滚。
AOP增强
Spring使用AOP(面向切面编程)来实现声明式事务,后续在讲Spring事务具体实现的时候会详细说明,说下动态代理和AOP增强。
动态代理是Spring实现AOP的默认方式,分为两种:JDK动态代理和CGLIB动态代理。JDK动态代理面向接口,通过反射生成目标代理接口的匿名实现类;CGLIB动态代理则通过继承,使用字节码增强技术(或者objenesis
类库)为目标代理类生成代理子类。Spring默认对接口实现使用JDK动态代理,对具体类使用CGLIB,同时也支持配置全局使用CGLIB来生成代理对象。
我们在切面配置中会使用到@Aspect注解,这里用到了Aspectj的切面表达式。Aspectj是java语言实现的一个AOP框架,使用静态代理模式,拥有完善的AOP功能,与Spring AOP互为补充。Spring采用了Aspectj强大的切面表达式定义方式,但是默认情况下仍然使用动态代理方式,并未使用Aspectj的编译器和织入器,当然也支持配置使用Aspectj静态代理替代动态代理方式。Aspectj功能更强大,比方说它支持对字段、POJO类进行增强,与之相对,Spring只支持对Bean方法级别进行增强。
Spring对方法的增强有五种方式:
- 前置增强(org.springframework.aop.BeforeAdvice):在目标方法执行之前进行增强;
- 后置增强(org.springframework.aop.AfterReturningAdvice):在目标方法执行之后进行增强;
- 环绕增强(org.aopalliance.intercept.MethodInterceptor):在目标方法执行前后都执行增强;
- 异常抛出增强(org.springframework.aop.ThrowsAdvice):在目标方法抛出异常后执行增强;
- 引介增强(org.springframework.aop.IntroductionInterceptor):为目标类添加新的方法和属性。
声明式事务的实现就是通过环绕增强的方式,在目标方法执行之前开启事务,在目标方法执行之后提交或者回滚事务,事务拦截器的继承关系图可以体现这一点:
Spring事务抽象
统一一致的事务抽象是Spring框架的一大优势,无论是全局事务还是本地事务,JTA、JDBC、Hibernate还是JPA,Spring都使用统一的编程模型,使得应用程序可以很容易地在全局事务与本地事务,或者不同的事务框架之间进行切换。下图是Spring事务抽象的核心类图:
接口PlatformTransactionManager定义了事务操作的行为,其依赖TransactionDefinition和TransactionStatus接口,其实大部分的事务属性和行为我们以MySQL数据库为例已经有过了解,这里再对应介绍下。
PlatformTransactionManager:事务管理器
- getTransaction方法:事务获取操作,根据事务属性定义,获取当前事务或者创建新事物;
- commit方法:事务提交操作,注意这里所说的提交并非直接提交事务,而是根据当前事务状态执行提交或者回滚操作;
- rollback方法:事务回滚操作,同样,也并非一定直接回滚事务,也有可能只是标记事务为只读,等待其他调用方执行回滚。
TransactionDefinition:事务属性定义
- getPropagationBehavior方法:返回事务的传播属性,默认是PROPAGATION_REQUIRED;
- getIsolationLevel方法:返回事务隔离级别,事务隔离级别只有在创建新事务时才有效,也就是说只对应传播属性PROPAGATION_REQUIRED和PROPAGATION_REQUIRES_NEW;
- getTimeout方法:返回事务超时时间,以秒为单位,同样只有在创建新事务时才有效;
- isReadOnly方法:是否优化为只读事务,支持这项属性的事务管理器会将事务标记为只读,只读事务不允许有写操作,不支持只读属性的事务管理器需要忽略这项设置,这一点跟其他事务属性定义不同,针对其他不支持的属性设置,事务管理器应该抛出异常。
- getName方法:返回事务名称,声明式事务中默认值为“类的完全限定名.方法名”。
TransactionStatus:当前事务状态
- isNewTransaction方法:当前方法是否创建了新事务(区别于使用现有事务以及没有事务);
- hasSavepoint方法:在嵌套事务场景中,判断当前事务是否包含保存点;
- setRollbackOnly和isRollbackOnly方法:只读属性设置(主要用于标记事务,等待回滚)和查询;
- flush方法:刷新底层会话中的修改到数据库,一般用于刷新如Hibernate/JPA的会话,是否生效由具体事务资源实现决定;
- isCompleted方法:判断当前事务是否已完成(已提交或者已回滚)。
部分Spring包含的对PlatformTransactionManager的实现类如下图所示:
AbstractPlatformTransactionManager抽象类实现了Spring事务的标准流程,其子类DataSourceTransactionManager是我们使用较多的JDBC单数据源事务管理器,而JtaTransactionManager是JTA(Java Transaction API)规范的实现类,另外两个则分别是JavaEE容器WebLogic和WebSphere的JTA事务管理器的具体实现。
Spring事务切面
之前提到,Spring采用AOP来实现声明式事务,那么事务的AOP切面是如何织入的呢?这一点涉及到AOP动态代理对象的生成过程。
代理对象生成的核心类是AbstractAutoProxyCreator,实现了BeanPostProcessor接口,会在Bean初始化完成之后,通过postProcessAfterInitialization方法生成代理对象。
看一下AbstractAutoProxyCreator
类的核心代码,主要关注三个方法:postProcessAfterInitialization、wrapIfNecessary和createProxy,为了突出核心流程,以注释代替了部分代码的具体实现,后续的源码分析也采用相同的处理。
public abstract class AbstractAutoProxyCreator extends ProxyProcessorSupport
implements SmartInstantiationAwareBeanPostProcessor, BeanFactoryAware {
@Override
public Object postProcessAfterInitialization(@Nullable Object bean, String beanName) {
if (bean != null) {
Object cacheKey = getCacheKey(bean.getClass(), beanName);
//当 Bean 被循环引用, 并且被暴露了,
// 则会通过 getEarlyBeanReference 来创建代理类;
// 通过判断 earlyProxyReferences 中
// 是否存在 beanName 来决定是否需要对 target 进行动态代理
if (this.earlyProxyReferences.remove(cacheKey) != bean) {
//该方法将会返回代理类
return wrapIfNecessary(bean, beanName, cacheKey);
}
}
return bean;
}
protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
//已经被处理过
// 1.判断当前bean是否在targetSourcedBeans缓存中存在(已经处理过),如果存在,则直接返回当前bean
if (StringUtils.hasLength(beanName) && this.targetSourcedBeans.contains(beanName)) {
return bean;
}
//不需要被织入逻辑的
// 2.在advisedBeans缓存中存在,并且value为false,则代表无需处理
if (Boolean.FALSE.equals(this.advisedBeans.get(cacheKey))) {
return bean;
}
//是不是基础的bean 是不是需要跳过的
// 3.bean的类是aop基础设施类 || bean应该跳过,则标记为无需处理,并返回
if (isInfrastructureClass(bean.getClass()) || shouldSkip(bean.getClass(), beanName)) {
this.advisedBeans.put(cacheKey, Boolean.FALSE);
return bean;
}
// Create proxy if we have advice.
// 返回匹配当前Bean的所有Advice\Advisor\Interceptor
Object[] specificInterceptors = getAdvicesAndAdvisorsForBean(bean.getClass(), beanName, null);
// 5.如果存在增强器则创建代理
if (specificInterceptors != DO_NOT_PROXY) {
this.advisedBeans.put(cacheKey, Boolean.TRUE);
//创建Bean对应的代理,SingletonTargetSource用于封装实现类的信息
// 5.1 创建代理对象:这边SingletonTargetSource的target属性存放的就是我们原来的bean实例(也就是被代理对象),
// 用于最后增加逻辑执行完毕后,通过反射执行我们真正的方法时使用(method.invoke(bean, args))
Object proxy = createProxy(
bean.getClass(), beanName, specificInterceptors, new SingletonTargetSource(bean));
// 5.2 创建完代理后,将cacheKey -> 代理类的class放到缓存
this.proxyTypes.put(cacheKey, proxy.getClass());
// 返回代理对象
return proxy;
}
//该Bean是不需要进行代理的,下次就不需要重复生成了
this.advisedBeans.put(cacheKey, Boolean.FALSE);
return bean;
}
protected Object createProxy(Class<?> beanClass, @Nullable String beanName,
@Nullable Object[] specificInterceptors, TargetSource targetSource) {
//如果beanFactory是ConfigurableListableBeanFactory的类型,暴露目标类
if (this.beanFactory instanceof ConfigurableListableBeanFactory) {
AutoProxyUtils.exposeTargetClass((ConfigurableListableBeanFactory) this.beanFactory, beanName, beanClass);
}
//创建一个ProxyFactory,当前ProxyCreator在创建代理时将需要用到的字段赋值到ProxyFactory中去
ProxyFactory proxyFactory = new ProxyFactory();
//将 当前的AnnotationAwareAspectJAutoProxyCreator 对象的属性赋值给ProxyFactory对象
proxyFactory.copyFrom(this);
// 处理 proxyTargetClass 属性
// 如果希望使用 CGLIB 来代理接口,可以配置
// proxy-target-class="true",这样不管有没有接口,都使用 CGLIB 来生成代理:
// <aop:config proxy-target-class="true"></aop:config>
if (!proxyFactory.isProxyTargetClass()) {
// 检查preserveTargetClass属性,判断beanClass是应该基于类代理还是基于接口代理
if (shouldProxyTargetClass(beanClass, beanName)) {
// 如果是基于类代理,则将proxyTargetClass赋值为true
proxyFactory.setProxyTargetClass(true);
}
else {
// 评估bean的代理接口
// 1. 有接口的,调用一次或多次:proxyFactory.addInterface(ifc);
// 2. 没有接口的,调用:proxyFactory.setProxyTargetClass(true);
evaluateProxyInterfaces(beanClass, proxyFactory);
}
}
// 这个方法主要来对前面传递进来的横切逻辑实例进行包装
// 注意:如果 specificInterceptors 中有 Advice 和 Interceptor,它们也会被包装成 Advisor
// 方法会整理合并得到最终的advisors (毕竟interceptorNames还指定了一些拦截器的)
// 至于调用的先后顺序,通过方法里的applyCommonInterceptorsFirst参数可以进行设置,
// 若applyCommonInterceptorsFirst为true,interceptorNames属性指定的Advisor优先调用。默认为true
Advisor[] advisors = buildAdvisors(beanName, specificInterceptors);
// 将advisors添加到proxyFactory
proxyFactory.addAdvisors(advisors);
// 设置要代理的类,将targetSource赋值给proxyFactory的targetSource属性,之后可以通过该属性拿到被代理的bean的实例
proxyFactory.setTargetSource(targetSource);
// 这个方法是交给子类的,子类可以继续去定制此proxyFactory
customizeProxyFactory(proxyFactory);
// 用来控制proxyFactory被配置之后,是否还允许修改通知。默认值为false(即在代理被配置之后,不允许修改代理类的配置)
proxyFactory.setFrozen(this.freezeProxy);
// 设置preFiltered的属性值,默认是false。子类:AbstractAdvisorAutoProxyCreator修改为true
// preFiltered字段意思为:是否已为特定目标类筛选Advisor
// 这个字段和DefaultAdvisorChainFactory.getInterceptorsAndDynamicInterceptionAdvice获取所有的Advisor有关
//CglibAopProxy和JdkDynamicAopProxy都会调用此方法,然后递归执行所有的Advisor
if (advisorsPreFiltered()) {
proxyFactory.setPreFiltered(true);
}
// 使用proxyFactory获取代理
return proxyFactory.getProxy(getProxyClassLoader());
}
}
最后是通过调用ProxyFactory#getProxy(java.lang.ClassLoader)方法来创建代理对象:
public class ProxyFactory extends ProxyCreatorSupport {
public Object getProxy(@Nullable ClassLoader classLoader) {
// 首先获取AopProxy对象,其主要有两个实现:JdkDynamicAopProxy和ObjenesisCglibAopProxy,
// 分别用于Jdk和Cglib代理类的生成,其getProxy()方法则用于获取具体的代理对象
// 1.createAopProxy:创建AopProxy
// 2.getProxy(classLoader):获取代理对象实例
return createAopProxy().getProxy(classLoader);
}
}
public class ProxyCreatorSupport extends AdvisedSupport {
private AopProxyFactory aopProxyFactory;
public ProxyCreatorSupport() {
this.aopProxyFactory = new DefaultAopProxyFactory();
}
protected final synchronized AopProxy createAopProxy() {
//主要是为了激活AdvisedSupportListener监听器
if (!this.active) {
activate();
}
// 2.创建AopProxy
return getAopProxyFactory().createAopProxy(this);
}
}
ProxyFactory的父类构造器实例化了DefaultAopProxyFactory类,从其源代码我们可以看到Spring动态代理方式选择策略的实现:如果目标类optimize,proxyTargetClass属性设置为true或者未指定需要代理的接口,则使用CGLIB生成代理对象,否则使用JDK动态代理。
public class DefaultAopProxyFactory implements AopProxyFactory, Serializable {
@Override
public AopProxy createAopProxy(AdvisedSupport config) throws AopConfigException {
// 1.判断使用JDK动态代理还是Cglib代理
// optimize:用于控制通过cglib创建的代理是否使用激进的优化策略。除非完全了解AOP如何处理代理优化,
// 否则不推荐使用这个配置,目前这个属性仅用于cglib代理,对jdk动态代理无效
// proxyTargetClass:默认为false,设置为true时,强制使用cglib代理,设置方式:<aop:aspectj-autoproxy proxy-target-class="true" />
// hasNoUserSuppliedProxyInterfaces:config是否存在代理接口或者只有SpringProxy一个接口
if (config.isOptimize() || config.isProxyTargetClass() || hasNoUserSuppliedProxyInterfaces(config)) {
// 拿到要被代理的对象的类型
Class<?> targetClass = config.getTargetClass();
if (targetClass == null) {
// TargetSource无法确定目标类:代理创建需要接口或目标。
throw new AopConfigException("TargetSource cannot determine target class: " +
"Either an interface or a target is required for proxy creation.");
}
// 要被代理的对象是接口 || targetClass是Proxy class 已经是个JDK的代理类(Proxy的子类,所有的JDK代理类都是此类的子类)
// 当且仅当使用getProxyClass方法或newProxyInstance方法动态生成指定的类作为代理类时,才返回true。
if (targetClass.isInterface() || Proxy.isProxyClass(targetClass)) {
// JDK动态代理,这边的入参config(AdvisedSupport)实际上是ProxyFactory对象
// 具体为:AbstractAutoProxyCreator中的proxyFactory.getProxy发起的调用,在ProxyCreatorSupport使用了this作为参数,
// 调用了的本方法,这边的this就是发起调用的proxyFactory对象,而proxyFactory对象中包含了要执行的的拦截器
return new JdkDynamicAopProxy(config);
}
// Cglib代理
return new ObjenesisCglibAopProxy(config);
}
else {
// JDK动态代理
return new JdkDynamicAopProxy(config);
}
}
}
Spring事务拦截
我们已经了解了AOP切面织入生成代理对象的过程,当Bean方法通过代理对象调用时,会触发对应的AOP增强拦截器,前面提到声明式事务是一种环绕增强,对应接口为MethodInterceptor,事务增强对该接口的实现为TransactionInterceptor,类图如下:
事务拦截器TransactionInterceptor在invoke方法中,通过调用父类TransactionAspectSupport的invokeWithinTransaction方法进行事务处理,该方法支持声明式事务和编程式事务。
public abstract class TransactionAspectSupport implements BeanFactoryAware, InitializingBean {
// invokeWithinTransaction最为核心的处理事务的模版方法了:
//protected修饰,不允许其他包和无关类调用
@Nullable
protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
final InvocationCallback invocation) throws Throwable {
// If the transaction attribute is null, the method is non-transactional.
// 查询目标方法事务属性、确定事务管理器、构造连接点标识(用于确认事务名称)
TransactionAttributeSource tas = getTransactionAttributeSource();
final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
// 根据事务的属性获取beanFactory中的PlatformTransactionManager(spring事务管理器的顶级接口),
// 一般这里或者的是DataSourceTransactiuonManager
final TransactionManager tm = determineTransactionManager(txAttr);
if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {
ReactiveTransactionSupport txSupport = this.transactionSupportCache.computeIfAbsent(method, key -> {
if (KotlinDetector.isKotlinType(method.getDeclaringClass()) && KotlinDelegate.isSuspend(method)) {
throw new TransactionUsageException(
"Unsupported annotated transaction on suspending function detected: " + method +
". Use TransactionalOperator.transactional extensions instead.");
}
ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(method.getReturnType());
if (adapter == null) {
throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " +
method.getReturnType());
}
return new ReactiveTransactionSupport(adapter);
});
return txSupport.invokeWithinTransaction(
method, targetClass, invocation, txAttr, (ReactiveTransactionManager) tm);
}
PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
// 目标方法唯一标识(类.方法,如service.UserServiceImpl.save)
final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
//如果txAttr为空或者tm 属于非CallbackPreferringPlatformTransactionManager,执行目标增强
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 (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;
}
//编程式事务处理(CallbackPreferringPlatformTransactionManager) 不做重点分析
else {
final ThrowableHolder throwableHolder = new ThrowableHolder();
//省略编程式事务处理相关代码
}
}
}
在讲Spring事务抽象时,有提到事务抽象的核心接口为PlatformTransactionManager,它负责管理事务行为,包括事务的获取、提交和回滚。在invokeWithinTransaction方法中,我们可以看到createTransactionIfNecessary、commitTransactionAfterReturning和completeTransactionAfterThrowing都是针对该接口编程,并不依赖于特定事务管理器,这里是对Spring事务抽象的实现。
public abstract class TransactionAspectSupport implements BeanFactoryAware, InitializingBean {
// 若有需要 创建一个TransactionInfo (具体的事务从事务管理器里面getTransaction()出来~)
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
// If no name specified, apply method identification as transaction name.
// 如果没有名称指定则使用方法唯一标识,并使用 DelegatingTransactionAttribute 包装 txAttr
if (txAttr != null && txAttr.getName() == null) {
txAttr = new DelegatingTransactionAttribute(txAttr) {
@Override
public String getName() {
return joinpointIdentification;
}
};
}
// 从事务管理器里,通过txAttr拿出来一个TransactionStatus
TransactionStatus status = null;
if (txAttr != null) {
if (tm != null) {
// 获取 TransactionStatus
status = tm.getTransaction(txAttr);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
"] because no transaction manager has been configured");
}
}
}
// 根据指定的属性与 status 等,转换成一个通用的TransactionInfo
return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}
protected TransactionInfo prepareTransactionInfo(@Nullable PlatformTransactionManager tm,
@Nullable TransactionAttribute txAttr, String joinpointIdentification,
@Nullable TransactionStatus status) {
// 构造一个TransactionInfo
TransactionInfo txInfo = new TransactionInfo(tm, txAttr, joinpointIdentification);
if (txAttr != null) {
//省略部分代码......
// 如果已存在不兼容的Tx,设置status
txInfo.newTransactionStatus(status);
}
//省略部分代码......
// 这句话是最重要的:把生成的TransactionInfo并绑定到当前线程的ThreadLocal
txInfo.bindToThread();
return txInfo;
}
//提交事务
protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) {
if (txInfo != null && txInfo.getTransactionStatus() != null) {
if (logger.isTraceEnabled()) {
logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "]");
}
// 提交事务
txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
}
}
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;
}
}
}
}
protected static final class TransactionInfo {
@Nullable
private final PlatformTransactionManager transactionManager;
}
}
另外,在获取事务时,AbstractPlatformTransactionManager#doBegin方法负责开启新事务,在DataSourceTransactionManager有如下代码:
public class DataSourceTransactionManager extends AbstractPlatformTransactionManager
implements ResourceTransactionManager, InitializingBean {
private DataSource dataSource;
@Override
protected void doBegin(Object transaction, TransactionDefinition definition) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
Connection con = null;
try {
if (!txObject.hasConnectionHolder() ||
txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
// 从DataSource里获取一个连接(这个DataSource一般是有连接池的)
Connection newCon = obtainDataSource().getConnection();
if (logger.isDebugEnabled()) {
logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
}
// 把这个链接用ConnectionHolder包装一下
txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
}
txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
con = txObject.getConnectionHolder().getConnection();
// 设置isReadOnly、设置隔离界别等
Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
txObject.setPreviousIsolationLevel(previousIsolationLevel);
txObject.setReadOnly(definition.isReadOnly());
// Switch to manual commit if necessary. This is very expensive in some JDBC drivers,
// so we don't want to do it unnecessarily (for example if we've explicitly
// configured the connection pool to set it already).
// 这里非常的关键,先看看Connection 是否是自动提交的
// 如果是 就con.setAutoCommit(false) 要不然数据库默认没执行一条SQL都是一个事务,就没法进行事务的管理了
if (con.getAutoCommit()) {
txObject.setMustRestoreAutoCommit(true);
if (logger.isDebugEnabled()) {
logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
}
//开启事务,设置autoCommit为false
con.setAutoCommit(false);
}
// ====因此从这后面,通过此Connection执行的所有SQL语句只要没有commit就都不会提交给数据库的=====
// 这个方法特别特别有意思 它自己`Statement stmt = con.createStatement()`拿到一个Statement
// 然后执行了一句SQL:`stmt.executeUpdate("SET TRANSACTION READ ONLY");`
// 所以:如果你仅仅只是查询。把事务的属性设置为readonly=true Spring对帮你对SQl进行优化的
// 需要注意的是:readonly=true 后,只能读,不能进行dml操作)
// (只能看到设置事物前数据的变化,看不到设置事物后数据的改变)
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()) {
//这里将当前的connection放入TransactionSynchronizationManager中持有,如果下次调用可以判断为已有的事务
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);
}
}
}
这里才真正开启了数据库事务。
Spring事务同步
提到事务传播机制时,我们经常提到一个条件“如果当前已有事务”,那么Spring是如何知道当前是否已经开启了事务呢?在AbstractPlatformTransactionManager中是这样做的:
public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
// 最为重要的一个方法,根据实物定义,获取到一个事务TransactionStatus
@Override
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
throws TransactionException {
// Use defaults if no transaction definition given.
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
//doGetTransaction()方法是抽象方法,具体的实现由具体的事务处理器提供(下面会以DataSourceTransactionManager为例子)
Object transaction = doGetTransaction();
boolean debugEnabled = logger.isDebugEnabled();
//检查当前线程是否存在事务 isExistingTransaction此方法默认返回false 但子类都复写了此方法
if (isExistingTransaction(transaction)) {
// Existing transaction found -> check propagation behavior to find out how to behave.
// handleExistingTransaction方法为处理已经存在事务的情况
// 这个方法的实现也很复杂,总之还是对一些传播属性进行解析,各种情况的考虑~~~~~ 如果有新事务产生 doBegin()就会被调用~~~~
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.
// PROPAGATION_MANDATORY 如果已经存在一个事务,支持当前事务。如果没有一个活动的事务,则抛出异常
if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
throw new IllegalTransactionStateException(
"No existing transaction found for transaction marked with propagation 'mandatory'");
}
//如果事务传播特性为required、required_new或nested
else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
// 挂起,但是doSuspend()由子类去实现~~~
// 挂起操作,触发相关的挂起注册的事件,把当前线程事物的所有属性都封装好,放到一个SuspendedResourcesHolder
// 然后清空清空一下`当前线程事务`
SuspendedResourcesHolder suspendedResources = suspend(null);
if (debugEnabled) {
logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
}
try {
// 创建一个新的事务状态 就是new DefaultTransactionStatus() 把个属性都赋值上
return startTransaction(def, transaction, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error ex) {
//重新开始 doResume由子类去实现
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);
}
// 这个方法相当于先newTransactionStatus,再prepareSynchronization这两步~~~
// 显然和上面的区别是:中间不回插入调用doBegin()方法,因为没有事务 begin个啥
boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
}
}
private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
//创建一个新的事务状态 就是new DefaultTransactionStatus() 把个属性都赋值上
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
// 开启事物,抽象方法,由子类去实现~
doBegin(transaction, definition);
// 初始化事物同步属性
//初始化和同步事务状态 是TransactionSynchronizationManager这个类 它内部维护了很多的ThreadLocal
prepareSynchronization(status, definition);
return status;
}
}
注意getTransaction方法是final的,无法被子类覆盖,保证了获取事务流程的一致和稳定。抽象方法doGetTransaction获取当前事务对象,方法isExistingTransaction判断当前事务对象是否存在活跃事务,具体逻辑由特定事务管理器实现,看下我们使用最多的DataSourceTransactionManager对应的实现:
public class DataSourceTransactionManager extends AbstractPlatformTransactionManager
implements ResourceTransactionManager, InitializingBean {
// 这里返回的是一个`DataSourceTransactionObject`
// 它是一个`JdbcTransactionObjectSupport`,所以它是SavepointManager、实现了SmartTransactionObject接口
@Override
protected Object doGetTransaction() {
DataSourceTransactionObject txObject = new DataSourceTransactionObject();
txObject.setSavepointAllowed(isNestedTransactionAllowed());
ConnectionHolder conHolder =
(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
txObject.setConnectionHolder(conHolder, false);
return txObject;
}
// 检查当前事务是否active
@Override
protected boolean isExistingTransaction(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
}
}
可以看到,获取当前事务对象时,使用了TransactionSynchronizationManager#getResource方法,类图如下:
TransactionSynchronizationManager通过ThreadLocal对象在当前线程记录了resources和synchronizations属性。resources是一个HashMap,用于记录当前参与事务的事务资源,方便进行事务同步,在DataSourceTransactionManager的例子中就是以dataSource作为key,保存了数据库连接,这样在同一个线程中,不同的方法调用就可以通过dataSource获取相同的数据库连接,从而保证所有操作在一个事务中进行。synchronizations属性是一个TransactionSynchronization对象的集合,AbstractPlatformTransactionManager类中定义了事务操作各个阶段的调用流程,以事务提交为例:
public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
try {
boolean beforeCompletionInvoked = false;
try {
boolean unexpectedRollback = false;
prepareForCommit(status);
triggerBeforeCommit(status);
triggerBeforeCompletion(status);
beforeCompletionInvoked = true;
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();
}
// Throw UnexpectedRollbackException if we have a global rollback-only
// marker but still didn't get a corresponding exception from commit.
if (unexpectedRollback) {
throw new UnexpectedRollbackException(
"Transaction silently rolled back because it has been marked as rollback-only");
}
}
catch (UnexpectedRollbackException ex) {
// can only be caused by doCommit
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
throw ex;
}
catch (TransactionException ex) {
// can only be caused by doCommit
if (isRollbackOnCommitFailure()) {
// 提交异常则回滚
doRollbackOnCommitException(status, ex);
}
else {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
}
throw ex;
}
catch (RuntimeException | Error ex) {
if (!beforeCompletionInvoked) {
triggerBeforeCompletion(status);
}
doRollbackOnCommitException(status, ex);
throw ex;
}
// Trigger afterCommit callbacks, with an exception thrown there
// propagated to callers but the transaction still considered as committed.
try {
triggerAfterCommit(status);
}
finally {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
}
}
finally {
// 清理事务信息
cleanupAfterCompletion(status);
}
}
}
我们可以看到,有很多trigger前缀的方法,这些方法用于在事务操作的各个阶段触发回调,从而可以精确控制在事务执行的不同阶段所要执行的操作,这些回调实际上都通过TransactionSynchronizationUtils来实现,它会遍历TransactionSynchronizationManager#synchronizations集合中的TransactionSynchronization对象,然后分别触发集合中各个元素对应方法的调用。例如:
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
@Override
public void afterCommit() {
// do something after commit
}
});
这段代码就在当前线程的事务synchronizations
属性中,添加了一个自定义同步类,如果当前存在事务,那么在事务管理器执行事务提交之后,就会触发afterCommit
方法,可以通过这种方式在事务执行的不同阶段自定义一些操作。
到这里,我们已经对Spring事务的实现原理和处理流程有了一定的了解。
小结
在声明式的事务处理中,主要有以下几个处理步骤:
- 获取事务的属性:tas.getTransactionAttribute(method, targetClass)
- 加载配置中配置的TransactionManager**:determineTransactionManager(txAttr);
- 在目标方法执行前获取事务并收集事务信息:createTransactionIfNecessary(tm, txAttr, joinpointIdentification)
- 执行目标方法:invocation.proceed()
- 出现异常,尝试异常处理:completeTransactionAfterThrowing(txInfo, ex);
- 提交事务前的事务信息消除:cleanupTransactionInfo(txInfo)
- 提交事务:commitTransactionAfterReturning(txInfo)
参考:
https://zhuanlan.zhihu.com/p/54067384
https://segmentfault.com/a/1190000022754620
网友评论