Spring事务源码分析责任链事务链事务不生效

文章目录

前言
- 带着问题分析源码
事务源码分析
- 寻找Spring事务源码类
- TransactionInterceptor调用栈
- 分析Spring AOP责任链
- 分析TransactionInterceptor源码
- - TransactionAttribute
  - DataSourceTransactionManager
  - TransactionInfo
  - - TransactionStatus
    - TransactionInfo形成事务链
  - 执行事务方法
  - 执行事务提交
  - - Restore事务链
    - 提交事务
  - 事务提交完成后
总结
- 问题解答
- - Spring通过AOP管理事务
  - 多线程事务问题
- 关键类图
- 事务链

前言

带着问题分析源码

Spring是如何管理事务的
在多线程下事务会怎么样
什么情况下事务不生效

事务源码分析

寻找Spring事务源码类

简单讲一下如何快速分析事务源码其实很简单将断点打到service方法中查看方法调用栈即可
根据关键字 Transaction 找到TransactionInterceptor 即是源码分析入口

TransactionInterceptor调用栈

问题? 方法是如何一步一步执行到TransactionInterceptor的

从上图方法调用栈可知 UserService首先被SpringAOP代理
被Spring管理的对象一般情况下都会被代理 在Spring中对象被代理有两种方式

JDK 动态代理
Cglib 动态代理

调用代理对象的方法都会被SpringAOP拦截从而进入如下方法
org.springframework.aop.framework.CglibAopProxy.DynamicAdvisedInterceptor#intercept

