Spring聲明式事務(wù)源碼詳解
作者：張?jiān)苿?/span>
維護(hù)公司之前的后臺(tái)管理系統(tǒng)，在開(kāi)發(fā)自測(cè)時(shí)發(fā)現(xiàn)mock接口失敗，數(shù)據(jù)庫(kù)仍插入成功。經(jīng)排查發(fā)現(xiàn)代碼中沒(méi)有指定具體事務(wù)，在添加注解@Transactional后事務(wù)按預(yù)期生效回滾。為此，本文通過(guò)分析源碼來(lái)介紹下事務(wù)失效的根源。
維護(hù)公司之前的后臺(tái)管理系統(tǒng)，在開(kāi)發(fā)自測(cè)時(shí)發(fā)現(xiàn)mock接口失敗，數(shù)據(jù)庫(kù)仍插入成功。經(jīng)排查發(fā)現(xiàn)代碼中沒(méi)有指定具體事務(wù)，在添加注解@Transactional后事務(wù)按預(yù)期生效回滾。為此，本文通過(guò)分析源碼來(lái)介紹下事務(wù)失效的根源。
1.事務(wù)的幾個(gè)小小例子
1.1 未添加事務(wù)，異常未回滾
代碼未執(zhí)行前數(shù)據(jù)庫(kù)信息：
圖片
@Component
public class TransactionalTest {
@Resource
BasicPriceUploadRecordMapper basicPriceUploadRecordMapper ;
public void onAddTransactionToException() {
BasicPriceUploadRecord base = new BasicPriceUploadRecord();
base.setErrorMsg("未添加注解事務(wù)拋異常");
base.setId(1824040965380245002L);
basicPriceUploadRecordMapper.updateByPrimaryKeySelective(base);
throw new RuntimeException();
}
}代碼執(zhí)行后數(shù)據(jù)庫(kù)信息：
圖片
雖然拋出異常，但數(shù)據(jù)庫(kù)仍修改成功了。
1.2 添加注解事務(wù)，異常回滾
代碼未執(zhí)行前數(shù)據(jù)庫(kù)信息：
圖片
@Component
public class TransactionalTest {
@Resource
BasicPriceUploadRecordMapper basicPriceUploadRecordMapper ;
/**
* 添加聲明式注解，出現(xiàn)異常情況
*/
@Transactional(transactionManager = "valuationTransactionManager", rollbackFor = Exception.class)
public void addTransactionToException() {
BasicPriceUploadRecord base = new BasicPriceUploadRecord();
base.setErrorMsg("添加注解事務(wù)拋異常");
base.setId(1824040965380245002L);
basicPriceUploadRecordMapper.updateByPrimaryKeySelective(base);
throw new RuntimeException();
}
}代碼執(zhí)行后數(shù)據(jù)庫(kù)信息：
圖片
沒(méi)有修改成功，證明添加注解事務(wù)生效，進(jìn)行了事務(wù)回滾操作。
通過(guò)上述簡(jiǎn)單例子，我們了解下聲明式事務(wù)@Transactional的一些相關(guān)信息。
2. Transactional
2.1 注解定義
是Spring框架中用于聲明式事務(wù)管理的一個(gè)注解，主要通過(guò) AOP（面向切面編程）來(lái)實(shí)現(xiàn)事務(wù)管理。可以被應(yīng)用于接口定義、接口方法、類定義或類的public方法上。
2.2 常用屬性
value|transactionManager：指定事務(wù)管理器名稱
propagation ：事務(wù)的傳播行為
isolation： 事務(wù)的隔離級(jí)別
timeout： 事務(wù)的超時(shí)時(shí)間
readOnly： 事務(wù)是否為只讀
rollbackFor：一個(gè)異常類數(shù)組，遇到這些異常時(shí)事務(wù)回滾
noRollbackFor：一個(gè)異常類數(shù)組，遇到這些異常時(shí)事務(wù)不回滾
事務(wù)的傳播行為：
required：當(dāng)前存在事務(wù)，則加入該事務(wù)；如果當(dāng)前沒(méi)有事務(wù)，則創(chuàng)建一個(gè)新的事務(wù)
supports：當(dāng)前存在事務(wù)，則加入該事務(wù)；如果當(dāng)前沒(méi)有事務(wù)，則以非事務(wù)方式執(zhí)行
mandatory：當(dāng)前存在事務(wù)，則加入該事務(wù)；如果當(dāng)前沒(méi)有事務(wù)，則拋出異常
requires_new：創(chuàng)建一個(gè)新的事務(wù)，并且掛起當(dāng)前事務(wù)
not_supported：以非事務(wù)方式執(zhí)行，并且掛起當(dāng)前事務(wù)
never：以非事務(wù)方式執(zhí)行，如果存在事務(wù)，則拋出異常
nested：當(dāng)前存在事務(wù)，則在嵌套事務(wù)內(nèi)執(zhí)行；如果當(dāng)前沒(méi)有事務(wù)，則其效果與required相同
事務(wù)的隔離級(jí)別：在spring中，用于控制事務(wù)之間的并發(fā)訪問(wèn)，以減少或避免事務(wù)之間的數(shù)據(jù)不一致問(wèn)題。開(kāi)發(fā)中基本都是 default 級(jí)別
default：使用底層數(shù)據(jù)庫(kù)默認(rèn)的隔離級(jí)別
read_uncommitted：最低級(jí)別的隔離，事務(wù)可以看到其他未提交事務(wù)的數(shù)據(jù)
read_committed：事務(wù)只能看到其他已提交事務(wù)的數(shù)據(jù)，可以避免臟讀
repeatable_read：事務(wù)在整個(gè)過(guò)程中多次讀取同一數(shù)據(jù)時(shí)，結(jié)果是一致的，可以避免不可重復(fù)讀
serializable：最高級(jí)別的隔離，事務(wù)之間完全隔離，可以避免幻讀3. 源碼分析
以下源碼均基于Spring4.3.12版本。主要從 創(chuàng)建事務(wù)、開(kāi)啟事務(wù)、提交事務(wù)、事務(wù)回滾 的維度來(lái)詳細(xì)分析聲明式事務(wù)。
3.1 事務(wù)簡(jiǎn)易流程圖
圖片
3.2 代理類生成
在Spring框架中，當(dāng)你配置了事務(wù)管理器并聲明了@Transactional注解時(shí)，Spring會(huì)在實(shí)例化bean時(shí)生成事務(wù)增強(qiáng)的代理類。創(chuàng)建代理類參考源碼路徑如下：
AbstractAutowireCapableBeanFactory.createBean=>
doCreateBean()=>
initializeBean()=>
applyBeanPostProcessorsAfterInitialization()=>
postProcessAfterInitialization()（BeanPostProcessor內(nèi)接口）=>
AbstractAutoProxyCreator.postProcessAfterInitialization()=>
wrapIfNecessary()=>
createProxy() 中 proxyFactory.setProxyTargetClass(true); //是否對(duì)類進(jìn)行代理的設(shè)置，true為cglib代理3.3 代理類中方法執(zhí)行入口
從TransactionInterceptor.invoke()方法開(kāi)始分析 (獲取代理類，調(diào)用父類TransactionAspectSupport.invokeWithinTransaction()方法，該方法會(huì)將代理類的方法納入事務(wù)中)。
public class TransactionInterceptor extends TransactionAspectSupport implements MethodInterceptor, Serializable {
public Object invoke(final MethodInvocation invocation) throws Throwable {
// 返回代理類的目標(biāo)類
Class<?> targetClass = invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null;
//事務(wù)中執(zhí)行被代理的方法
return this.invokeWithinTransaction(invocation.getMethod(), targetClass, new InvocationCallback() {
public Object proceedWithInvocation() throws Throwable {
return invocation.proceed();
}
});
}
}3.4 主要核心邏輯
TransactionAspectSupport.invokeWithinTransaction()方法負(fù)責(zé)獲取事務(wù)屬性和事務(wù)管理器，然后針對(duì)聲明式事務(wù)和編程式事務(wù)區(qū)分處理流程（此處源碼忽略編程式事務(wù)）。
protected Object invokeWithinTransaction(Method method, Class<?> targetClass, final TransactionAspectSupport.InvocationCallback invocation) throws Throwable {
// 獲取事務(wù)屬性 TransactionDefinition對(duì)象（回顧規(guī)則，隔離級(jí)別，只讀等）
final TransactionAttribute txAttr = this.getTransactionAttributeSource().getTransactionAttribute(method, targetClass);
// 根據(jù)事務(wù)屬性和方法，獲取對(duì)應(yīng)的事務(wù)管理器，（后續(xù)用于做事務(wù)的提交，回滾等操作），數(shù)據(jù)庫(kù)的一些信息，
final PlatformTransactionManager tm = this.determineTransactionManager(txAttr);
// 獲取事務(wù)方法全路徑，
final String joinpointIdentification = this.methodIdentification(method, targetClass, txAttr);
//響應(yīng)式編程事務(wù)，大多數(shù)情況下都會(huì)執(zhí)行到 else中的語(yǔ)句；
// CallbackPreferringPlatformTransactionManager 可以通過(guò)回掉函數(shù)來(lái)處理事務(wù)的提交和回滾操作， 此處不考慮，此處源碼可以忽略
if (txAttr != null && tm instanceof CallbackPreferringPlatformTransactionManager) {
// 此處省略，此處為編程式事務(wù) 處理邏輯
} else {
//創(chuàng)建事務(wù)，事務(wù)屬性等信息會(huì)被保存進(jìn) TransactionInfo,便于后續(xù)流程中的提交和回滾操作，詳情見(jiàn)下文
TransactionAspectSupport.TransactionInfo txInfo = this.createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
Object retVal = null;
try {
// 執(zhí)行目標(biāo)的方法 （執(zhí)行具體的業(yè)務(wù)邏輯）
retVal = invocation.proceedWithInvocation();
} catch (Throwable var15) {
//異常處理
this.completeTransactionAfterThrowing(txInfo, var15);
throw var15;
} finally {
//清除當(dāng)前節(jié)點(diǎn)的事務(wù)消息，將舊事務(wù)節(jié)點(diǎn)消息通過(guò)ThreadLoacl更新到當(dāng)前線程（事務(wù)的掛起操作就是在這執(zhí)行）
this.cleanupTransactionInfo(txInfo);
}
//提交事務(wù)
this.commitTransactionAfterReturning(txInfo);
return retVal;
}
}3.4.1 開(kāi)啟事務(wù)
TransactionAspectSupport.createTransactionIfNecessary() 方法作用是檢查當(dāng)前是否存在事務(wù)，如果存在，則根據(jù)一定的規(guī)則創(chuàng)建一個(gè)新的事務(wù)。
protected TransactionAspectSupport.TransactionInfo createTransactionIfNecessary(PlatformTransactionManager tm, TransactionAttribute txAttr, final String joinpointIdentification) {
//如果事務(wù)名稱不為空，則使用方法唯一標(biāo)識(shí)。并使用 DelegatingTransactionAttribute 封裝 txAttr
if (txAttr != null && ((TransactionAttributerollbackOn)txAttr).getName() == null) {
txAttr = new DelegatingTransactionAttribute((TransactionAttribute)txAttr) {
public String getName() {
return joinpointIdentification;
}
};
}
TransactionStatus status = null;
if (txAttr != null) {
if (tm != null) {
// 獲取事務(wù)狀態(tài)。內(nèi)部判斷是否開(kāi)啟事務(wù)綁定線程與數(shù)據(jù)庫(kù)連接。詳情見(jiàn)下文
status = tm.getTransaction((TransactionDefinition)txAttr);
} else if (this.logger.isDebugEnabled()) {
this.logger.debug("Skipping transactional joinpoint [" + joinpointIdentification + "] because no transaction manager has been configured");
}
}
//構(gòu)建事務(wù)消息，根據(jù)指定的屬性與狀態(tài)status 構(gòu)建一個(gè) TransactionInfo。將已經(jīng)建立連接的事務(wù)所有信息，都記錄在ThreadLocal下的TransactionInfo 實(shí)例中，包括目標(biāo)方法的所有狀態(tài)信息，如果事務(wù)執(zhí)行失敗，spring會(huì)根據(jù)TransactionInfo中的信息來(lái)進(jìn)行回滾等后續(xù)操作
return this.prepareTransactionInfo(tm, (TransactionAttribute)txAttr, joinpointIdentification, status);
}1)AbstractPlatformTransactionManager.getTransaction() 獲取當(dāng)前事務(wù)對(duì)象。通過(guò)這個(gè)方法，可以獲取到關(guān)于事務(wù)的詳細(xì)信息，如事務(wù)的狀態(tài)、相關(guān)屬性等。
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
throws TransactionException {
//definition 中存儲(chǔ)的事務(wù)的注解信息，超時(shí)時(shí)間和隔離級(jí)別等
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
// 獲取當(dāng)前事務(wù)
Object transaction = doGetTransaction();
boolean debugEnabled = logger.isDebugEnabled();
// 判斷當(dāng)前線程是否存在事務(wù)
if (isExistingTransaction(transaction)) {
// 處理已經(jīng)存在的事務(wù)
return handleExistingTransaction(def, transaction, debugEnabled);
}
// 事務(wù)超時(shí)設(shè)置驗(yàn)證，超時(shí)時(shí)間小于-1 拋異常
if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
}
// 如果當(dāng)前線程不存在事務(wù)且 事務(wù)傳播行為是 MANDATORY（用當(dāng)前事務(wù)，如果當(dāng)前沒(méi)有事務(wù)，則拋出異常） 拋異常
if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
throw new IllegalTransactionStateException(
"No existing transaction found for transaction marked with propagation 'mandatory'");
}
//以下三種事務(wù)傳播行為 需要開(kāi)啟新事務(wù)
else if (def.getPropagationBehavior() == TransactionDefinition.propagation_required ||
def.getPropagationBehavior() == TransactionDefinition.propagation_requires_new ||
def.getPropagationBehavior() == TransactionDefinition.propagation_nested) {
//掛起原事務(wù)，因?yàn)檫@里不存在原事務(wù) 故設(shè)置為null。
//當(dāng)一個(gè)事務(wù)方法內(nèi)部調(diào)用了另一個(gè)事務(wù)方法時(shí)，如果第二個(gè)事務(wù)方法需要獨(dú)立于第一個(gè)事務(wù)方法，那么可以使用 suspend 方法來(lái)掛起當(dāng)前事務(wù)，然后再開(kāi)始一個(gè)新的事務(wù)
AbstractPlatformTransactionManager.SuspendedResourcesHolder suspendedResources = this.suspend((Object)null);
try {
boolean newSynchronization = this.getTransactionSynchronization() != 2;
DefaultTransactionStatus status = this.newTransactionStatus((TransactionDefinition)definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
//開(kāi)啟事務(wù)
this.doBegin(transaction, (TransactionDefinition)definition);
//同步事務(wù)狀態(tài)及書(shū)屋屬性
this.prepareSynchronization(status, (TransactionDefinition)definition);
return status;
} catch (RuntimeException var7) {
this.resume((Object)null, suspendedResources);
throw var7;
}
}
else {
boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);//0
//創(chuàng)建默認(rèn)狀態(tài) 詳情見(jiàn) 下文
return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
}
}1.1AbstractPlatformTransactionManager.doGetTransaction() 方法可能用于執(zhí)行獲取事務(wù)的具體操作。它可能會(huì)根據(jù)一些條件或規(guī)則，去查找和獲取當(dāng)前的事務(wù)對(duì)象，并進(jìn)行相應(yīng)的處理。
protected Object doGetTransaction() {
DataSourceTransactionManager.DataSourceTransactionObject txObject = new DataSourceTransactionManager.DataSourceTransactionObject();
//是否允許在一個(gè)事務(wù)內(nèi)部開(kāi)啟另一個(gè)事務(wù)。
txObject.setSavepointAllowed(this.isNestedTransactionAllowed());
// this.dataSource數(shù)據(jù)源 配置
//判斷當(dāng)前線程如果已經(jīng)記錄數(shù)據(jù)庫(kù)連接則使用原連接
ConnectionHolder conHolder = (ConnectionHolder)TransactionSynchronizationManager.getResource(this.dataSource);
//false 表示不是新創(chuàng)建連接
txObject.setConnectionHolder(conHolder, false);
return txObject;
}1.1.1this.dataSource() 是我們配置DataSourceTransactionManager時(shí)傳入的。
<bean id="valuationTransactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="valuationDataSource"/>
</bean>1.1.2TransactionSynchronizationManager.getResource() 方法的作用主要是獲取與當(dāng)前事務(wù)相關(guān)聯(lián)的資源。TransactionSynchronizationManager 持有一個(gè)ThreadLocal的實(shí)例，存在一個(gè)key為dataSource ，value為ConnectionHolder 的Map信息。
//ThreadLocal 存放 ConnectionHolder 信息，ConnectionHolder 可以理解為Connection（數(shù)據(jù)庫(kù)連接）的包裝類，其中最主要屬性為 Connection
private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal("Transactional resources");
// 獲取ConnectionHolder
public static Object getResource(Object key) {
Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
//獲取連接信息
Object value = doGetResource(actualKey);
return value;
}
//具體執(zhí)行獲取連接信息操作
private static Object doGetResource(Object actualKey) {
//從 ThreadLoacl中獲取
Map<Object, Object> map = (Map)resources.get();
if (map == null) {
return null;
} else {
Object value = map.get(actualKey);
if (value instanceof ResourceHolder && ((ResourceHolder)value).isVoid()) {
map.remove(actualKey);
if (map.isEmpty()) {
resources.remove();
}
value = null;
}
return value;
}
}1.2AbstractPlatformTransactionManager.isExistingTransaction() 方法用于判斷是否存在正在進(jìn)行的事務(wù)。它可以幫助我們確定當(dāng)前的執(zhí)行環(huán)境是否處于事務(wù)中，以便進(jìn)行相應(yīng)的處理。
protected boolean isExistingTransaction(Object transaction) {
DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
return txObject.getConnectionHolder() != null && txObject.getConnectionHolder().isTransactionActive();
}1.3AbstractPlatformTransactionManager.suspend() 掛起事務(wù),對(duì)有無(wú)同步的事務(wù)采取不同方案，doSuspend()執(zhí)行掛起具體操作。
protected final AbstractPlatformTransactionManager.SuspendedResourcesHolder suspend(Object transaction) throws TransactionException {
//如果有同步的事務(wù)，則優(yōu)先掛起同步事務(wù)
if (TransactionSynchronizationManager.isSynchronizationActive()) {
List suspendedSynchronizations = this.doSuspendSynchronization();
try {
Object suspendedResources = null;
if (transaction != null) {
//執(zhí)行掛起操作
suspendedResources = this.doSuspend(transaction);
}
//重置事務(wù)名稱
String name = TransactionSynchronizationManager.getCurrentTransactionName();
TransactionSynchronizationManager.setCurrentTransactionName((String)null);
//重置只讀狀態(tài)
boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();
TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);
//重置隔離級(jí)別
Integer isolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
TransactionSynchronizationManager.setCurrentTransactionIsolationLevel((Integer)null);
//重置事務(wù)激活狀態(tài)
boolean wasActive = TransactionSynchronizationManager.isActualTransactionActive();
TransactionSynchronizationManager.setActualTransactionActive(false);
//返回掛起的事務(wù)
return new AbstractPlatformTransactionManager.SuspendedResourcesHolder(suspendedResources, suspendedSynchronizations, name, readOnly, isolationLevel, wasActive);
} catch (RuntimeException var8) {
this.doResumeSynchronization(suspendedSynchronizations);
throw var8;
}
} else if (transaction != null) {
Object suspendedResources = this.doSuspend(transaction);
return new AbstractPlatformTransactionManager.SuspendedResourcesHolder(suspendedResources);
} else {
return null;
}
}1.3.1 AbstractPlatformTransactionManager.doSuspend()執(zhí)行掛起操作只是將當(dāng)前ConnectionHolder設(shè)置為null，返回原有事務(wù)消息，方便后續(xù)恢復(fù)原有事務(wù)消息，并將當(dāng)前正在進(jìn)行的事務(wù)信息進(jìn)行重置。
protected Object doSuspend(Object transaction) {
DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
txObject.setConnectionHolder((ConnectionHolder)null);
//接觸綁定
return TransactionSynchronizationManager.unbindResource(this.dataSource);
}1.3.1.1TransactionSynchronizationManager.unbindResource()解除綁定操作，將現(xiàn)有的事務(wù)消息remove并返回上一級(jí)
public static Object unbindResource(Object key) throws IllegalStateException {
Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
//解綁操作，移除資源
Object value = doUnbindResource(actualKey);
if (value == null) {
throw new IllegalStateException("No value for key [" + actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
} else {
return value;
}
}1.4 AbstractPlatformTransactionManager.doBegin()數(shù)據(jù)庫(kù)連接獲取，當(dāng)新事務(wù)時(shí)，則獲取新的數(shù)據(jù)庫(kù)連接，并為其設(shè)置隔離級(jí)別，是否只讀等屬性。
protected void doBegin(Object transaction, TransactionDefinition definition) {
DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
Connection con = null;
try {
//新事務(wù)開(kāi)啟時(shí)將 ConnectionHolder 設(shè)置為null
if (txObject.getConnectionHolder() == null || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
//獲取新的數(shù)據(jù)庫(kù)連接
Connection newCon = this.dataSource.getConnection();
txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
}
txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
con = txObject.getConnectionHolder().getConnection();
//設(shè)置事務(wù)隔離級(jí)別 和readOnly屬性
Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
txObject.setPreviousIsolationLevel(previousIsolationLevel);
if (con.getAutoCommit()) {
txObject.setMustRestoreAutoCommit(true);
// 交給Spring控制事務(wù)提交
con.setAutoCommit(false);
}
this.prepareTransactionalConnection(con, definition);
//設(shè)置當(dāng)前線程的事務(wù)激活狀態(tài)
txObject.getConnectionHolder().setTransactionActive(true);
int timeout = this.determineTimeout(definition);
if (timeout != -1) {
// 設(shè)置超時(shí)時(shí)間
txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
}
if (txObject.isNewConnectionHolder()) {
TransactionSynchronizationManager.bindResource(this.getDataSource(), txObject.getConnectionHolder());
}
} catch (Throwable var7) {
if (txObject.isNewConnectionHolder()) {
DataSourceUtils.releaseConnection(con, this.dataSource);
txObject.setConnectionHolder((ConnectionHolder)null, false);
}
throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", var7);
}
}1.5AbstractPlatformTransactionManager.prepareTransactionStatus()創(chuàng)建默認(rèn)Status,如果不需要開(kāi)始事務(wù) （比如SUPPORTS），則返回一個(gè)默認(rèn)的狀態(tài)。
protected final DefaultTransactionStatus prepareTransactionStatus(TransactionDefinition definition, Object transaction, boolean newTransaction, boolean newSynchronization, boolean debug, Object suspendedResources) {
DefaultTransactionStatus status = this.newTransactionStatus(definition, transaction, newTransaction, newSynchronization, debug, suspendedResources);
this.prepareSynchronization(status, definition);
return status;
}
protected DefaultTransactionStatus newTransactionStatus(TransactionDefinition definition, Object transaction, boolean newTransaction, boolean newSynchronization, boolean debug, Object suspendedResources) {
boolean actualNewSynchronization = newSynchronization && !TransactionSynchronizationManager.isSynchronizationActive();
//創(chuàng)建 DefaultTransactionStatus 對(duì)象
return new DefaultTransactionStatus(transaction, newTransaction, actualNewSynchronization, definition.isReadOnly(), debug, suspendedResources);
}1.6AbstractPlatformTransactionManager.handleExistingTransaction()針對(duì)不同的傳播行為做不同的處理方法，比如掛起原事務(wù)開(kāi)啟新事務(wù)等等。
private TransactionStatus handleExistingTransaction(TransactionDefinition definition, Object transaction, boolean debugEnabled) throws TransactionException {
//當(dāng)傳播行為是 NEVER 時(shí)拋出異常
if (definition.getPropagationBehavior() == 5) {
throw new IllegalTransactionStateException("Existing transaction found for transaction marked with propagation 'never'");
} else {
AbstractPlatformTransactionManager.SuspendedResourcesHolder suspendedResources;
boolean newSynchronization;
//當(dāng)傳播方式為NOT_SUPPORTED 時(shí)掛起當(dāng)前事務(wù)，然后在無(wú)事務(wù)的狀態(tài)下運(yùn)行
if (definition.getPropagationBehavior() == 4) {
//掛起事務(wù)
suspendedResources = this.suspend(transaction);
newSynchronization = this.getTransactionSynchronization() == 0;
//返回默認(rèn)status
return this.prepareTransactionStatus(definition, (Object)null, false, newSynchronization, debugEnabled, suspendedResources);
//當(dāng)傳播方式為REQUIRES_NEW時(shí)，掛起當(dāng)前事務(wù)，然后啟動(dòng)新事務(wù)
} else if (definition.getPropagationBehavior() == 3) {
//掛起原事務(wù)
suspendedResources = this.suspend(transaction);
try {
newSynchronization = this.getTransactionSynchronization() != 2;
DefaultTransactionStatus status = this.newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
//啟動(dòng)新的事務(wù)
this.doBegin(transaction, definition);
this.prepareSynchronization(status, definition);
return status;
} catch (Error|RuntimeException var7) {
this.resumeAfterBeginException(transaction, suspendedResources, var7);
throw var7;
}
} else {
boolean newSynchronization;
//當(dāng)傳播方式為NESTED時(shí)，設(shè)置事務(wù)的保存點(diǎn)
//存在事務(wù)，將該事務(wù)標(biāo)注保存點(diǎn)，形成嵌套事務(wù)
//嵌套事務(wù)中的子事務(wù)出現(xiàn)異常不會(huì)影響到父事務(wù)保存點(diǎn)之前的操作
if (definition.getPropagationBehavior() == 6) {
if (!this.isNestedTransactionAllowed()) {
throw new NestedTransactionNotSupportedException("Transaction manager does not allow nested transactions by default - specify 'nestedTransactionAllowed' property with value 'true'");
} else {
if (this.useSavepointForNestedTransaction()) {
DefaultTransactionStatus status = this.prepareTransactionStatus(definition, transaction, false, false, debugEnabled, (Object)null);
//創(chuàng)建保存點(diǎn)，回滾時(shí)，只回滾到該保存點(diǎn)
status.createAndHoldSavepoint();
return status;
} else {
newSynchronization = this.getTransactionSynchronization() != 2;
DefaultTransactionStatus status = this.newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, (Object)null);
//如果不支持保存點(diǎn)，就啟動(dòng)新的事務(wù)
this.doBegin(transaction, definition);
this.prepareSynchronization(status, definition);
return status;
}
}
} else {
newSynchronization = this.getTransactionSynchronization() != 2;
return this.prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, (Object)null);
}
}
}
}3.4.2 回滾事務(wù)
TransactionAspectSupport.completeTransactionAfterThrowing() 判斷事務(wù)是否存在，如不存在就不需要回滾，如果存在則在判斷是否滿足回滾條件。
protected void completeTransactionAfterThrowing(TransactionAspectSupport.TransactionInfo txInfo, Throwable ex) {
//判斷是否存在事務(wù)
if (txInfo != null && txInfo.hasTransaction()) {
// 判斷是否滿足回滾條件。拋出的異常類型，和定義的回滾規(guī)則進(jìn)行匹配
if (txInfo.transactionAttribute.rollbackOn(ex)) {
try {
// 回滾處理
txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
}
//省略代碼
} else {
try {
//不滿足回滾條件 出現(xiàn)異常
txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
}
//省略代碼
}
}
}1 AbstractPlatformTransactionManager.rollback()當(dāng)在事務(wù)執(zhí)行過(guò)程中出現(xiàn)異常或其他需要回滾的情況時(shí)，就會(huì)調(diào)用這個(gè)方法，將事務(wù)進(jìn)行回滾操作，撤銷之前所做的數(shù)據(jù)庫(kù)操作，以保證數(shù)據(jù)的一致性。
public final void rollback(TransactionStatus status) throws TransactionException {
//判斷事務(wù)是否已經(jīng)完成，回滾時(shí)拋出異常
if (status.isCompleted()) {
throw new IllegalTransactionStateException("Transaction is already completed - do not call commit or rollback more than once per transaction");
} else {
DefaultTransactionStatus defStatus = (DefaultTransactionStatus)status;
// 執(zhí)行回滾操作。
this.processRollback(defStatus);
}
}1.1AbstractPlatformTransactionManager.processRollback()方法主要用于處理事務(wù)的回滾操作。通過(guò)這個(gè)方法，可以確保事務(wù)在需要回滾時(shí)能夠正確地執(zhí)行回滾操作，保持?jǐn)?shù)據(jù)的完整性。
private void processRollback(DefaultTransactionStatus status) {
try {
try {
//解綁線程和會(huì)話綁定關(guān)系
this.triggerBeforeCompletion(status);
if (status.hasSavepoint()) {
//如果有保存點(diǎn)（當(dāng)前事務(wù)為單獨(dú)的線程則會(huì)退到保存點(diǎn)）
status.rollbackToHeldSavepoint();
} else if (status.isNewTransaction()) {
//如果是新事務(wù)直接回滾。調(diào)用數(shù)據(jù)庫(kù)連接并調(diào)用rollback方法進(jìn)行回滾。使用底層數(shù)據(jù)庫(kù)連接提供的API
this.doRollback(status);
} else if (status.hasTransaction()) {
if (status.isLocalRollbackOnly() || !this.isGlobalRollbackOnParticipationFailure()) {
//如果當(dāng)前事務(wù)不是獨(dú)立的事務(wù)，則只能等待事務(wù)鏈執(zhí)行完成后再做回滾操作
this.doSetRollbackOnly(status);
}
}
}
//catch 等代碼
// 關(guān)閉會(huì)話，重置屬性
this.triggerAfterCompletion(status, 1);
} finally {
//清理并恢復(fù)掛起的事務(wù)
this.cleanupAfterCompletion(status);
}
}3.4.3 提交事務(wù)
TransactionAspectSupport.commitTransactionAfterReturning() 基本上和回滾一樣，都是先判斷是否有事務(wù)，在操作提交。
protected void commitTransactionAfterReturning(TransactionAspectSupport.TransactionInfo txInfo) {
if (txInfo != null && txInfo.hasTransaction()) {
//提交事務(wù)
txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
}
}1) AbstractPlatformTransactionManager.commit() 創(chuàng)建默認(rèn)Status prepareTransactionStatu,發(fā)現(xiàn)是否有回滾標(biāo)記，然后進(jìn)行回滾。如果判斷無(wú)需回滾就可以直接提交。
public final void commit(TransactionStatus status) throws TransactionException {
// 事務(wù)狀態(tài)已完成則拋異常
if (status.isCompleted()) {
throw new IllegalTransactionStateException("Transaction is already completed - do not call commit or rollback more than once per transaction");
} else {
DefaultTransactionStatus defStatus = (DefaultTransactionStatus)status;
//發(fā)現(xiàn)回滾標(biāo)記
if (defStatus.isLocalRollbackOnly()) {
//回滾
this.processRollback(defStatus);
} else if (!this.shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
//回滾
this.processRollback(defStatus);
if (status.isNewTransaction() || this.isFailEarlyOnGlobalRollbackOnly()) {
throw new UnexpectedRollbackException("Transaction rolled back because it has been marked as rollback-only");
}
} else {
// 提交操作
this.processCommit(defStatus);
}
}
}1.1 AbstractPlatformTransactionManager.processCommit()處理事務(wù)的提交操作
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
try {
boolean beforeCompletionInvoked = false;
try {
this.prepareForCommit(status);
this.triggerBeforeCommit(status);
this.triggerBeforeCompletion(status);
beforeCompletionInvoked = true;
boolean globalRollbackOnly = false;
if (status.isNewTransaction() || this.isFailEarlyOnGlobalRollbackOnly()) {
globalRollbackOnly = status.isGlobalRollbackOnly();
}
if (status.hasSavepoint()) {
//釋放保存點(diǎn)信息
status.releaseHeldSavepoint();
} else if (status.isNewTransaction()) {
// 是一個(gè)新的事務(wù) 則提交。 獲取數(shù)據(jù)庫(kù)連接后使用數(shù)據(jù)庫(kù)API進(jìn)行提交事務(wù)
this.doCommit(status);
}
if (globalRollbackOnly) {
throw new UnexpectedRollbackException("Transaction silently rolled back because it has been marked as rollback-only");
}
} catch (TransactionException var20) {
if (this.isRollbackOnCommitFailure()) {
//提交異常回滾
this.doRollbackOnCommitException(status, var20);
} else {
this.triggerAfterCompletion(status, 2);
}
throw var20;
}
//省略其它異常攔截
try {
this.triggerAfterCommit(status);
} finally {
this.triggerAfterCompletion(status, 0);
}
} finally {
// 清理事務(wù)消息
this.cleanupAfterCompletion(status);
}
}3.4.4 清除事務(wù)信息
AbstractPlatformTransactionManager.cleanupAfterCompletion() 這個(gè)方法主要用于在事務(wù)完成后進(jìn)行清理工作。它可能會(huì)負(fù)責(zé)釋放資源、清理臨時(shí)數(shù)據(jù)等，以確保系統(tǒng)處于良好的狀態(tài)。
private void cleanupAfterCompletion(DefaultTransactionStatus status) {
//將當(dāng)前事務(wù)設(shè)置為完成狀態(tài)
status.setCompleted();
if (status.isNewSynchronization()) {
// 清空當(dāng)前事務(wù)消息
TransactionSynchronizationManager.clear();
}
if (status.isNewTransaction()) {
//如果是新事務(wù) 則在事務(wù)完成之后做清理操作
this.doCleanupAfterCompletion(status.getTransaction());
}
if (status.getSuspendedResources() != null) {
// 將原事務(wù)從掛起狀態(tài)恢復(fù)
this.resume(status.getTransaction(), (AbstractPlatformTransactionManager.SuspendedResourcesHolder)status.getSuspendedResources());
}
}1AbstractPlatformTransactionManager.doCleanupAfterCompletion()在新事務(wù)完成后會(huì)調(diào)用resetConnectionAfterTransaction方法重置數(shù)據(jù)庫(kù)連接信息，并判單如果是新的數(shù)據(jù)庫(kù)連接則將其放回連接池。
protected void doCleanupAfterCompletion(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
if (txObject.isNewConnectionHolder()) {
// 將數(shù)據(jù)庫(kù)連接從當(dāng)前線程中解除綁定
TransactionSynchronizationManager.unbindResource(this.dataSource);
}
Connection con = txObject.getConnectionHolder().getConnection();
try {
// 恢復(fù)數(shù)據(jù)庫(kù)連接的autoCommit狀態(tài)
if (txObject.isMustRestoreAutoCommit()) {
con.setAutoCommit(true);
}
// 負(fù)責(zé)重置數(shù)據(jù)庫(kù)連接信息，包括隔離級(jí)別、readOnly屬性等
DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel());
}
catch (Throwable ex) {
logger.debug("Could not reset JDBC Connection after transaction", ex);
}
if (txObject.isNewConnectionHolder()) {
// 如果是新的數(shù)據(jù)庫(kù)連接則將數(shù)據(jù)庫(kù)連接放回連接池
DataSourceUtils.releaseConnection(con, this.dataSource);
}
txObject.getConnectionHolder().clear();
}2) AbstractPlatformTransactionManager.resume() 如果事務(wù)執(zhí)行前有事務(wù)掛起，那么當(dāng)前事務(wù)執(zhí)行結(jié)束后需要將掛起的事務(wù)恢復(fù)，掛起事務(wù)時(shí)保存了原事務(wù)信息，重置了當(dāng)前事務(wù)信息，所以恢復(fù)操作就是將當(dāng)前的事務(wù)信息設(shè)置為之前保存的原事務(wù)信息。
protected final void resume(Object transaction, AbstractPlatformTransactionManager.SuspendedResourcesHolder resourcesHolder) throws TransactionException {
if (resourcesHolder != null) {
Object suspendedResources = resourcesHolder.suspendedResources;
if (suspendedResources != null) {
// 執(zhí)行 恢復(fù)掛起事務(wù) ，綁定資源bindResource
this.doResume(transaction, suspendedResources);
}
List<TransactionSynchronization> suspendedSynchronizations = resourcesHolder.suspendedSynchronizations;
if (suspendedSynchronizations != null) {
TransactionSynchronizationManager.setActualTransactionActive(resourcesHolder.wasActive);
TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(resourcesHolder.isolationLevel);
TransactionSynchronizationManager.setCurrentTransactionReadOnly(resourcesHolder.readOnly);
TransactionSynchronizationManager.setCurrentTransactionName(resourcesHolder.name);
this.doResumeSynchronization(suspendedSynchronizations);
}
}
}3) TransactionAspectSupport.cleanupTransactionInfo()清除當(dāng)前節(jié)點(diǎn)的事務(wù)消息，將舊事務(wù)節(jié)點(diǎn)信息通過(guò)thradLoacl更新到當(dāng)前線程。
protected void cleanupTransactionInfo(TransactionAspectSupport.TransactionInfo txInfo) {
if (txInfo != null) {
//從當(dāng)前線程的 ThreadLocal 獲取上層的事務(wù)信息，將當(dāng)前事務(wù)出棧，繼續(xù)執(zhí)行上層事務(wù)
txInfo.restoreThreadLocalStatus();
}
}
private void restoreThreadLocalStatus() {
//當(dāng)前事務(wù)處理完之后，恢復(fù)上層事務(wù)上下文
TransactionAspectSupport.transactionInfoHolder.set(this.oldTransactionInfo);
}總結(jié)
如果方法正常執(zhí)行完成且沒(méi)有異常，調(diào)用commitTransactionAfterReturning()方法。如果執(zhí)行中出現(xiàn)異常，調(diào)用completeTransactionAfterThrowing()方法。
兩個(gè)方法內(nèi)部都會(huì)判斷是否存在事務(wù)以及是否滿足回滾條件來(lái)決定最終執(zhí)行提交操作還是回滾操作。
上述例子1.1中，未添加事務(wù)，不受事務(wù)控制，因此修改操作生效。
上述例子1.2中，因加入事務(wù)注解，指定回滾異常類型，在completeTransactionAfterThrowing()中邏輯判斷需要回滾，調(diào)用txInfo.getTransactionManager().rollback()方法執(zhí)行回滾操作，因此修改操作不生效。
常見(jiàn)事務(wù)失效的原因及解決方案
場(chǎng)景一：同類中非事務(wù)方法調(diào)用事務(wù)方法
@Service
public class MyService {
@Transactional
public void transactionalMethod() {
// 事務(wù)邏輯
}
public void nonTransactionalCall() {
transactionalMethod(); // 非事務(wù)方法中調(diào)用事務(wù)方法
}
}
失效原因：事務(wù)是基于動(dòng)態(tài)代理實(shí)現(xiàn)的，但本類中調(diào)用另一個(gè)方法默認(rèn)是this調(diào)用關(guān)系，并非動(dòng)態(tài)代理，故失效
解決方案：要么將操作移動(dòng)到事務(wù)中，要么調(diào)用另一個(gè)Service中的事務(wù)方法
@Service
public class MyService {
@Resource
MyService2 myService2;
public void nonTransactionalCall() {
myService2.transactionalMethod(); // 非事務(wù)方法中調(diào)用事務(wù)方法
}
}
@Service
public class MyService2 {
@Transactional
public void transactionalMethod() {
// 事務(wù)邏輯
}
}場(chǎng)景二：事務(wù)方法中拋出檢查型異常且未被正確處理
//失效場(chǎng)景
@Service
public class MyService {
@Transactional
public void transactionalMethod() {
throw new Exception(); // 拋出檢查型異常
}
}
失效原因：@Transactional 注解默認(rèn)處理RuntimeException，即只有拋出運(yùn)行時(shí)異常，才會(huì)觸發(fā)事務(wù)回滾
解決方案：@Transactional 設(shè)置為 @Transactional(rollbackFor =Exception.class) 或者直接拋出運(yùn)行時(shí)異常
解決方案1
@Service
public class MyService {
@Transactional
public void transactionalMethod() {
throw new RuntimeException(); // 拋出非檢查型異常
}
}
解決方案2
@Service
public class MyService {
@Transactional(rollbackFor =Exception.class)
public void transactionalMethod() {
throw new Exception();
}
}場(chǎng)景三：多線程問(wèn)題
public class MyService {
@Resource
private ValuationMapper valuationMapper;
@Transactional
public void transactionalMethod() {
new Thread(() ->{
Valuation v = new Valuation();
v.setUserName("張三");
valuationMapper.insert(v);
}).start();
}
}
原因：Spring的事務(wù)是通過(guò)數(shù)據(jù)庫(kù)連接來(lái)實(shí)現(xiàn)不同線程使用不同的數(shù)據(jù)庫(kù)連接，且放在ThreadLocal中，基于同一個(gè)數(shù)據(jù)庫(kù)連接的事務(wù)才能同時(shí)提交或回滾，多線程場(chǎng)景下，拿到的數(shù)據(jù)庫(kù)連接不是同一個(gè)
解決方案：
1.采用分布式事務(wù)保證
2.自己實(shí)現(xiàn)事務(wù)回滾場(chǎng)景四：錯(cuò)誤使用事務(wù)傳播特性
public class MyService {
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void methodA() {
// 一些事務(wù)操作
methodB();
}
@Transactional(propagation = Propagation.NOT_SUPPORTED)
public void methodB() {
// 這里不會(huì)執(zhí)行事務(wù)操作，因?yàn)?methodA 中傳播特性配置錯(cuò)誤
}
public static void main(String[] args) {
MyService example = new MyService();
example.methodA();
}
}
原因：methodA 配置為 REQUIRES_NEW，而 methodB 配置為 NOT_SUPPORTED 不支持事務(wù)傳播特性，導(dǎo)致事務(wù)傳播特性失效
解決方案：修改 methodB 傳播特性關(guān)于作者
張?jiān)苿? 采貨俠JAVA開(kāi)發(fā)工程師
大道至簡(jiǎn)。再?gòu)?fù)雜的技術(shù)，歸根結(jié)底也是在基礎(chǔ)技術(shù)之上實(shí)現(xiàn)的。
責(zé)任編輯：武曉燕
來(lái)源：
轉(zhuǎn)轉(zhuǎn)技術(shù)
