• 【spring】事务底层的实现流程


    事务简单介绍

    • 本文源码基于spring-framework-5.3.10。
    • 事务是基于AOP的机制进行实现的!

    Spring事务基本执行原理

    • 一个Bean在执行Bean的创建生命周期时,会经过InfrastructureAdvisorAutoProxyCreator的初始化后的方法,会判断当前当前Bean对象是否和BeanFactoryTransactionAttributeSourceAdvisor匹配,匹配逻辑为判断该Bean的类上是否存在@Transactional注解,或者类中的某个方法上是否存在@Transactional注解,如果存在则表示该Bean需要进行动态代理产生一个代理对象作为Bean对象。

    Spring事务基本执行流程

    • 利用所配置的PlatformTransactionManager事务管理器新建一个数据库连接。
    • 修改数据库连接的autocommit为false。
    • 执行MethodInvocation.proceed()方法,简单理解就是执行业务方法,其中就会执行sql。
    • 如果没有抛异常,则提交。
    • 如果抛了异常,则回滚。

    源码执行流程

    • 加了@Transactional注解的类,或者类中拥有@Transactional注解的方法,都会生成代理对象作为bean。
    • 代理对象执行方法时。
    • 获取当前正在执行的方法上的@Transactional注解的信息TransactionAttribute。
    • 查看@Transactional注解上是否指定了TransactionManager,如果没有指定,则默认获取TransactionManager类型的bean作为TransactionManager。
    • 对于TransactionManager有一个限制,必须是PlatformTransactionManager。
    • 生成一个joinpointIdentification,作为事务的名字。
    • 开始创建事务。
    • 创建事务成功后执行业务方法。
    • 如果执行业务方法出现异常,则会进行回滚,然后执行完finally中的方法后再将异常抛出。
    • 如果执行业务方法没有出现异常,那么则会执行完finally中的方法后再进行提交。

    创建事务源码流程

    • 得到一个TransactionStatus对象、
    • 将PlatformTransactionManager、TransactionAttribute、TransactionStatus构造成为一个TransactionInfo对象,并返回TransactionInfo对象。

    回滚事务源码流程

    • 判断当前异常是否需要回滚。不需要回滚直接走提价的流程。
    • 触发同步器的beforeCompletion()。
    • 调用数据库连接对象的rollback()。
    • 触发同步器的afterCompletion()。
    • 判断是否有事务挂起。
    • 如果有则把挂起的事务重新设置到TransactionSynchronizationManager中去,并执行同步器的resume()方法。

    提交事务源码流程

    • 触发同步器的beforeCommit。
    • 触发同步器的beforeCompletion()。
    • 调用数据库连接对象的commit()。
    • 触发同步器的afterCommit。
    • 触发同步器的afterCompletion()。
    • 判断是否有事务挂起。
    • 如果有则把挂起的事务重新设置到TransactionSynchronizationManager中去,并执行同步器的resume()方法。

    创建TransactionStatus的源码流程

    • 调用AbstractPlatformTransactionManager类中的getTransaction(txAttr)方法,实际上这个方法就是真正去开启事务的方法。
    • 调用DataSourceTransactionManager中的doGetTransaction()得到一个事务对象,得到的事务对象中可能持有也可能没有持有数据库连接对象。不同船舶机制下,是否持有事务,逻辑不同!
    传播机制 含义 之前方法持有事务的逻辑 之前方法未持有事务的逻辑
    REQUIRED(TransactionDefinition.PROPAGATION_REQUIRED) 支持当前事务,如果没有事务会创建一个新的事务 在当前事务运行 创建一个新的事务
    SUPPORTS(TransactionDefinition.PROPAGATION_SUPPORTS) 支持当前事务,如果没有事务的话以非事务方式执行 在当前事务运行 使用非事务的方式运行
    MANDATORY(TransactionDefinition.PROPAGATION_MANDATORY) 支持当前事务,如果没有事务抛出异常 在当前事务运行 抛异常
    REQUIRES_NEW(TransactionDefinition.PROPAGATION_REQUIRES_NEW) 创建一个新的事务并挂起当前事务 创建一个新的事务并挂起当前事务 创建一个新的事务
    NOT_SUPPORTED(TransactionDefinition.PROPAGATION_NOT_SUPPORTED) 以非事务方式执行,如果当前存在事务则将当前事务挂起 挂起当前事务 使用非事务的方式运行
    NEVER(TransactionDefinition.PROPAGATION_NEVER) 以非事务方式进行,如果存在事务则抛出异常 抛异常 使用非事务的方式运行
    NESTED(TransactionDefinition.PROPAGATION_NESTED) 如果当前存在事务,则在嵌套事务内执行。如果当前没有事务,则创建一个新事务 利用数据库连接对象,设置一个savepoint,比如mysql就支持,在一个事务中,可以在某个位置设置一个savepoint,后续可以只回滚到某个savepoint 创建一个新的事务

    doBegin源码流程

    • 如果事务对象中没有持有数据库连接对象,那么则调用DataSource获取一个数据库连接对象,并设置到事务对象中去
    • 设置当前数据库连接的隔离级别。
    • 设置数据库连接的autoCommit为false。
    • 设置数据库连接的timeout。
    • 把获得的数据库连接对象通过TransactionSynchronizationManager设置到当前线程的ThreadLocal中。

    同步器使用方式

    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                System.out.println("after commit...");
            }
    }
    
    

    开启事务的注解@EnableTransactionManagement源码分析

    /**
     * 这个注解导入了TransactionManagementConfigurationSelector类
     */
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Import(TransactionManagementConfigurationSelector.class)
    public @interface EnableTransactionManagement {
    
    	boolean proxyTargetClass() default false;
    
    	AdviceMode mode() default AdviceMode.PROXY;
    
    	int order() default Ordered.LOWEST_PRECEDENCE;
    }
    

    导入的TransactionManagementConfigurationSelector源码分析

    /**
     * 在调用process方法的时候会调用到这里
     */
    protected String[] selectImports(AdviceMode adviceMode) {
    	switch (adviceMode) {
    		case PROXY:
    			// 默认是PROXY。往Spring容器中添加了两个Bean:AutoProxyRegistrar、ProxyTransactionManagementConfiguration
    			return new String[] {AutoProxyRegistrar.class.getName(),
    					ProxyTransactionManagementConfiguration.class.getName()};
    		case ASPECTJ:
    			// 表示不用动态代理技术,用ASPECTJ技术,比较麻烦了
    			return new String[] {determineTransactionAspectClass()};
    		default:
    			return null;
    	}
    }
    

    AutoProxyRegistrar的Bean源码分析

    /**
     * 这个类实现了ImportBeanDefinitionRegistrar,他在启动的时候会调用registerBeanDefinitions方法。
     * 最核心的逻辑是往Spring容器中注册了一个InfrastructureAdvisorAutoProxyCreator的Bean。
     * InfrastructureAdvisorAutoProxyCreator继承了AbstractAdvisorAutoProxyCreator,所以这个类的主要作用就是开启自动代理的作用,也就是一个BeanPostProcessor,会在初始化后步骤中去寻找Advisor类型的Bean,并判断当前某个Bean是否有匹配的Advisor,是否需要利用动态代理产生一个代理对象。
     */
    public class AutoProxyRegistrar implements ImportBeanDefinitionRegistrar {
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
    		boolean candidateFound = false;
    		Set<String> annTypes = importingClassMetadata.getAnnotationTypes();
    		for (String annType : annTypes) {
    			AnnotationAttributes candidate = AnnotationConfigUtils.attributesFor(importingClassMetadata, annType);
    			if (candidate == null) {
    				continue;
    			}
    			Object mode = candidate.get("mode");
    			Object proxyTargetClass = candidate.get("proxyTargetClass");
    			if (mode != null && proxyTargetClass != null && AdviceMode.class == mode.getClass() &&
    					Boolean.class == proxyTargetClass.getClass()) {
    				candidateFound = true;
    				if (mode == AdviceMode.PROXY) {
    					// 注册InfrastructureAdvisorAutoProxyCreator,才可以Bean进行AOP
    					AopConfigUtils.registerAutoProxyCreatorIfNecessary(registry);
    					if ((Boolean) proxyTargetClass) {
    						// 设置InfrastructureAdvisorAutoProxyCreator的proxyTargetClass为true
    						AopConfigUtils.forceAutoProxyCreatorToUseClassProxying(registry);
    						return;
    					}
    				}
    			}
    		}
    		if (!candidateFound && logger.isInfoEnabled()) {
    			String name = getClass().getSimpleName();
    			logger.info(String.format("%s was imported but no annotations were found " +
    					"having both 'mode' and 'proxyTargetClass' attributes of type " +
    					"AdviceMode and boolean respectively. This means that auto proxy " +
    					"creator registration and configuration may not have occurred as " +
    					"intended, and components may not be proxied as expected. Check to " +
    					"ensure that %s has been @Import'ed on the same class where these " +
    					"annotations are declared; otherwise remove the import of %s " +
    					"altogether.", name, name, name));
    		}
    	}
    }
    

    导入的ProxyTransactionManagementConfiguration源码分析

    /**
     * ProxyTransactionManagementConfiguration是一个配置类,它又定义了另外三个bean:
     * BeanFactoryTransactionAttributeSourceAdvisor:一个Advisor
     * AnnotationTransactionAttributeSource:相当于BeanFactoryTransactionAttributeSourceAdvisor中的Pointcut
     * TransactionInterceptor:相当于BeanFactoryTransactionAttributeSourceAdvisor中的Advice
     */
    @Configuration(proxyBeanMethods = false)
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {
    
    	// 定义一个Advisor的Bean
    	@Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
    	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    	public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(
    			TransactionAttributeSource transactionAttributeSource, TransactionInterceptor transactionInterceptor) {
    
    		BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
    		advisor.setTransactionAttributeSource(transactionAttributeSource);
    		advisor.setAdvice(transactionInterceptor);
    		if (this.enableTx != null) {
    			advisor.setOrder(this.enableTx.<Integer>getNumber("order"));
    		}
    		return advisor;
    	}
    
    	// 定义Advisor的Pointcut
    	@Bean
    	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    	public TransactionAttributeSource transactionAttributeSource() {
    		// AnnotationTransactionAttributeSource中定义了一个Pointcut
    		// 并且AnnotationTransactionAttributeSource可以用来解析@Transactional注解,并得到一个RuleBasedTransactionAttribute对象
    		return new AnnotationTransactionAttributeSource();
    	}
    
    	// 定义Advisor的Advice
    	// 开启事务、回滚、提交都在这个TransactionInterceptor中
    	@Bean
    	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    	public TransactionInterceptor transactionInterceptor(TransactionAttributeSource transactionAttributeSource) {
    		TransactionInterceptor interceptor = new TransactionInterceptor();
    		interceptor.setTransactionAttributeSource(transactionAttributeSource);
    		if (this.txManager != null) {
    			interceptor.setTransactionManager(this.txManager);
    		}
    		return interceptor;
    	}
    
    }
    

    使用用到的Pointcut源码分析

    public class BeanFactoryTransactionAttributeSourceAdvisor extends AbstractBeanFactoryPointcutAdvisor {
    
    	/**
    	 * pointcut就是一个new TransactionAttributeSourcePointcut对象
    	 */
    	private final TransactionAttributeSourcePointcut pointcut = new TransactionAttributeSourcePointcut() {
    		@Override
    		@Nullable
    		protected TransactionAttributeSource getTransactionAttributeSource() {
    			return transactionAttributeSource;
    		}
    	};
    
    	/**
    	 * 获取pointcut采用的是内部类的方式构建
    	 */
    	@Override
    	public Pointcut getPointcut() {
    		return this.pointcut;
    	}
    
    }
    
    /**
     * 构建Pointcut用到的类
     */
    abstract class TransactionAttributeSourcePointcut extends StaticMethodMatcherPointcut implements Serializable {
    
    	// 外部构造方法调用到这里,其实就是设置new TransactionAttributeSourceClassFilter()
    	// 这里面主要进行类的判断
    	protected TransactionAttributeSourcePointcut() {
    		setClassFilter(new TransactionAttributeSourceClassFilter());
    	}
    
    	// 判断方法是否匹配
    	@Override
    	public boolean matches(Method method, Class<?> targetClass) {
    		// 调用外部重写的方法
    		TransactionAttributeSource tas = getTransactionAttributeSource();
    		return (tas == null || tas.getTransactionAttribute(method, targetClass) != null);
    	}
    
    	// 具体的匹配逻辑
    	private class TransactionAttributeSourceClassFilter implements ClassFilter {
    
    		@Override
    		public boolean matches(Class<?> clazz) {
    			// 事务内部的一些类,直接返回false
    			if (TransactionalProxy.class.isAssignableFrom(clazz) ||
    					TransactionManager.class.isAssignableFrom(clazz) ||
    					PersistenceExceptionTranslator.class.isAssignableFrom(clazz)) {
    				return false;
    			}
    			// 调用外部重写的方法
    			TransactionAttributeSource tas = getTransactionAttributeSource();
    			// 判断有没有@Transaction注解
    			return (tas == null || tas.isCandidateClass(clazz));
    		}
    	}
    }
    
    /**
     * 源码位置:org.springframework.transaction.annotation.AnnotationTransactionAttributeSource.isCandidateClass(Class<?>)
     * 判断是否能成为候选者
     */
    public boolean isCandidateClass(Class<?> targetClass) {
    	for (TransactionAnnotationParser parser : this.annotationParsers) {
    		if (parser.isCandidateClass(targetClass)) {
    			return true;
    		}
    	}
    	return false;
    }
    
    /**
     * 源码位置:org.springframework.transaction.interceptor.AbstractFallbackTransactionAttributeSource.getTransactionAttribute(Method, Class<?>)
     * 判断当前类上面是否有@Transactional注解
     */
    public boolean isCandidateClass(Class<?> targetClass) {
    	return AnnotationUtils.isCandidateClass(targetClass, Transactional.class);
    }
    
    /**
     * 源码位置:org.springframework.transaction.annotation.SpringTransactionAnnotationParser.isCandidateClass(Class<?>)
     * 获取方法上或者类上是否有@Transaction注解
     */
    public TransactionAttribute getTransactionAttribute(Method method, @Nullable Class<?> targetClass) {
    	// Object类直接返回null
    	if (method.getDeclaringClass() == Object.class) {
    		return null;
    	}
    
    	// First, see if we have a cached value.
    	// 检查缓存里的结果,缓存里存了当前类和方法是否存在Transactional注解
    	Object cacheKey = getCacheKey(method, targetClass);
    	TransactionAttribute cached = this.attributeCache.get(cacheKey);
    	if (cached != null) {
    		// Value will either be canonical value indicating there is no transaction attribute,
    		// or an actual transaction attribute.
    		// 缓存中没有,直接返回null
    		if (cached == NULL_TRANSACTION_ATTRIBUTE) {
    			return null;
    		}
    		else {
    			return cached;
    		}
    	}
    	else {
    		// We need to work it out.
    		// 解析。实际对象为RuleBasedTransactionAttribute
    		TransactionAttribute txAttr = computeTransactionAttribute(method, targetClass);
    		// Put it in the cache.
    		// 为空,缓存一个空的
    		if (txAttr == null) {
    			this.attributeCache.put(cacheKey, NULL_TRANSACTION_ATTRIBUTE);
    		}
    		else {
    			String methodIdentification = ClassUtils.getQualifiedMethodName(method, targetClass);
    			if (txAttr instanceof DefaultTransactionAttribute) {
    				DefaultTransactionAttribute dta = (DefaultTransactionAttribute) txAttr;
    				dta.setDescriptor(methodIdentification);
    				dta.resolveAttributeStrings(this.embeddedValueResolver);
    			}
    			if (logger.isTraceEnabled()) {
    				logger.trace("Adding transactional method '" + methodIdentification + "' with attribute: " + txAttr);
    			}
    			// 不为空,缓存结果
    			this.attributeCache.put(cacheKey, txAttr);
    		}
    		return txAttr;
    	}
    }
    

    最终执行的Advisor:TransactionInterceptor源码分析

    /**
     * 源码位置:org.springframework.transaction.interceptor.TransactionInterceptor.invoke(MethodInvocation)
     */
    public Object invoke(MethodInvocation invocation) throws Throwable {
    	// Work out the target class: may be {@code null}.
    	// The TransactionAttributeSource should be passed the target class
    	// as well as the method, which may be from an interface.
    	// 获取代理类
    	Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
    
    	// Adapt to TransactionAspectSupport's invokeWithinTransaction...
    	return invokeWithinTransaction(invocation.getMethod(), targetClass, new CoroutinesInvocationCallback() {
    		@Override
    		@Nullable
    		public Object proceedWithInvocation() throws Throwable {
    			// 执行后续的Interceptor,以及被代理的方法
    			return invocation.proceed(); // test() sql
    		}
    		@Override
    		public Object getTarget() {
    			return invocation.getThis();
    		}
    		@Override
    		public Object[] getArguments() {
    			return invocation.getArguments();
    		}
    	});
    }
    
    /**
     * 具体的执行代码
     */
    protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
    		final InvocationCallback invocation) throws Throwable {
    
    	// If the transaction attribute is null, the method is non-transactional.
    	// TransactionAttribute就是@Transactional中的配置
    	TransactionAttributeSource tas = getTransactionAttributeSource();
    	// 获取@Transactional注解中的属性值
    	final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
    
    	// 返回Spring容器中类型为TransactionManager的Bean对象。事务的开启,提交,回滚都会用到TransactionManager对象。
    	final TransactionManager tm = determineTransactionManager(txAttr);
    
    	// ReactiveTransactionManager用得少,并且它只是执行方式是响应式的,原理流程和普通的是一样的
    	if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {
    		boolean isSuspendingFunction = KotlinDetector.isSuspendingFunction(method);
    		boolean hasSuspendingFlowReturnType = isSuspendingFunction &&
    				COROUTINES_FLOW_CLASS_NAME.equals(new MethodParameter(method, -1).getParameterType().getName());
    		if (isSuspendingFunction && !(invocation instanceof CoroutinesInvocationCallback)) {
    			throw new IllegalStateException("Coroutines invocation not supported: " + method);
    		}
    		CoroutinesInvocationCallback corInv = (isSuspendingFunction ? (CoroutinesInvocationCallback) invocation : null);
    
    		ReactiveTransactionSupport txSupport = this.transactionSupportCache.computeIfAbsent(method, key -> {
    			Class<?> reactiveType =
    					(isSuspendingFunction ? (hasSuspendingFlowReturnType ? Flux.class : Mono.class) : method.getReturnType());
    			ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(reactiveType);
    			if (adapter == null) {
    				throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " +
    						method.getReturnType());
    			}
    			return new ReactiveTransactionSupport(adapter);
    		});
    
    		InvocationCallback callback = invocation;
    		if (corInv != null) {
    			callback = () -> CoroutinesUtils.invokeSuspendingFunction(method, corInv.getTarget(), corInv.getArguments());
    		}
    		Object result = txSupport.invokeWithinTransaction(method, targetClass, callback, txAttr, (ReactiveTransactionManager) tm);
    		if (corInv != null) {
    			Publisher<?> pr = (Publisher<?>) result;
    			return (hasSuspendingFlowReturnType ? KotlinDelegate.asFlow(pr) :
    					KotlinDelegate.awaitSingleOrNull(pr, corInv.getContinuation()));
    		}
    		return result;
    	}
    
    	// 把tm强制转换为PlatformTransactionManager,所以我们在定义时得定义PlatformTransactionManager类型
    	PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
    
    	// joinpoint的唯一标识,就是当前在执行的方法名字
    	final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
    
    	// CallbackPreferringPlatformTransactionManager表示拥有回调功能的PlatformTransactionManager,也不常用
    	if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
    		// Standard transaction demarcation with getTransaction and commit/rollback calls.
    		// 如果有必要就创建事务,这里就涉及到事务传播机制的实现了
    		// TransactionInfo表示一个逻辑事务,比如两个逻辑事务属于同一个物理事务
    		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.
    			// 执行下一个Interceptor或被代理对象中的方法
    			retVal = invocation.proceedWithInvocation(); //test
    		}
    		catch (Throwable ex) {
    			// target invocation exception
    			// 抛异常了,则回滚事务
    			completeTransactionAfterThrowing(txInfo, ex);
    			throw ex;
    		}
    		finally {
    			cleanupTransactionInfo(txInfo);
    		}
    
    		if (retVal != null && 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;
    	}
    
    	else {
    		Object result;
    		final ThrowableHolder throwableHolder = new ThrowableHolder();
    
    		// It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
    		try {
    			result = ((CallbackPreferringPlatformTransactionManager) ptm).execute(txAttr, status -> {
    				TransactionInfo txInfo = prepareTransactionInfo(ptm, txAttr, joinpointIdentification, status);
    				try {
    					Object retVal = invocation.proceedWithInvocation();
    					if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
    						// Set rollback-only in case of Vavr failure matching our rollback rules...
    						retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
    					}
    					return retVal;
    				}
    				catch (Throwable ex) {
    					if (txAttr.rollbackOn(ex)) {
    						// A RuntimeException: will lead to a rollback.
    						if (ex instanceof RuntimeException) {
    							throw (RuntimeException) ex;
    						}
    						else {
    							throw new ThrowableHolderException(ex);
    						}
    					}
    					else {
    						// A normal return value: will lead to a commit.
    						throwableHolder.throwable = ex;
    						return null;
    					}
    				}
    				finally {
    					cleanupTransactionInfo(txInfo);
    				}
    			});
    		}
    		catch (ThrowableHolderException ex) {
    			throw ex.getCause();
    		}
    		catch (TransactionSystemException ex2) {
    			if (throwableHolder.throwable != null) {
    				logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
    				ex2.initApplicationException(throwableHolder.throwable);
    			}
    			throw ex2;
    		}
    		catch (Throwable ex2) {
    			if (throwableHolder.throwable != null) {
    				logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
    			}
    			throw ex2;
    		}
    
    		// Check result state: It might indicate a Throwable to rethrow.
    		if (throwableHolder.throwable != null) {
    			throw throwableHolder.throwable;
    		}
    		return result;
    	}
    }
    
    /**
     * 创建事务的逻辑
     */
    protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
    		@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
    
    	// If no name specified, apply method identification as transaction name.
    	// name为空,取方法名字
    	if (txAttr != null && txAttr.getName() == null) {
    		txAttr = new DelegatingTransactionAttribute(txAttr) {
    			@Override
    			public String getName() {
    				return joinpointIdentification;
    			}
    		};
    	}
    
    	// 每个逻辑事务都会创建一个TransactionStatus,但是TransactionStatus中有一个属性代表当前逻辑事务底层的物理事务是不是新的
    	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");
    			}
    		}
    	}
    
    	// 返回一个TransactionInfo对象,表示得到了一个事务,可能是新创建的一个事务,也可能是拿到的已有的事务
    	return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
    }
    

    获取(开启)事务的逻辑

    /**
     * 获取事务的逻辑
     * 源码位置:org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(TransactionDefinition)
     * definition:Transaction注解的信息的对象
     */
    public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
    		throws TransactionException {
    
    	// Use defaults if no transaction definition given.
    	TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
    
    	// 从ThreadLocal中拿到txObject对象
    	Object transaction = doGetTransaction();
    	boolean debugEnabled = logger.isDebugEnabled();
    
    	// transaction.getConnectionHolder().isTransactionActive()
    	// 判断是不是存在一个事务
    	if (isExistingTransaction(transaction)) {
    		// Existing transaction found -> check propagation behavior to find out how to behave.
    		return handleExistingTransaction(def, transaction, debugEnabled);
    	}
    
    	// Check definition settings for new transaction.
    	// 配置的数据库返回的超时时间小于-1,抛异常。
    	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'");
    	}
    	// 在当前Thread中没有事务的前提下,以下三个是等价的。
    	// PROPAGATION_REQUIRED :支持当前事务,如果没有事务会创建一个新的事务
    	// PROPAGATION_REQUIRES_NEW :创建一个新的事务并挂起当前事务
    	// PROPAGATION_NESTED:	如果当前存在事务,则在嵌套事务内执行。如果当前没有事务,则创建一个事务。
    	else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
    			def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
    			def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
    		// 没有事务需要挂起,不过TransactionSynchronization有可能需要挂起。直接调用TransactionSynchronizationManager.initSynchronization方法会有需要被挂起的。
    		// suspendedResources表示当前线程被挂起的资源持有对象(数据库连接、TransactionSynchronization)
    		SuspendedResourcesHolder suspendedResources = suspend(null);
    		if (debugEnabled) {
    			logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
    		}
    		try {
    			// 开启事务方法
    			// 开启事务后,transaction中就会有数据库连接了,并且isTransactionActive为true
    			// 并返回TransactionStatus对象,该对象保存了很多信息,包括被挂起的资源
    			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);
    	}
    }
    
    /**
     * 开启Transaction事务
     */
    private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
    		boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
    
    	// 是否开启一个新的TransactionSynchronization
    	boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    
    	// 开启的这个事务的状态信息:
    	// 事务的定义、用来保存数据库连接的对象、是否是新事务,是否是新的TransactionSynchronization
    	DefaultTransactionStatus status = newTransactionStatus(
    			definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
    
    	// 开启事务
    	doBegin(transaction, definition);
    
    	// 如果需要新开一个TransactionSynchronization,就把新创建的事务的一些状态信息设置到TransactionSynchronizationManager中
    	prepareSynchronization(status, definition);
    	return status;
    }
    
    /**
     * 开启事务!DataSource层面
     * 源码位置:org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(Object, TransactionDefinition)
     */
    protected void doBegin(Object transaction, TransactionDefinition definition) {
    	DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
    	Connection con = null;
    
    	try {
    
    		// 如果当前线程中所使用的DataSource还没有创建过数据库连接,就获取一个新的数据库连接
    		if (!txObject.hasConnectionHolder() ||
    				txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
    			// 得到连接对象
    			Connection newCon = obtainDataSource().getConnection();
    			if (logger.isDebugEnabled()) {
    				logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
    			}
    			// 设置到DataSourceTransactionObject。注意这里设置的true,表示txObject的链接一个新的
    			txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
    		}
    
    		txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
    		// 得到连接对象
    		con = txObject.getConnectionHolder().getConnection();
    
    		// 根据@Transactional注解中的设置,设置Connection的readOnly与隔离级别
    		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).
    		// 保证autocommit是false。autocommit为true的时候设置autocommit为false
    		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.
    		// 把新建的数据库连接设置到resources中,resources就是一个ThreadLocal<Map<Object, Object>>,事务管理器中的设置的DataSource对象为key,数据库连接对象为value
    		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.support.AbstractPlatformTransactionManager.suspend(Object)
     */
    protected final SuspendedResourcesHolder suspend(@Nullable Object transaction) throws TransactionException {
    	// synchronizations是一个ThreadLocal<Set<TransactionSynchronization>>
    	// 我们可以在任何地方通过TransactionSynchronizationManager给当前线程添加TransactionSynchronization,
    
    	// 这里判断有没有开启事务。在prepareSynchronization方法中开启!
    	if (TransactionSynchronizationManager.isSynchronizationActive()) {
    		// 调用TransactionSynchronization的suspend方法,并清空和返回当前线程中所有的TransactionSynchronization对象
    		List<TransactionSynchronization> suspendedSynchronizations = doSuspendSynchronization();
    		try {
    			Object suspendedResources = null;
    			if (transaction != null) {
    				// 挂起事务,把transaction中的Connection清空,并把resources中的key-value进行移除,并返回数据库连接Connection对象
    				suspendedResources = doSuspend(transaction);
    			}
    
    			// 获取并清空当前线程中关于TransactionSynchronizationManager的设置
    			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);
    
    			// 将当前线程中的数据库连接对象、TransactionSynchronization对象、TransactionSynchronizationManager中的设置构造成一个对象
    			// 表示被挂起的资源持有对象,持有了当前线程中的事务对象、TransactionSynchronization对象
    			// suspendedResources数据库连接、suspendedSynchronizations自己定义的同步器、name事务的名称、readOnly事务是不是只读、isolationLevel事务隔离级别、wasActive切面的Active
    			return new SuspendedResourcesHolder(
    					suspendedResources, suspendedSynchronizations, name, readOnly, isolationLevel, wasActive);
    		}
    		catch (RuntimeException | Error ex) {
    			// doSuspend failed - original transaction is still active...
    			doResumeSynchronization(suspendedSynchronizations);
    			throw ex;
    		}
    	}
    	else if (transaction != null) {
    		// Transaction active but no synchronization active.
    		Object suspendedResources = doSuspend(transaction);
    		return new SuspendedResourcesHolder(suspendedResources);
    	}
    	else {
    		// Neither transaction nor synchronization active.
    		return null;
    	}
    }
    
    /**
     * 调用TransactionSynchronization的suspend方法,并清空和返回当前线程中所有的TransactionSynchronization对象
     */
    private List<TransactionSynchronization> doSuspendSynchronization() {
    	// 从synchronizations(一个ThreadLocal)中拿到所设置的TransactionSynchronization对象
    	List<TransactionSynchronization> suspendedSynchronizations =
    			TransactionSynchronizationManager.getSynchronizations();
    
    	// 调用TransactionSynchronization对象的suspend()
    	for (TransactionSynchronization synchronization : suspendedSynchronizations) {
    		synchronization.suspend();
    	}
    
    	// 清空synchronizations
    	TransactionSynchronizationManager.clearSynchronization();
    
    	// 把获取到的TransactionSynchronization返回
    	return suspendedSynchronizations;
    }
    

    之前存在事务的执行逻辑

    /**
     * 之前存在事务的执行逻辑
     * 源码位置:org.springframework.transaction.support.AbstractPlatformTransactionManager.handleExistingTransaction(TransactionDefinition, Object, boolean)
     */
    private TransactionStatus handleExistingTransaction(
    		TransactionDefinition definition, Object transaction, boolean debugEnabled)
    		throws TransactionException {
    
    	// PROPAGATION_NEVER:以非事务方式进行,如果存在事务则抛出异常
    	if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
    		throw new IllegalTransactionStateException(
    				"Existing transaction found for transaction marked with propagation 'never'");
    	}
    
    	// PROPAGATION_NOT_SUPPORTED:以非事务方式执行,如果当前存在事务则将当前事务挂起
    	if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
    		if (debugEnabled) {
    			logger.debug("Suspending current transaction");
    		}
    		// 把当前事务挂起,其中就会把数据库连接对象从ThreadLocal中移除
    		Object suspendedResources = suspend(transaction);
    		boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
    		return prepareTransactionStatus(
    				definition, null, false, newSynchronization, debugEnabled, suspendedResources);
    	}
    
    	// PROPAGATION_REQUIRES_NEW:创建一个新的事务并挂起当前事务
    	if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
    		if (debugEnabled) {
    			logger.debug("Suspending current transaction, creating new transaction with name [" +
    					definition.getName() + "]");
    		}
    		// 调用挂起的逻辑
    		SuspendedResourcesHolder suspendedResources = suspend(transaction);
    
    		// 开启新事务的逻辑
    		try {
    			return startTransaction(definition, transaction, debugEnabled, suspendedResources);
    		}
    		catch (RuntimeException | Error beginEx) {
    			resumeAfterBeginException(transaction, suspendedResources, beginEx);
    			throw beginEx;
    		}
    	}
    
    	// PROPAGATION_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() + "]");
    		}
    		if (useSavepointForNestedTransaction()) {
    			// Create savepoint within existing Spring-managed transaction,
    			// through the SavepointManager API implemented by TransactionStatus.
    			// Usually uses JDBC 3.0 savepoints. Never activates Spring synchronization.
    			DefaultTransactionStatus status =
    					prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
    			// 创建一个savepoint
    			status.createAndHoldSavepoint();
    			return status;
    		}
    		else {
    			// Nested transaction through nested begin and commit/rollback calls.
    			// Usually only for JTA: Spring synchronization might get activated here
    			// in case of a pre-existing JTA transaction.
    			return startTransaction(definition, transaction, debugEnabled, null);
    		}
    	}
    
    	// Assumably PROPAGATION_SUPPORTS or PROPAGATION_REQUIRED.
    	if (debugEnabled) {
    		logger.debug("Participating in existing transaction");
    	}
    	if (isValidateExistingTransaction()) {
    		if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
    			Integer currentIsolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
    			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 existing transaction: " +
    						(currentIsolationLevel != 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");
    			}
    		}
    	}
    
    	// 如果依然是Propagation.REQUIRED
    	boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
    	return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
    }
    

    事务的提交逻辑

    /**
     * 事务的提交
     */
    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());
    	}
    }
    
    /**
     * 事务提交的准备逻辑
     */
    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;
    
    	// 可以通过TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();来设置
    	// 事务本来是可以要提交的,但是可以强制回滚。比如报错后更有好的提示。
    	if (defStatus.isLocalRollbackOnly()) {
    		if (defStatus.isDebug()) {
    			logger.debug("Transactional code has requested rollback");
    		}
    		processRollback(defStatus, false);
    		return;
    	}
    
    	// 判断此事务在之前是否设置了需要回滚,跟globalRollbackOnParticipationFailure有关
    	if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
    		if (defStatus.isDebug()) {
    			logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");
    		}
    		processRollback(defStatus, true);
    		return;
    	}
    
    	// 提交
    	processCommit(defStatus);
    }
    
    /**
     * 提交外部逻辑
     */
    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);
    	}
    }
    
    /**
     * 提交的核心逻辑,直接调用Connection的提交方法
     */
    protected void doCommit(DefaultTransactionStatus status) {
    	DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction();
    	Connection con = txObject.getConnectionHolder().getConnection();
    	if (status.isDebug()) {
    		logger.debug("Committing JDBC transaction on Connection [" + con + "]");
    	}
    	try {
    		con.commit();
    	}
    	catch (SQLException ex) {
    		throw translateException("JDBC commit", ex);
    	}
    }
    
    /**
     * 恢复被挂起的资源到当前线程中
     */
    private void cleanupAfterCompletion(DefaultTransactionStatus status) {
    	status.setCompleted();
    	if (status.isNewSynchronization()) {
    		TransactionSynchronizationManager.clear();
    	}
    	// 判断当前事务执行的方法,是不是创建这个事务的方法
    	if (status.isNewTransaction()) {
    		// 这里会去关闭数据库连接
    		doCleanupAfterCompletion(status.getTransaction());
    	}
    
    	// 恢复被挂起的资源到当前线程中
    	if (status.getSuspendedResources() != null) {
    		if (status.isDebug()) {
    			logger.debug("Resuming suspended transaction after completion of inner transaction");
    		}
    		Object transaction = (status.hasTransaction() ? status.getTransaction() : null);
    		// 恢复
    		resume(transaction, (SuspendedResourcesHolder) status.getSuspendedResources());
    	}
    }
    

    回滚事务的逻辑

    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);
    		}
    
    		// transactionAttribute的实现类为RuleBasedTransactionAttribute,父类为DefaultTransactionAttribute
    		// 判断配置的rollBackFor的异常信息
    		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 {
    			// We don't roll back on this exception.
    			// Will still roll back if TransactionStatus.isRollbackOnly() is true.
    			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;
    			}
    		}
    	}
    }
    
    /**
     * 判断回滚条件是否满足
     * 源码位置:org.springframework.transaction.interceptor.RuleBasedTransactionAttribute.rollbackOn(Throwable)
     */
    public boolean rollbackOn(Throwable ex) {
    	RollbackRuleAttribute winner = null;
    	int deepest = Integer.MAX_VALUE;
    
    	if (this.rollbackRules != null) {
    		// 遍历所有的RollbackRuleAttribute,判断现在抛出的异常ex是否匹配RollbackRuleAttribute中指定的异常类型的子类或本身
    		for (RollbackRuleAttribute rule : this.rollbackRules) {
    			int depth = rule.getDepth(ex);
    			if (depth >= 0 && depth < deepest) {
    				deepest = depth;
    				winner = rule;
    			}
    		}
    	}
    
    	// User superclass behavior (rollback on unchecked) if no rule matches.
    	// 没有匹配的规则,调用父类判断是不是运行时异常
    	if (winner == null) {
    		return super.rollbackOn(ex);
    	}
    
    	// ex所匹配的RollbackRuleAttribute,可能是NoRollbackRuleAttribute,如果是匹配的NoRollbackRuleAttribute,那就表示现在这个异常ex不用回滚
    	return !(winner instanceof NoRollbackRuleAttribute);
    }
    
    /**
     * 回滚前的准备逻辑
     * 源码位置:org.springframework.transaction.support.AbstractPlatformTransactionManager.rollback(TransactionStatus)
     */
    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, false);
    }
    
    /**
     * 回滚
     */
    private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
    	try {
    		boolean unexpectedRollback = unexpected;
    
    		try {
    			// 只会触发完成前的同步器逻辑
    			triggerBeforeCompletion(status);
    
    			// 比如mysql中的savepoint
    			if (status.hasSavepoint()) {
    				if (status.isDebug()) {
    					logger.debug("Rolling back transaction to savepoint");
    				}
    				// 回滚到上一个savepoint位置
    				status.rollbackToHeldSavepoint();
    			}
    			else if (status.isNewTransaction()) {
    				if (status.isDebug()) {
    					logger.debug("Initiating transaction rollback");
    				}
    				// 如果当前执行的方法是新开了一个事务,那么就直接回滚
    				doRollback(status);
    			}
    			else {
    				// Participating in larger transaction
    				// 如果当前执行的方法,是公用了一个已存在的事务,而当前执行的方法抛了异常,则要判断整个事务到底要不要回滚,看具体配置
    				if (status.hasTransaction()) {
    
    					// 如果一个事务中有两个方法,第二个方法抛异常了,那么第二个方法就相当于执行失败需要回滚,如果globalRollbackOnParticipationFailure为true,那么第一个方法在没有抛异常的情况下也要回滚
    					if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
    						if (status.isDebug()) {
    							logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
    						}
    						// 直接将rollbackOnly设置到ConnectionHolder中去,表示整个事务的sql都要回滚
    						doSetRollbackOnly(status);
    					}
    					else {
    						if (status.isDebug()) {
    							logger.debug("Participating transaction failed - letting transaction originator decide on rollback");
    						}
    					}
    				}
    				else {
    					logger.debug("Should roll back transaction but cannot - no transaction available");
    				}
    				// Unexpected rollback only matters here if we're asked to fail early
    				if (!isFailEarlyOnGlobalRollbackOnly()) {
    					unexpectedRollback = false;
    				}
    			}
    		}
    		catch (RuntimeException | Error ex) {
    			triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
    			throw ex;
    		}
    
    		triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
    
    		// Raise UnexpectedRollbackException if we had a global rollback-only marker
    		if (unexpectedRollback) {
    			throw new UnexpectedRollbackException(
    					"Transaction rolled back because it has been marked as rollback-only");
    		}
    	}
    	finally {
    		cleanupAfterCompletion(status);
    	}
    }
    

    结束语

    • 获取更多本文的前置知识文章,以及新的有价值的文章,让我们一起成为架构师!
    • 关注公众号,可以让你对MySQL、并发编程、spring源码有深入的了解!
    • 关注公众号,后续持续高效的学习JVM!
    • 这个公众号,无广告!!!每日更新!!!
      作者公众号.jpg
  • 相关阅读:
    观察者模式
    IP风险查询:抵御DDoS攻击和CC攻击的关键一步
    PyCharm安装和使用教程
    数据库分析与优化
    生物活化量子点纳米载体/离子/功能化量子点修饰细胞膜/纳米颗粒修饰细胞膜荧光探针的制备
    天花板级别的python读取文件方法,真的香.......
    Apache OFBiz 路径遍历导致RCE漏洞复现(CVE-2024-36104)
    HTML下的点名
    分布式锁之防止超卖 --mysql原子操作,乐观锁,redis事务,乐观锁
    vue+iview中日期时间选择器不能选择当前日期之前包括时分秒
  • 原文地址:https://www.cnblogs.com/zfcq/p/15989393.html