public Object intercept(Object proxy, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {Object oldProxy = null;boolean setProxyContext = false;Object target = null;TargetSource targetSource = this.advised.getTargetSource();try {if (this.advised.exposeProxy) {// Make invocation available if necessary.oldProxy = AopContext.setCurrentProxy(proxy);setProxyContext = true;}// Get as late as possible to minimize the time we "own" the target, in case it comes from a pool...target = targetSource.getTarget();Class<?> targetClass = (target != null ? target.getClass() : null);// 获取代理对象的拦截器链 责任链模式List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);Object retVal;// Check whether we only have one InvokerInterceptor: that is,// no real advice, but just reflective invocation of the target.if (chain.isEmpty() && Modifier.isPublic(method.getModifiers())) {//如果目标方法没有拦截器链则直接执行目标方法// We can skip creating a MethodInvocation: just invoke the target directly.// Note that the final invoker must be an InvokerInterceptor, so we know// it does nothing but a reflective operation on the target, and no hot// swapping or fancy proxying.Object[] argsToUse = AopProxyUtils.adaptArgumentsIfNecessary(method, args);retVal = methodProxy.invoke(target, argsToUse);}else {//创建方法方法调用对象 //我们看到chain对象已经在构造方法的入参 是实现责任链调用的关键// We need to create a method invocation...retVal = new CglibMethodInvocation(proxy, target, method, args, targetClass, chain, methodProxy).proceed();}//获取方法调用结果retVal = processReturnType(proxy, target, method, retVal);return retVal;}finally {if (target != null && !targetSource.isStatic()) {targetSource.releaseTarget(target);}if (setProxyContext) {// Restore old proxy.AopContext.setCurrentProxy(oldProxy);}}
}

674行获取拦截器链责任链模式在各种框架中大量应用 Okhttp Netty Dubbo
- List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass)
- 我们先看返回结果
  org.springframework.transaction.interceptor.TransactionInterceptor

问题? TransactionInterceptor是怎么加入到AOP 拦截器链中的
答案在getInterceptorsAndDynamicInterceptionAdvice方法中我们来简单分析一下
org.springframework.aop.framework.DefaultAdvisorChainFactory#getInterceptorsAndDynamicInterceptionAdvice

调用栈如下

org.springframework.aop.framework.CglibAopProxy.DynamicAdvisedInterceptor#intercept
#674 org.springframework.aop.framework.AdvisedSupport#getInterceptorsAndDynamicInterceptionAdvice
##483 org.springframework.aop.framework.AdvisorChainFactory#getInterceptorsAndDynamicInterceptionAdvice
###    org.springframework.aop.framework.DefaultAdvisorChainFactory#getInterceptorsAndDynamicInterceptionAdvice

源码分析

//中间省略了部分源码
public List<Object> getInterceptorsAndDynamicInterceptionAdvice(Advised config, //AOP代理配置的抽象 org.springframework.transaction.interceptor.BeanFactoryTransactionAttributeSourceAdvisor Method method, //代理对象方法 public java.lang.Integer com.deanjo.service.UserServiceImpl.updateById(com.deanjo.share.param.UserModifyParam)@Nullable Class<?> targetClass) //代理目标类 com.deanjo.service.UserServiceImpl
{//具体的代理配置的实现//org.springframework.transaction.interceptor.BeanFactoryTransactionAttributeSourceAdvisor 事务代理配置实Advisor[] advisors = config.getAdvisors();//拦截器链List<Object> interceptorList = new ArrayList<>(advisors.length);for (Advisor advisor : advisors) {//PointcutAdvisor 我们常用的AOP日志拦截 对切入点进行拦截// BeanFactoryTransactionAttributeSourceAdvisor instanceof PointcutAdvisor)if (advisor instanceof PointcutAdvisor) { //TransactionInterceptor就是再这里获取到的 并加入到拦截器链中//方法上或者类上必须要有 注解 @Transactional//事务通过xml配置都会将 TransactionInterceptor加入到AOP拦截器链中...}else if (advisor instanceof IntroductionAdvisor) {//IntroductionAdvisor 对目标类增加新的属性和行为 本人没用过//是不是pinpointhe和skywalking这类链路追踪框架会用 后面再专门研究}else {//AOP的通用拦截器Interceptor[] interceptors = registry.getInterceptors(advisor);interceptorList.addAll(Arrays.asList(interceptors));}}return interceptorList;
}

核心逻辑
- 通过各个具体的Advisors去获取org.aopalliance.intercept.Interceptor拦截器
- 方法上或者类上如有有 @Transactional注解
  或者过xml配置都会将 TransactionInterceptor加入到AOP拦截器链中

分析Spring AOP责任链

通过创建CglibMethodInvocation对象实现责任链调用
org.springframework.aop.framework.CglibAopProxy.DynamicAdvisedInterceptor#intercept

688行创建责任链并执行各种源码框架对责任链的实现大同小异

//创建责任链调用对象 CglibMethodInvocation 并开始调用proceed
retVal = new CglibMethodInvocation(proxy, target, method, args, targetClass, chain, methodProxy).proceed();//CglibMethodInvocation 构造方法
public CglibMethodInvocation(Object proxy,@Nullable Object target, Method method,Object[] arguments, @Nullable Class<?> targetClass,List<Object> interceptorsAndDynamicMethodMatchers, // 拦截器链MethodProxy methodProxy) {super(proxy, target, method, arguments, targetClass, interceptorsAndDynamicMethodMatchers);// Only use method proxy for public methods not derived from java.lang.Objectthis.methodProxy = (Modifier.isPublic(method.getModifiers()) &&method.getDeclaringClass() != Object.class && !AopUtils.isEqualsMethod(method) &&!AopUtils.isHashCodeMethod(method) && !AopUtils.isToStringMethod(method) ?methodProxy : null);}

执行责任链调用 org.springframework.aop.framework.ReflectiveMethodInvocation#proceed

public Object proceed() throws Throwable {// 责任链 如果当前链的索引等于责任链的长度则当前链是最后一个链执行完成后直接返回//    We start with an index of -1 and increment early.if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {//执行目标方法return invokeJoinpoint();}// 获取链上的拦截器 默认从0开始递增Object interceptorOrInterceptionAdvice =this.interceptorsAndDynamicMethodMatchers.get(++this.currentInterceptorIndex);//判断Advice是否是 InterceptorAndDynamicMethodMatcher 动态方法匹配拦截器 //下面是InterceptorAndDynamicMethodMatcher的结构 这里具体的使用场景欢迎交流/* class InterceptorAndDynamicMethodMatcher {final MethodInterceptor interceptor;final MethodMatcher methodMatcher;public InterceptorAndDynamicMethodMatcher(MethodInterceptor interceptor, MethodMatcher methodMatcher) {this.interceptor = interceptor;this.methodMatcher = methodMatcher;}}*///TransactonIntercepters只是 MethodInterceptorif (interceptorOrInterceptionAdvice instanceof InterceptorAndDynamicMethodMatcher) {// Evaluate dynamic method matcher here: static part will already have// been evaluated and found to match.InterceptorAndDynamicMethodMatcher dm =(InterceptorAndDynamicMethodMatcher) interceptorOrInterceptionAdvice;Class<?> targetClass = (this.targetClass != null ? this.targetClass : this.method.getDeclaringClass());if (dm.methodMatcher.matches(this.method, targetClass, this.arguments)) {return dm.interceptor.invoke(this);}else {// Dynamic matching failed.// Skip this interceptor and invoke the next in the chain.return proceed();}}// 当前链就是是一个方法拦截器 直接执行else {// It's an interceptor, so we just invoke it: The pointcut will have// been evaluated statically before this object was constructed.// 注意这里入参是 this 即当前业务方法 如下图 return ((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);}
}

分析TransactionInterceptor源码

拦截器链TransactionInterceptor执行源码分析

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, invocation::proceed);}

重点分析org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction

/*** General delegate for around-advice-based subclasses, delegating to several other template* methods on this class. Able to handle {@link CallbackPreferringPlatformTransactionManager}* as well as regular {@link PlatformTransactionManager} implementations.* @param method the Method being invoked* @param targetClass the target class that we're invoking the method on* @param invocation the callback to use for proceeding with the target invocation* @return the return value of the method, if any* @throws Throwable propagated from the target invocation*/
@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);//确定事务管理器 org.springframework.jdbc.datasource.DataSourceTransactionManager final PlatformTransactionManager tm = determineTransactionManager(txAttr);//获取需要事务的方法签名final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);// 判断事务管理器是否是 CallbackPreferringPlatformTransactionManager//DataSourceTransactionManager显然不是CallbackPreferringPlatformTransactionManager//CallbackPreferringPlatformTransactionManager像是一种基于回调执行事务管理器if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {// Standard transaction demarcation with getTransaction and commit/rollback calls.//创建TransactionInfo对象 入参是DataSourceTransactionManager TransactionAttribute joinpointIdentificationTransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);Object retVal = null;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 exceptioncompleteTransactionAfterThrowing(txInfo, ex);throw ex;}finally {cleanupTransactionInfo(txInfo);}commitTransactionAfterReturning(txInfo);return retVal;}else {final ThrowableHolder throwableHolder = new ThrowableHolder();// It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.try {Object result = ((CallbackPreferringPlatformTransactionManager) tm).execute(txAttr, status -> {TransactionInfo txInfo = prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);try {return invocation.proceedWithInvocation();}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);}});// Check result state: It might indicate a Throwable to rethrow.if (throwableHolder.throwable != null) {throw throwableHolder.throwable;}return result;}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;}}
}

