spring技術內幕讀書筆記--事務處理
阿新 • • 發佈：2019-01-28
事務模組類結構
宣告式事務
宣告式事務處理大致分為以下幾個步驟
1）從配置檔案中讀取事務配置，並將事務元資料轉化為spring內部元資料物件。
涉及類：TransactionAttributeSourceAdvisor（此通知器類會將事務處理的屬性資訊抽象到TransactionAttribute物件中）
2）將事務處理物件與當前執行緒繫結，通過TransactionInfo和TransactionStatus這兩個資料物件進行事務處理過程中相關執行場景的記錄和傳遞。
3）PlatformTransactionManager對事務的具體實現。
- 宣告式事務處理啟動
宣告式事務配置
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource" />
</bean>
<bean id="toBeTestService" class="org.springframework.transaction.interceptor.TransactionProxyFactoryBean" transactionAttributes中的屬性為事務處理的攔截規則，由NameMatchTransactionAttributeSource實現，規則會被封裝到一個map中
如：key為insert*，value為PROPAGATION_REQUIRED
TransactionProxyFactoryBean是個FactoryBean，是生成具有事務特性代理類的入口。
TransactionProxyFactoryBean繼承關係
先大致看一下事務處理代理物件的建立流程
下面來具體分析
TransactionProxyFactoryBean實現了InitializingBean與FactoryBean，所以TransactionProxyFactoryBean的IoC容器初始化方法為afterPropertiesSet方法。
@Override
public void afterPropertiesSet() {
if (this.target == null) {
throw new IllegalArgumentException("Property 'target' is required");
}
if (this.target instanceof String) {
throw new IllegalArgumentException("'target' needs to be a bean reference, not a bean name as value");
}
if (this.proxyClassLoader == null) {
this.proxyClassLoader = ClassUtils.getDefaultClassLoader();
}
//建立事務代理物件
ProxyFactory proxyFactory = new ProxyFactory();
//設定自定義前置攔截器
if (this.preInterceptors != null) {
for (Object interceptor : this.preInterceptors) {
proxyFactory.addAdvisor(this.advisorAdapterRegistry.wrap(interceptor));
}
}
//建立事務處理的AOP通知器
proxyFactory.addAdvisor(this.advisorAdapterRegistry.wrap(createMainInterceptor()));
//設定自定義後置攔截器
if (this.postInterceptors != null) {
for (Object interceptor : this.postInterceptors) {
proxyFactory.addAdvisor(this.advisorAdapterRegistry.wrap(interceptor));
}
}
proxyFactory.copyFrom(this);
//建立AOP的目標源
TargetSource targetSource = createTargetSource(this.target);
proxyFactory.setTargetSource(targetSource);
if (this.proxyInterfaces != null) {
proxyFactory.setInterfaces(this.proxyInterfaces);
}
else if (!isProxyTargetClass()) {
// Rely on AOP infrastructure to tell us what interfaces to proxy.
proxyFactory.setInterfaces(
ClassUtils.getAllInterfacesForClass(targetSource.getTargetClass(), this.proxyClassLoader));
}
postProcessProxyFactory(proxyFactory);
//建立事務代理物件
this.proxy = proxyFactory.getProxy(this.proxyClassLoader);
}
/**
* 建立事務處理的AOP通知器
*/
protected Object createMainInterceptor() {
this.transactionInterceptor.afterPropertiesSet();
if (this.pointcut != null) {
//使用預設通知器，並配置事務處理攔截器
return new DefaultPointcutAdvisor(this.pointcut, this.transactionInterceptor);
}
else {
//若沒有配置切點，使用TransactionAttributeSourceAdvisor通知器，並配置事務處理攔截器
return new TransactionAttributeSourceAdvisor(this.transactionInterceptor);
}
}- 事務處理通知器初始化
這裡通過TransactionAttributeSourceAdvisor通知器來進行分析
Advisor需要有兩個元素，切點（Pointcut）和切面（Interceptor）這裡的切點為TransactionAttributeSourceAdvisor類中宣告的TransactionAttributeSourcePointcut，並通過TransactionAttributeSourcePointcut內部的matches方法來做目標方法是否攔截的規則匹配
@Override
public boolean matches(Method method, Class<?> targetClass) {
if (TransactionalProxy.class.isAssignableFrom(targetClass)) {
return false;
}
//從TransactionInterceptor中獲取TransactionAttributeSource（宣告中配置的攔截規則）物件，具體實現為
//NameMatchTransactionAttributeSource實現類
TransactionAttributeSource tas = getTransactionAttributeSource();
return (tas == null || tas.getTransactionAttribute(method, targetClass) != null);
}具體的匹配規則實現（NameMatchTransactionAttributeSource.getTransactionAttribute）
/**
* 根據方法名進行正則匹配看是否有對應的事務屬性物件（主要是一些配置的隔離級別，readonly等等），若呢個匹配
* 到具體的TransactionAttribute物件，說明TransactionInterceptor做好了對呼叫目標的方法新增事務處理
* 的準備
*/
@Override
public TransactionAttribute getTransactionAttribute(Method method, Class<?> targetClass) {
if (!ClassUtils.isUserLevelMethod(method)) {
return null;
}
// Look for direct name match.
String methodName = method.getName();
TransactionAttribute attr = this.nameMap.get(methodName);
if (attr == null) {
// Look for most specific name match.
String bestNameMatch = null;
for (String mappedName : this.nameMap.keySet()) {
if (isMatch(methodName, mappedName) &&
(bestNameMatch == null || bestNameMatch.length() <= mappedName.length())) {
attr = this.nameMap.get(mappedName);
bestNameMatch = mappedName;
}
}
}
return attr;
}- 事務攔截器的具體實現
由於事務AOP是通過CglibAopProxy實現的，大概說明下呼叫流程
所以我們看TransactionInterceptor中的invoke方法
@Override
public Object invoke(final MethodInvocation invocation) throws Throwable {
Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
return invokeWithinTransaction(invocation.getMethod(), targetClass, new InvocationCallback() {
@Override
public Object proceedWithInvocation() throws Throwable {
return invocation.proceed();
}
});
}protected Object invokeWithinTransaction(Method method, Class<?> targetClass, final InvocationCallback invocation)
throws Throwable {
// 通過TransactionAttributeSource物件取得事務的屬性配置
final TransactionAttribute txAttr = getTransactionAttributeSource().getTransactionAttribute(method, targetClass);
// 根據TransactionProxyFactoryBean的配置資訊獲得具體的事務處理器
final PlatformTransactionManager tm = determineTransactionManager(txAttr);
final String joinpointIdentification = methodIdentification(method, targetClass);
// 不同型別的PlatformTransactionManager呼叫方式不同，CallbackPreferringPlatformTransactionManager需要回調函式來實現事務流程，而我們常用的DataSourceTransactionManager就不是CallbackPreferringPlatformTransactionManager
if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
// 建立事務物件，其中儲存了事務狀態物件
// TransactionInfo是一個事務物件，物件中包含事務的所有屬性，包括PlatformTransactionManager、TransactionAttribute、TransactionStatus
// ⚠️事務的管理都是通過TransactionInfo物件來完成，它封裝了事務物件和事務處理的狀態資訊
TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
Object retVal = null;
try {
// 進行攔截器和目標方法的呼叫
retVal = invocation.proceedWithInvocation();
}
catch (Throwable ex) {
// 回滾
completeTransactionAfterThrowing(txInfo, ex);
throw ex;
}
finally {
cleanupTransactionInfo(txInfo);
}
// 提交
commitTransactionAfterReturning(txInfo);
return retVal;
}
else {
// It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
try {
Object result = ((CallbackPreferringPlatformTransactionManager) tm).execute(txAttr,
new TransactionCallback<Object>() {
@Override
public Object doInTransaction(TransactionStatus 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.
return new ThrowableHolder(ex);
}
}
finally {
cleanupTransactionInfo(txInfo);
}
}
});
// Check result: It might indicate a Throwable to rethrow.
if (result instanceof ThrowableHolder) {
throw ((ThrowableHolder) result).getThrowable();
}
else {
return result;
}
}
catch (ThrowableHolderException ex) {
throw ex.getCause();
}
}
}流程圖
1）讀取事務方法的事務屬性配置
2）獲取PlatformTransactionManager事務處理器的具體實現
3）決定是否建立新的事務（後面會具體分析）
4）對攔截器和目標物件進行呼叫
5）根據執行進行commit或rollback
- 事務的建立
程式設計式建立事務
DataSourceTransactionManager manager = new DataSourceTransactionManager();
TransactionDefinition td = new DefaultTransactionDefinition();
TransactionStatus transaction = manager.getTransaction(td);
try{
//todo
}catch (Exception e){
manager.rollback(transaction);
}
manager.commit(transaction);事務建立流程圖
大致過程一句話：通過DataSourceTransactionManager建立一個TransactionStatus，並將TransactionStatus設定到當前的TransactionInfo中，同時將此TransactionInfo與當前執行緒繫結。
下面我們來分析DataSourceTransactionManager.getTransaction方法
@Override
public final TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {
// 獲取ConnectionHolder資料來源連線物件
Object transaction = doGetTransaction();
boolean debugEnabled = logger.isDebugEnabled();
if (definition == null) {
definition = new DefaultTransactionDefinition();
}
// 是否是巢狀事務，若是，則根據事務的傳播屬性來處理事務的產生
if (isExistingTransaction(transaction)) {
return handleExistingTransaction(definition, transaction, debugEnabled);
}
// 檢查事務屬性中timeout的設定是否合理
if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout());
}
// 根據事務的傳播屬性進行相應的處理
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);
// 建立TransactionStatus
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
// 開啟事務（呼叫connect物件的setAutoCommit(false)和setTransactionActive(true)）
doBegin(transaction, definition);
// 初始化當前執行緒的threadlocal物件
prepareSynchronization(status, definition);
return status;
}
catch (RuntimeException ex) {
resume(null, suspendedResources);
throw ex;
}
catch (Error err) {
resume(null, suspendedResources);
throw err;
}
}
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);
}
}建立新事物比較好理解，根據事務屬性的配置建立事務物件並儲存到TransactionStatus中，並將事務屬性資訊與當前執行緒繫結。
對於巢狀式事務，需要關注下AbstractPlatformTransactionManager.handleExistingTransaction方法，會根據事務屬性中配置的傳播屬性進行事務的處理。
private TransactionStatus handleExistingTransaction(
TransactionDefinition definition, Object transaction, boolean debugEnabled)
throws TransactionException {
// 事務傳播屬性為不容許存在事務
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
throw new IllegalTransactionStateException(
"Existing transaction found for transaction marked with propagation 'never'");
}
// 事務傳播屬性為不開啟新事務
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
if (debugEnabled) {
logger.debug("Suspending current transaction");
}
// 將當前事務掛起
Object suspendedResources = suspend(transaction);
boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
// 這裡transaction為null，newTransaction為false，說明事務方法不需要放在事務環境中執行，會將之前的事務資訊儲存在當前的TransactionStatus物件中
return prepareTransactionStatus(
definition, null, false, newSynchronization, debugEnabled, suspendedResources);
}
// 事務傳播屬性為開啟新事物
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 {
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
doBegin(transaction, definition);
prepareSynchronization(status, definition);
return status;
}
catch (RuntimeException beginEx) {
resumeAfterBeginException(transaction, suspendedResources, beginEx);
throw beginEx;
}
catch (Error beginErr) {
resumeAfterBeginException(transaction, suspendedResources, beginErr);
throw beginErr;
}
}
// 巢狀事務的建立
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);
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.
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, null);
doBegin(transaction, definition);
prepareSynchronization(status, definition);
return status;
}
}
// 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");
}
}
}
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
// 第三個引數false代表當前事務方法沒有使用新事務
return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
}- 事務的掛起
事務的掛起原理機制大概理解為
1）獲取上一次事務的資料來源物件封裝並返回SuspendedResourcesHolder物件
2）呼叫doBegin方法建立新的資料來源連線物件與新的事務資訊物件
2）將SuspendedResourcesHolder設定到新的TransactionStatus中
protected final SuspendedResourcesHolder suspend(Object transaction) throws TransactionException {
if (TransactionSynchronizationManager.isSynchronizationActive()) {
List<TransactionSynchronization> suspendedSynchronizations = doSuspendSynchronization();
try {
Object suspendedResources = null;
if (transaction != null) {
// 獲取上一次事務的資料來源物件
suspendedResources = doSuspend(transaction);
}
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);
// 封裝並返回上一次事務的事務物件
return new SuspendedResourcesHolder(
suspendedResources, suspendedSynchronizations, name, readOnly, isolationLevel, wasActive);
}
catch (RuntimeException ex) {
// doSuspend failed - original transaction is still active...
doResumeSynchronization(suspendedSynchronizations);
throw ex;
}
catch (Error err) {
// doSuspend failed - original transaction is still active...
doResumeSynchronization(suspendedSynchronizations);
throw err;
}
}
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;
}
}@Override
protected Object doSuspend(Object transaction) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
txObject.setConnectionHolder(null);
// 獲取上一次事務的資料來源連線物件，並將其從當前ThreadLocal中移除
ConnectionHolder conHolder = (ConnectionHolder)
TransactionSynchronizationManager.unbindResource(this.dataSource);
return conHolder;
}- 事務的提交
直接看事務提交的流程AbstractPlatformTransactionManager.processCommit
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
try {
boolean beforeCompletionInvoked = false;
try {
// 事務提交的準備工作，如果集成了mybatis，事務的真正提交在此方法中完成，spring的事務物件僅僅做事務狀態的記錄
prepareForCommit(status);
triggerBeforeCommit(status);
triggerBeforeCompletion(status);
beforeCompletionInvoked = true;
boolean globalRollbackOnly = false;
if (status.isNewTransaction() || isFailEarlyOnGlobalRollbackOnly()) {
globalRollbackOnly = status.isGlobalRollbackOnly();
}
// 巢狀事務的處理
if (status.hasSavepoint()) {
if (status.isDebug()) {
logger.debug("Releasing transaction savepoint");
}
status.releaseHeldSavepoint();
}
// 是否是一個新事物
else if (status.isNewTransaction()) {
if (status.isDebug()) {
logger.debug("Initiating transaction commit");
}
// 具體的事務提交由具體的事務處理器來完成
doCommit(status);
}
// Throw UnexpectedRollbackException if we have a global rollback-only
// marker but still didn't get a corresponding exception from commit.
if (globalRollbackOnly) {
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 ex) {
if (!beforeCompletionInvoked) {
triggerBeforeCompletion(status);
}
doRollbackOnCommitException(status, ex);
throw ex;
}
catch (Error err) {
if (!beforeCompletionInvoked) {
triggerBeforeCompletion(status);
}
doRollbackOnCommitException(status, err);
throw err;
}
// 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);
}
}⚠️：如果是mybatis整合模式下，spring不負責具體的提交回滾操作，只是做事務狀態的記錄！！具體可以參考mybatis原始碼org.mybatis.spring.SqlSessionUtils.SqlSessionSynchronization類
- 事務回滾
直接看程式碼吧
private void processRollback(DefaultTransactionStatus status) {
try {
try {
// 若集成了mybatis，真正資料來源的回滾操作發生在這裡
triggerBeforeCompletion(status);
if (status.hasSavepoint()) {
if (status.isDebug()) {
logger.debug("Rolling back transaction to savepoint");
}
status.rollbackToHeldSavepoint();
}
else if (status.isNewTransaction()) {
if (status.isDebug()) {
logger.debug("Initiating transaction rollback");
}
doRollback(status);
}
else if (status.hasTransaction()) {
if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) {
if (status.isDebug()) {
logger.debug("Participating transaction failed - marking existing transaction as rollback-only");
}
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");
}
}
catch (RuntimeException ex) {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
throw ex;
}
catch (Error err) {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
throw err;
}
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
}
finally {
cleanupAfterCompletion(status);
}
}PlatformTransactionManager具體實現
事務處理最重要的4個方法
1）doGetTrancation：建立資料來源物件並封裝返回事務物件
2）doBegin：開啟事務
3）doCommit：提交
4）doRollback：回滾
都是由在TransactionProxyFactoryBean中配置的具體的TransactionManager實現
大致看下DataSourceTransactionManager和HibernateTransactionManager的執行流程，原始碼比較簡單，暫不做分析了