TransactionAttribute

282-283行获取事务属性 org.springframework.transaction.interceptor.TransactionAttribute

TransactionAttributeSource tas = getTransactionAttributeSource();
final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);

事务传播机制事务隔离级别超时时间等信息

org.springframework.transaction.interceptor.AbstractFallbackTransactionAttributeSource#getTransactionAttribute

/*** Determine the transaction attribute for this method invocation.* <p>Defaults to the class's transaction attribute if no method attribute is found.* @param method the method for the current invocation (never {@code null})* @param targetClass the target class for this invocation (may be {@code null})* @return a TransactionAttribute for this method, or {@code null} if the method* is not transactional*/
public TransactionAttribute getTransactionAttribute(Method method, @Nullable Class<?> targetClass) {// 先从缓存中获取TransactionAttribute// 如果缓存没有则调用方法 org.springframework.transaction.interceptor.AbstractFallbackTransactionAttributeSource#computeTransactionAttribute // 如果有则直接返回TransactionAttribute
}/*** Same signature as {@link #getTransactionAttribute}, but doesn't cache the result.* {@link #getTransactionAttribute} is effectively a caching decorator for this method.* <p>As of 4.1.8, this method can be overridden.* @since 4.1.8* @see #getTransactionAttribute*/
protected TransactionAttribute computeTransactionAttribute(Method method, @Nullable Class<?> targetClass) {// 事务方法必须是public修饰// 查找事务属性// 核心方法  org.springframework.transaction.interceptor.AbstractFallbackTransactionAttributeSource#findTransactionAttribute(java.lang.reflect.Method)//查找逻辑 大概是解析方法上或者类上的 @Transactional注解的属性
}

DataSourceTransactionManager

284行确定事务管理器
org.springframework.transaction.interceptor.TransactionAspectSupport#determineTransactionManager
- 返回org.springframework.jdbc.datasource.DataSourceTransactionManager
  - 感觉常用的都是这个事务管理器
  - 还有一些JTA的事务管理器比如 WebLogicJtaTransactionManager

TransactionInfo

TransactionAttributeSource TransactionAttribute DataSourceTransactionManager
都准备好后开始创建TransactionInfo对象多个TransactionInfo构成事务链

TransactionStatus

TransactionStatus对象是TransactionInfo关键核心属性 TransactionStatus是接口
我们默认使用的是org.springframework.transaction.support.DefaultTransactionStatus对象

//org.springframework.transaction.TransactionStatus
public interface TransactionStatus extends SavepointManager, Flushable {/*** Return whether the present transaction is new; otherwise participating* in an existing transaction, or potentially not running in an actual* transaction in the first place.*///该属性是核心属性 事务在提交时会判断该属性是否为true 如果为true则提交事务boolean isNewTransaction();/*** Return whether this transaction internally carries a savepoint,* that is, has been created as nested transaction based on a savepoint.* <p>This method is mainly here for diagnostic purposes, alongside* {@link #isNewTransaction()}. For programmatic handling of custom* savepoints, use the operations provided by {@link SavepointManager}.* @see #isNewTransaction()* @see #createSavepoint()* @see #rollbackToSavepoint(Object)* @see #releaseSavepoint(Object)*/boolean hasSavepoint();/*** Set the transaction rollback-only. This instructs the transaction manager* that the only possible outcome of the transaction may be a rollback, as* alternative to throwing an exception which would in turn trigger a rollback.* <p>This is mainly intended for transactions managed by* {@link org.springframework.transaction.support.TransactionTemplate} or* {@link org.springframework.transaction.interceptor.TransactionInterceptor},* where the actual commit/rollback decision is made by the container.* @see org.springframework.transaction.support.TransactionCallback#doInTransaction* @see org.springframework.transaction.interceptor.TransactionAttribute#rollbackOn*/void setRollbackOnly();/*** Return whether the transaction has been marked as rollback-only* (either by the application or by the transaction infrastructure).*/boolean isRollbackOnly();/*** Flush the underlying session to the datastore, if applicable:* for example, all affected Hibernate/JPA sessions.* <p>This is effectively just a hint and may be a no-op if the underlying* transaction manager does not have a flush concept. A flush signal may* get applied to the primary resource or to transaction synchronizations,* depending on the underlying resource.*/@Overridevoid flush();
}//org.springframework.transaction.support.AbstractTransactionStatus
public abstract class AbstractTransactionStatus implements TransactionStatus {/*** Return whether this transaction is completed, that is,* whether it has already been committed or rolled back.* @see PlatformTransactionManager#commit* @see PlatformTransactionManager#rollback*/boolean isCompleted();private boolean rollbackOnly = false;private boolean completed = false;@Nullableprivate Object savepoint;
}//org.springframework.transaction.support.DefaultTransactionStatus
public class DefaultTransactionStatus extends AbstractTransactionStatus {@Nullableprivate final Object transaction;//很重要的属性 默认传播机制下只有事务链的最外层事务该属性才会trueprivate final boolean newTransaction;//是否在食物链上进行事务属性同步比如传播机制 隔离级别 默认传播机制下为trueprivate final boolean newSynchronization;private final boolean readOnly;private final boolean debug;@Nullableprivate final Object suspendedResources;
}

289行创建TransactionInfo对象
org.springframework.transaction.interceptor.TransactionAspectSupport#createTransactionIfNecessary

/*** Create a transaction if necessary based on the given TransactionAttribute.* <p>Allows callers to perform custom TransactionAttribute lookups through* the TransactionAttributeSource.* @param txAttr the TransactionAttribute (may be {@code null})* @param joinpointIdentification the fully qualified method name* (used for monitoring and logging purposes)* @return a TransactionInfo object, whether or not a transaction was created.* The {@code hasTransaction()} method on TransactionInfo can be used to* tell if there was a transaction created.* @see #getTransactionAttributeSource()*/
@SuppressWarnings("serial")
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {//准备事务的名称 默认是事务方法的签名// If no name specified, apply method identification as transaction name.if (txAttr != null && txAttr.getName() == null) {txAttr = new DelegatingTransactionAttribute(txAttr) {@Overridepublic String getName() {return joinpointIdentification;}};}//获取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");}}}return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

289#474行获取TransactionStatus对象源码org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction

/*** This implementation handles propagation behavior. Delegates to* {@code doGetTransaction}, {@code isExistingTransaction}* and {@code doBegin}.* @see #doGetTransaction* @see #isExistingTransaction* @see #doBegin*/
@Override
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {//获取DataSourceTransactionObjectObject transaction = doGetTransaction();// Cache debug flag to avoid repeated checks.boolean debugEnabled = logger.isDebugEnabled();if (definition == null) {// Use defaults if no transaction definition given.definition = new DefaultTransactionDefinition();}//判断当前线程中的ConnectionHolder是否已被激活事务//主要是判断DataSourceTransactionObject的ConnectionHolder和isTransactionActiveif (isExistingTransaction(transaction)) {// Existing transaction found -> check propagation behavior to find out how to behave.//当前线程中的ConnectionHolder已被激活事务 则创建TransactionStatus //创建的TransactionStatus关键属性isNewTransactionw为falsereturn handleExistingTransaction(definition, transaction, debugEnabled);}//检查超时时间 默认是-1用不超时 // Check definition settings for new transaction.if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());} // 下面的各种if else 都是为了根据事务传播机制来构造TransactionStatus// No existing transaction found -> check propagation behavior to find out how to proceed.if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {throw new IllegalTransactionStateException("No existing transaction found for transaction marked with propagation 'mandatory'");}else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {SuspendedResourcesHolder suspendedResources = suspend(null);if (debugEnabled) {logger.debug("Creating new transaction with name [" + definition.getName() + "]: " + definition);}try {boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);//当前ConnectionHodler没有激活的事务//创建DefaultTransactionStatus对象 //注意第三个构造参数isNewTransactionw为true 这是食物链的first tDefaultTransactionStatus status = newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);//设置DataSourceTransactionObject属性//ConnectionHolder previousIsolationLevel transactionActive//初始化ConnectionHolder的java.sql.Connection 数据库连接 三次握手//设置java.sql.Connection autoCommit为falsedoBegin(transaction, definition);//判断DefaultTransactionStatus当前事务状态newSynchronization是否为true//则初始化 TransactionSynchronizationManager中各种TreadLocal容器prepareSynchronization(status, definition);//返回DefaultTransactionStatusreturn status;}catch (RuntimeException | Error ex) {resume(null, suspendedResources);throw ex;}}else {// Create "empty" transaction: no actual transaction, but potentially synchronization.if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {logger.warn("Custom isolation level specified but no actual transaction initiated; " +"isolation level will effectively be ignored: " + definition);}boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null);}
}

关键方法org.springframework.jdbc.datasource.DataSourceTransactionManager#doGetTransaction

/*** 获取DataSourceTransactionObject* org.springframework.jdbc.datasource.DataSourceTransactionManager.DataSourceTransactionObject* 从TransactionSynchronizationManager中获取ConnectionHolder对象* 如果是事务链最外层事务则当前TreadLocal中没有ConnectionHolder对象* org.springframework.jdbc.datasource.ConnectionHolder *  ConnectionHolder是数据库连接对象的封装*  ConnectionHolder存放在TheadLocal中*/
@Override
protected Object doGetTransaction() {DataSourceTransactionObject txObject = new DataSourceTransactionObject();txObject.setSavepointAllowed(isNestedTransactionAllowed());//从ThreadLocal中获取ConnectionHolder并设置到DataSourceTransactionObject//org.springframework.jdbc.datasource.ConnectionHolder /**TransactionSynchronizationManager是事务同步管理里面有各种TreadLocalorg.springframework.transaction.support.TransactionSynchronizationManager#resourcesprivate static final ThreadLocal<Map<Object, Object>> resources =new NamedThreadLocal<>("Transactional resources");*/ConnectionHolder conHolder =(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());txObject.setConnectionHolder(conHolder, false);return txObject;
}

关键对象TransactionSynchronizationManager 事务同步器org.springframework.transaction.support.TransactionSynchronizationManage

public abstract class TransactionSynchronizationManager {private static final Log logger = LogFactory.getLog(TransactionSynchronizationManager.class);private static final ThreadLocal<Map<Object, Object>> resources =new NamedThreadLocal<>("Transactional resources");private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =new NamedThreadLocal<>("Transaction synchronizations");private static final ThreadLocal<String> currentTransactionName =new NamedThreadLocal<>("Current transaction name");private static final ThreadLocal<Boolean> currentTransactionReadOnly =new NamedThreadLocal<>("Current transaction read-only status");private static final ThreadLocal<Integer> currentTransactionIsolationLevel =new NamedThreadLocal<>("Current transaction isolation level");private static final ThreadLocal<Boolean> actualTransactionActive =new NamedThreadLocal<>("Actual transaction active");

关键方法判断当前事务是否已被激活
org.springframework.jdbc.datasource.DataSourceTransactionManager#isExistingTransaction

//txObject.hasConnectionHolder()不为空 ConnectionHolder.isTransactionActivetrue则事务被激活
@Override
protected boolean isExistingTransaction(Object transaction) {DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;//ConnectionHolder是否为空 isTransactionActive是否为truereturn (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
}

关键方法当前ConnectionHolder已被激活事务org.springframework.transaction.support.AbstractPlatformTransactionManager#handleExistingTransaction

/*** Create a TransactionStatus for an existing transaction.*/
private TransactionStatus handleExistingTransaction(TransactionDefinition definition,  //事务定义 传播机制 隔离级别Object transaction, //DataSourceTransactionObject 持有ConnectionHolder对象boolean debugEnabled) throws TransactionException {//前面的很长的逻辑是 其他事务传播机制的 以后再分析 我们默认的传播机制使用不到...//是否同步事务属性 默认的传播机制是同步的boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);//构造TransactionStatus对象 第三车构造参数isNewTransaction为false//因为当前ConnectionHolder已被激活事务return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
}

关键方法设置DataSourceTransactionObject关键属性ConnectionHodler
org.springframework.transaction.support.AbstractPlatformTransactionManager#doBegin

//调用栈
org.springframework.transaction.interceptor.TransactionAspectSupport#createTransactionIfNecessary
#474 org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction
##378 org.springframework.transaction.support.AbstractPlatformTransactionManager#doBegin

/**
* This implementation sets the isolation level but ignores the timeout.*/
@Override
protected void doBegin(Object transaction, TransactionDefinition definition) {DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;Connection con = null;try {if (!txObject.hasConnectionHolder() ||txObject.getConnectionHolder().isSynchronizedWithTransaction()) {//从连接池中获取连接 可能会发生tcp三次握手 Connection newCon = obtainDataSource().getConnection();if (logger.isDebugEnabled()) {logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");}//将连接设置到DataSourceTransactionObjecttxObject.setConnectionHolder(new ConnectionHolder(newCon), true);}txObject.getConnectionHolder().setSynchronizedWithTransaction(true);con = txObject.getConnectionHolder().getConnection();//获取事务隔离级别Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);//将获取到的事务隔离几杯设置到上一个隔离级别属性中txObject.setPreviousIsolationLevel(previousIsolationLevel);// 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//数据库连接 Connection默认是自动提交事务的 需要在当前生命周期中设置为false// configured the connection pool to set it already).if (con.getAutoCommit()) {txObject.setMustRestoreAutoCommit(true);if (logger.isDebugEnabled()) {logger.debug("Switching JDBC Connection [" + con + "] to manual commit");}//设置为手动提交事务con.setAutoCommit(false);}//设置事务为read only 会请求数据产生网络IO//stmt.executeUpdate("SET TRANSACTION READ ONLY")prepareTransactionalConnection(con, definition);//设置当前ConnectionHolde已激活事务txObject.getConnectionHolder().setTransactionActive(true);//设置超时时间int timeout = determineTimeout(definition);if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {txObject.getConnectionHolder().setTimeoutInSeconds(timeout);}// 将DataSource和ConnectionHolder绑定到当前线程中// Bind the connection holder to the thread.if (txObject.isNewConnectionHolder()) {/*** key DataSource value : ConnectionHolder* private static final ThreadLocal<Map<Object, Object>> resources =new NamedThreadLocal<>("Transactional resources");*/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);}
}//设置事务为readOnly
protected void prepareTransactionalConnection(Connection con, TransactionDefinition definition)throws SQLException {if (isEnforceReadOnly() && definition.isReadOnly()) {Statement stmt = con.createStatement();try {stmt.executeUpdate("SET TRANSACTION READ ONLY");}finally {stmt.close();}}
}

关键方法org.springframework.transaction.support.AbstractPlatformTransactionManager#prepareSynchronization

/*** Initialize transaction synchronization as appropriate.*/
protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {//isNewSynchronizatio是否同步食物链属性 默认传播机制下为trueif (status.isNewSynchronization()) {//初始化  TransactionSynchronizationManager中各种TreadLocal容器//将事务信息存入线程上下文中TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ?definition.getIsolationLevel() : null);TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());TransactionSynchronizationManager.initSynchronization();}
}

TransactionInfo形成事务链

484行创建TransactionInfo对象
org.springframework.transaction.interceptor.TransactionAspectSupport#prepareTransactionInfo

//调用栈
org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction
#289 org.springframework.transaction.interceptor.TransactionAspectSupport#createTransactionIfNecessary
##483 org.springframework.transaction.interceptor.TransactionAspectSupport#prepareTransactionInfo

//org.springframework.transaction.interceptor.TransactionAspectSupport.TransactionInfo
protected final class TransactionInfo {//事务管理器      @Nullableprivate final PlatformTransactionManager transactionManager;//事务属性@Nullableprivate final TransactionAttribute transactionAttribute;//事务名称private final String joinpointIdentification;//关键属性 事务状态@Nullableprivate TransactionStatus transactionStatus;//上一个事务 形成事务链@Nullableprivate TransactionInfo oldTransactionInfo;
}

源码分析

protected TransactionInfo prepareTransactionInfo(@Nullable PlatformTransactionManager tm, // 事务管理器 DataSourceTransactionManager@Nullable TransactionAttribute txAttr, // 事务定义 隔离级别 传播机制 等信息String joinpointIdentification,//事务名称 默认是当前业务方法的签名@Nullable TransactionStatus status // 核心属性 事务状态) {// 直接new TransactionInfo 食物链上的每个TransactionInfo都是new出来的TransactionInfo txInfo = new TransactionInfo(tm, txAttr, joinpointIdentification);if (txAttr != null) {// We need a transaction for this method...if (logger.isTraceEnabled()) {logger.trace("Getting transaction for [" + txInfo.getJoinpointIdentification() + "]");}// The transaction manager will flag an error if an incompatible tx already exists.// 设置TransactionStatus TransactionStatus每次也都是new出来的对象 txInfo.newTransactionStatus(status);}else {// The TransactionInfo.hasTransaction() method will return false. We created it only// to preserve the integrity of the ThreadLocal stack maintained in this class.if (logger.isTraceEnabled()) {logger.trace("Don't need to create transaction for [" + joinpointIdentification +"]: This method isn't transactional.");}}// We always bind the TransactionInfo to the thread, even if we didn't create// a new transaction here. This guarantees that the TransactionInfo stack// will be managed correctly even if no transaction was created by this aspect.// 将TransactionInfo绑定到当前线程中 形成事务链txInfo.bindToThread();return txInfo;
}

关键方法org.springframework.transaction.interceptor.TransactionAspectSupport.TransactionInfo#bindToThread

private void bindToThread() {// Expose current TransactionStatus, preserving any existing TransactionStatus// for restoration after this transaction is complete.// 将旧的事务取出赋值到oldTransactionInfo 形成事务链this.oldTransactionInfo = transactionInfoHolder.get();transactionInfoHolder.set(this);
}//事务链绑定在当前线程上下文中
public abstract class TransactionAspectSupport implements BeanFactoryAware, InitializingBean {/*** Holder to support the {@code currentTransactionStatus()} method,* and to support communication between different cooperating advices* (e.g. before and after advice) if the aspect involves more than a* single method (as will be the case for around advice).*/private static final ThreadLocal<TransactionInfo> transactionInfoHolder =new NamedThreadLocal<>("Current aspect-driven transaction");
}

执行事务方法

294行 springframework.transaction.interceptor.TransactionAspectSupport.InvocationCallback#proceedWithInvocation

org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction
#294行 org.springframework.transaction.interceptor.TransactionAspectSupport.InvocationCallback#proceedWithInvocation

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

责任链是如何执行的

invocation.proceedWithInvocation();执行链上的下一个拦截器

会再次执行到ReflectiveMethodInvocation#proceed方法

public Object proceed() throws Throwable {//此时拦截器链索引等于拦截器的长度即 链中没有还未执行的拦截器//    We start with an index of -1 and increment early.if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {return invokeJoinpoint();
}

执行事务提交

业务方法已执行完 SQL语句已发送给数据库并执行
事务提交部分分为两步

Restore事务链
提交事务

Restore事务链

302行将当前线程ThreadLocal中的事务链向上Restore
org.springframework.transaction.interceptor.TransactionAspectSupport#cleanupTransactionInfo

/*** Reset the TransactionInfo ThreadLocal.* <p>Call this in all cases: exception or normal return!* @param txInfo information about the current transaction (may be {@code null})*/
protected void cleanupTransactionInfo(@Nullable TransactionInfo txInfo) {if (txInfo != null) {txInfo.restoreThreadLocalStatus();}
}private void restoreThreadLocalStatus() {// Use stack to restore old transaction TransactionInfo.// Will be null if none was set.//当前业务方法已执行完 将事务链向上restoretransactionInfoHolder.set(this.oldTransactionInfo);
}

提交事务

304行执行事务提交方法org.springframework.transaction.interceptor.TransactionAspectSupport#commitTransactionAfterReturning

关键方法org.springframework.transaction.support.AbstractPlatformTransactionManager#commit

/*** This implementation of commit handles participating in existing* transactions and programmatic rollback requests.* Delegates to {@code isRollbackOnly}, {@code doCommit}* and {@code rollback}.* @see org.springframework.transaction.TransactionStatus#isRollbackOnly()* @see #doCommit* @see #rollback*/
@Override
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;// 判断是否需要回滚 从DefaultTransactionStatus对象中读取 则为LocalRollbackOnlyif (defStatus.isLocalRollbackOnly()) {if (defStatus.isDebug()) {logger.debug("Transactional code has requested rollback");}processRollback(defStatus, false);return;}//判断是否需要回滚 从DefaultTransactionStatus中的//DataSourceTransactionObject的ConnectionHolder.Connection中读取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);
}

因为事务链跟Treadlocal绑定我们在业务代码中可以自己控制是否需要回滚事务
通过获取DefaultTransactionStatus手动设置RollbackOnly来控制事务是否回滚
TransactionAspectSupport.currentTransactionStatus().setRollbackOnly()

关键方法org.springframework.transaction.support.AbstractPlatformTransactionManager#processCommit

/*** Process an actual commit.* Rollback-only flags have already been checked and applied.* @param status object representing the transaction* @throws TransactionException in case of commit failure*/
private void processCommit(DefaultTransactionStatus status) throws TransactionException {try {boolean beforeCompletionInvoked = false;try {boolean unexpectedRollback = false;//事务准备提交 prepareForCommit(status);//事务提交前触发 TransactionSynchronization//private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =//new NamedThreadLocal<>("Transaction synchronizations");triggerBeforeCommit(status);//事务提交前生命周期结束 触发TransactionSynchronization triggerBeforeCompletion(status);beforeCompletionInvoked = true;//事务的savePoint用于嵌套事务 默认传播机制用不到 不做分析if (status.hasSavepoint()) {if (status.isDebug()) {logger.debug("Releasing transaction savepoint");}unexpectedRollback = status.isGlobalRollbackOnly();status.releaseHeldSavepoint();}//核心核心核心 事务是否提交 就是判断 DefaultTransactionStatus的isNewTransaction//一路分析源码我们得知在默认事务传播机制下事务链上的最外层事务的isNewTransaction才为true//如果不是最外层事务不会真正提交 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 doCommittriggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);throw ex;}catch (TransactionException ex) {// can only be caused by doCommitif (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 {//事务提交后 触发TransactionSynchronizationtriggerAfterCommit(status);}finally {//只有当事务链中最外层事务提交后 才会触发TransactionSynchronization#invokeAfterCompletion方法triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);}}finally {cleanupAfterCompletion(status);}
}

只有当前事务的isNewTransaction为true才会执行真正的提交方法
如果是默认的事务传播机制只有事务链的最外层事务isNewTransaction才为true‘

关键方法真正执行事务提交org.springframework.jdbc.datasource.DataSourceTransactionManager#doCommit

@Override
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 {//提交事务 下面就是JDBC的源码了con.commit();}catch (SQLException ex) {throw new TransactionSystemException("Could not commit JDBC transaction", ex);}
}

事务提交完成后

793行设置当前事务状态以及清理动作org.springframework.transaction.support.AbstractPlatformTransactionManager#cleanupAfterCompletion

org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction
#304 org.springframework.transaction.interceptor.TransactionAspectSupport#commitTransactionAfterReturning
##533 org.springframework.transaction.PlatformTransactionManager#commit
###714 org.springframework.transaction.support.AbstractPlatformTransactionManager#processCommit
####793 org.springframework.transaction.support.AbstractPlatformTransactionManager#cleanupAfterCompletion

/*** Clean up after completion, clearing synchronization if necessary,* and invoking doCleanupAfterCompletion.* @param status object representing the transaction* @see #doCleanupAfterCompletion*/
private void cleanupAfterCompletion(DefaultTransactionStatus status) {//设置DefaultTransactionStatus#completed为true 标示事务已完成status.setCompleted();//清除当前线程中TreadLocal中的事务信息if (status.isNewSynchronization()) {TransactionSynchronizationManager.clear();}//如果是事务链的最外层事务则从TreadLocal中清除绑定资源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);//如果之前有挂起的事务则恢复挂起的事务到当前线程ThreadLocal中resume(transaction, (SuspendedResourcesHolder) status.getSuspendedResources());}
}//解绑资源
@Override
protected void doCleanupAfterCompletion(Object transaction) {DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;// 清除resources key:DataTransactonResourceObject vlaue:TrasactionDefinition// private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<>("Transactional resources"); // Remove the connection holder from the thread, if exposed.if (txObject.isNewConnectionHolder()) {TransactionSynchronizationManager.unbindResource(obtainDataSource());}// 将Connection连接的autoCommint重新设置为自动提交 恢复事务隔离级别// Reset connection.Connection con = txObject.getConnectionHolder().getConnection();try {if (txObject.isMustRestoreAutoCommit()) {con.setAutoCommit(true);}DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel());}catch (Throwable ex) {logger.debug("Could not reset JDBC Connection after transaction", ex);}if (txObject.isNewConnectionHolder()) {if (logger.isDebugEnabled()) {logger.debug("Releasing JDBC Connection [" + con + "] after transaction");}//释放Connection 连接都是租用的 需要释放连接使其可以继续被租用DataSourceUtils.releaseConnection(con, this.dataSource);}//清除ConnectionHolder的一些信息txObject.getConnectionHolder().clear();
}

总结

问题解答

Spring通过AOP管理事务

核心对象如下

org.springframework.transaction.PlatformTransactionManager
- org.springframework.transaction.support.AbstractPlatformTransactionManager
  - org.springframework.jdbc.datasource.DataSourceTransactionManager
  - org.aopalliance.intercept.MethodInterceptor
    - org.springframework.transaction.interceptor.TransactionAspectSupport
      - org.springframework.transaction.interceptor.TransactionInterceptor
  - org.springframework.transaction.interceptor.TransactionAspectSupport.TransactionInfo
  - org.springframework.transaction.TransactionDefinition
    - org.springframework.transaction.interceptor.TransactionAttribute
  - org.springframework.transaction.TransactionStatus
  - org.springframework.jdbc.datasource.DataSourceTransactionManager.DataSourceTransactionObject

多线程事务问题

阅读源码得知 Spring实现事务管理使用了大量的ThreadLocal存储事务信息. 在多线程状态下事务无法管理

禁止使用Java8 parallelStream来调用事务方法
什么情况下事务不会被spring管理不生效
- 当外层的事务方法的对象不是代理对象无法被AO P
  - 即使被Spring托管的对象也有可能不是代理对象
  - 本人遇到 @Bean管理的对象不是代理对象原因没有深究大佬可以留言解答
- 当事务方法或者类上没有@Transactional并且没有使用xml配置切面管理事务
- 非public方法不会被Spring事务管理
- **业务Service内部调用不会被事务管理 **
  - 下面的this调用事务也会被统一管理 updateById2执行完后并不会提供提交原因是updateById2和updateById使用的是同一个数据库connection对象详细的原理我后面会再出博客解释

this调用事务也被Spring管理的情况

因为 userService依然是代理对象

@Transactional(rollbackFor = Exception.class)public void testTransaction(UserModifyParam userModifyParam){userService.updateById(userModifyParam);this.innerInvoke(userModifyParam);}private void innerInvoke(UserModifyParam userModifyParam){//userService此时依然是代理对象userService.updateById2(userModifyParam);}

关键类图

org.springframework.transaction.interceptor.TransactionInterceptor
org.springframework.transaction.interceptor.TransactionAspectSupport.TransactionInfo
org.springframework.transaction.support.DefaultTransactionStatus

事务链