个人随笔
目录
一个简单的例子来探寻Spring事务Trasanction执行源码的主脉络(二)：执行代理方法
2021-08-14 18:29:50

通过上一篇文章一个简单的例子来探寻SpringAOP执行源码的主脉络(二)：执行代理方法 我们知道了AOP怎么执行代理方法，主要是靠对应的Advisor，我们知道spring会为我们的切面里面的方法比befor，after,around等方法都生成对应的Advisor，所以我们估计spring事务的代理方法逻辑跟AOP是差不多的，主要看的就是对应的Advisor里面的执行逻辑，在一个简单的例子来探寻Spring事务Trasanction执行源码的主脉络(一)：创建事务代理对象文章中，我们已经知道了事务对应的Advisor是BeanFactoryTransactionAttributeSourceAdvisor，那应该是跑不了了。

我们来调试下看看

果然不出预料，那么前面的逻辑肯定是跟AOP差不多的。

  1. public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  2.         Object oldProxy = null;
  3.         boolean setProxyContext = false;
  5.         TargetSource targetSource = this.advised.targetSource;
  6.         Object target = null;
  8.         try {
  9.             if (!this.equalsDefined && AopUtils.isEqualsMethod(method)) {
  10.                 // The target does not implement the equals(Object) method itself.
  11.                 return equals(args[0]);
  12.             }
  13.             else if (!this.hashCodeDefined && AopUtils.isHashCodeMethod(method)) {
  14.                 // The target does not implement the hashCode() method itself.
  15.                 return hashCode();
  16.             }
  17.             else if (method.getDeclaringClass() == DecoratingProxy.class) {
  18.                 // There is only getDecoratedClass() declared -> dispatch to proxy config.
  19.                 return AopProxyUtils.ultimateTargetClass(this.advised);
  20.             }
  21.             else if (!this.advised.opaque && method.getDeclaringClass().isInterface() &&
  22.                     method.getDeclaringClass().isAssignableFrom(Advised.class)) {
  23.                 // Service invocations on ProxyConfig with the proxy config...
  24.                 return AopUtils.invokeJoinpointUsingReflection(this.advised, method, args);
  25.             }
  27.             Object retVal;
  29.             if (this.advised.exposeProxy) {
  30.                 // Make invocation available if necessary.
  31.                 oldProxy = AopContext.setCurrentProxy(proxy);
  32.                 setProxyContext = true;
  33.             }
  35.             // Get as late as possible to minimize the time we "own" the target,
  36.             // in case it comes from a pool.
  37.             target = targetSource.getTarget();
  38.             Class<?> targetClass = (target != null ? target.getClass() : null);
  40.             // Get the interception chain for this method.
  41.             List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);
  43.             // Check whether we have any advice. If we don't, we can fallback on direct
  44.             // reflective invocation of the target, and avoid creating a MethodInvocation.
  45.             if (chain.isEmpty()) {
  46.                 // We can skip creating a MethodInvocation: just invoke the target directly
  47.                 // Note that the final invoker must be an InvokerInterceptor so we know it does
  48.                 // nothing but a reflective operation on the target, and no hot swapping or fancy proxying.
  49.                 Object[] argsToUse = AopProxyUtils.adaptArgumentsIfNecessary(method, args);
  50.                 retVal = AopUtils.invokeJoinpointUsingReflection(target, method, argsToUse);
  51.             }
  52.             else {
  53.                 // We need to create a method invocation...
  54.                 MethodInvocation invocation =
  55.                         new ReflectiveMethodInvocation(proxy, target, method, args, targetClass, chain);
  56.                 // Proceed to the joinpoint through the interceptor chain.
  57.                 retVal = invocation.proceed();
  58.             }
  60.             // Massage return value if necessary.
  61.             Class<?> returnType = method.getReturnType();
  62.             if (retVal != null && retVal == target &&
  63.                     returnType != Object.class && returnType.isInstance(proxy) &&
  64.                     !RawTargetAccess.class.isAssignableFrom(method.getDeclaringClass())) {
  65.                 // Special case: it returned "this" and the return type of the method
  66.                 // is type-compatible. Note that we can't help if the target sets
  67.                 // a reference to itself in another returned object.
  68.                 retVal = proxy;
  69.             }
  70.             else if (retVal == null && returnType != Void.TYPE && returnType.isPrimitive()) {
  71.                 throw new AopInvocationException(
  72.                         "Null return value from advice does not match primitive return type for: " + method);
  73.             }
  74.             return retVal;
  75.         }
  76.         finally {
  77.             if (target != null && !targetSource.isStatic()) {
  78.                 // Must have come from TargetSource.
  79.                 targetSource.releaseTarget(target);
  80.             }
  81.             if (setProxyContext) {
  82.                 // Restore old proxy.
  83.                 AopContext.setCurrentProxy(oldProxy);
  84.             }
  85.         }
  86.     }

不过在这里我们发现

  1. List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);

这里生成的调用链里面的对象是TransactionInterceptor这个是该advisor里面的advice(通知)的invoke方法，在生成代理对象的步骤中也会把这个bean加入到容器中的。前面的调用模式跟AOP一个样，接下来我们直接进入到这个对象的invoke方法

  1. @Override
  2.     @Nullable
  3.     public Object invoke(MethodInvocation invocation) throws Throwable {
  4.         // Work out the target class: may be {@code null}.
  5.         // The TransactionAttributeSource should be passed the target class
  6.         // as well as the method, which may be from an interface.
  7.         Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
  9.         // Adapt to TransactionAspectSupport's invokeWithinTransaction...
  10.         return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
  11.     }

我们直接进入到

  1. return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);

方法中，真相应该就在这里

  1. @Nullable
  2.     protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
  3.             final InvocationCallback invocation) throws Throwable {
  5.         // If the transaction attribute is null, the method is non-transactional.
  6.         //这个是我们的那个advisor
  7.         TransactionAttributeSource tas = getTransactionAttributeSource();
  8.         final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
  9.         //这个是我们之前指定的：DataSourceTransactionManager
  10.         final TransactionManager tm = determineTransactionManager(txAttr);
  12.         if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {
  13.             ReactiveTransactionSupport txSupport = this.transactionSupportCache.computeIfAbsent(method, key -> {
  14.                 if (KotlinDetector.isKotlinType(method.getDeclaringClass()) && KotlinDelegate.isSuspend(method)) {
  15.                     throw new TransactionUsageException(
  16.                             "Unsupported annotated transaction on suspending function detected: " + method +
  17.                             ". Use TransactionalOperator.transactional extensions instead.");
  18.                 }
  19.                 ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(method.getReturnType());
  20.                 if (adapter == null) {
  21.                     throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " +
  22.                             method.getReturnType());
  23.                 }
  24.                 return new ReactiveTransactionSupport(adapter);
  25.             });
  26.             return txSupport.invokeWithinTransaction(
  27.                     method, targetClass, invocation, txAttr, (ReactiveTransactionManager) tm);
  28.         }
  30.         PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
  31.         final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
  33.         if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
  34.             // Standard transaction demarcation with getTransaction and commit/rollback calls.
  35.             //这里是创建事务，如果需要的话，也就是如果你在被代理的方法那里加了@Transation注解的话，这里就会创建一个事务
  36.             TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
  38.             Object retVal;
  39.             try {
  40.                 // This is an around advice: Invoke the next interceptor in the chain.
  41.                 // This will normally result in a target object being invoked.
  42.                 retVal = invocation.proceedWithInvocation();
  43.             }
  44.             catch (Throwable ex) {
  45.                 // target invocation exception
  46.                 completeTransactionAfterThrowing(txInfo, ex);
  47.                 throw ex;
  48.             }
  49.             finally {
  50.                 cleanupTransactionInfo(txInfo);
  51.             }
  53.             if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
  54.                 // Set rollback-only in case of Vavr failure matching our rollback rules...
  55.                 TransactionStatus status = txInfo.getTransactionStatus();
  56.                 if (status != null && txAttr != null) {
  57.                     retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
  58.                 }
  59.             }
  61.             commitTransactionAfterReturning(txInfo);
  62.             return retVal;
  63.         }
  65.         else {
  66.             Object result;
  67.             final ThrowableHolder throwableHolder = new ThrowableHolder();
  69.             // It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
  70.             try {
  71.                 result = ((CallbackPreferringPlatformTransactionManager) ptm).execute(txAttr, status -> {
  72.                     TransactionInfo txInfo = prepareTransactionInfo(ptm, txAttr, joinpointIdentification, status);
  73.                     try {
  74.                         Object retVal = invocation.proceedWithInvocation();
  75.                         if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
  76.                             // Set rollback-only in case of Vavr failure matching our rollback rules...
  77.                             retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
  78.                         }
  79.                         return retVal;
  80.                     }
  81.                     catch (Throwable ex) {
  82.                         if (txAttr.rollbackOn(ex)) {
  83.                             // A RuntimeException: will lead to a rollback.
  84.                             if (ex instanceof RuntimeException) {
  85.                                 throw (RuntimeException) ex;
  86.                             }
  87.                             else {
  88.                                 throw new ThrowableHolderException(ex);
  89.                             }
  90.                         }
  91.                         else {
  92.                             // A normal return value: will lead to a commit.
  93.                             throwableHolder.throwable = ex;
  94.                             return null;
  95.                         }
  96.                     }
  97.                     finally {
  98.                         cleanupTransactionInfo(txInfo);
  99.                     }
  100.                 });
  101.             }
  102.             catch (ThrowableHolderException ex) {
  103.                 throw ex.getCause();
  104.             }
  105.             catch (TransactionSystemException ex2) {
  106.                 if (throwableHolder.throwable != null) {
  107.                     logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
  108.                     ex2.initApplicationException(throwableHolder.throwable);
  109.                 }
  110.                 throw ex2;
  111.             }
  112.             catch (Throwable ex2) {
  113.                 if (throwableHolder.throwable != null) {
  114.                     logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
  115.                 }
  116.                 throw ex2;
  117.             }
  119.             // Check result state: It might indicate a Throwable to rethrow.
  120.             if (throwableHolder.throwable != null) {
  121.                 throw throwableHolder.throwable;
  122.             }
  123.             return result;
  124.         }
  125.     }

我们先去看看创建事务的逻辑

  1. @SuppressWarnings("serial")
  2.     protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
  3.             @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
  5.         // If no name specified, apply method identification as transaction name.
  6.         if (txAttr != null && txAttr.getName() == null) {
  7.             txAttr = new DelegatingTransactionAttribute(txAttr) {
  8.                 @Override
  9.                 public String getName() {
  10.                     return joinpointIdentification;
  11.                 }
  12.             };
  13.         }
  15.         TransactionStatus status = null;
  16.         if (txAttr != null) {
  17.             if (tm != null) {
  18.                 status = tm.getTransaction(txAttr);
  19.             }
  20.             else {
  21.                 if (logger.isDebugEnabled()) {
  22.                     logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
  23.                             "] because no transaction manager has been configured");
  24.                 }
  25.             }
  26.         }
  27.         return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
  28.     }

继续进入tm.getTransaction(txAttr);

  1. @Override
  2.     public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
  3.             throws TransactionException {
  5.         // Use defaults if no transaction definition given.
  6.         //这里默认的是PROPAGATION_REQUIRED,ISOLATION_DEFAULT
  7.         TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
  9.         Object transaction = doGetTransaction();
  10.         boolean debugEnabled = logger.isDebugEnabled();
  12.         if (isExistingTransaction(transaction)) {
  13.             // Existing transaction found -> check propagation behavior to find out how to behave.
  14.             return handleExistingTransaction(def, transaction, debugEnabled);
  15.         }
  17.         // Check definition settings for new transaction.
  18.         if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
  19.             throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
  20.         }
  22.         // No existing transaction found -> check propagation behavior to find out how to proceed.
  23.         if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
  24.             throw new IllegalTransactionStateException(
  25.                     "No existing transaction found for transaction marked with propagation 'mandatory'");
  26.         }
  27.         else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
  28.                 def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
  29.                 def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
  30.             SuspendedResourcesHolder suspendedResources = suspend(null);
  31.             if (debugEnabled) {
  32.                 logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
  33.             }
  34.             try {
  35.                 //这里会开启一个事务
  36.                 return startTransaction(def, transaction, debugEnabled, suspendedResources);
  37.             }
  38.             catch (RuntimeException | Error ex) {
  39.                 resume(null, suspendedResources);
  40.                 throw ex;
  41.             }
  42.         }
  43.         else {
  44.             // Create "empty" transaction: no actual transaction, but potentially synchronization.
  45.             if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
  46.                 logger.warn("Custom isolation level specified but no actual transaction initiated; " +
  47.                         "isolation level will effectively be ignored: " + def);
  48.             }
  49.             boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
  50.             return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
  51.         }
  52.     }

我们具体看看怎么开启事务

  1. /**
  2.      * Start a new transaction.
  3.      */
  4.     private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
  5.             boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
  7.         boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
  8.         DefaultTransactionStatus status = newTransactionStatus(
  9.                 definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
  10.         doBegin(transaction, definition);
  11.         prepareSynchronization(status, definition);
  12.         return status;
  13.     }

当然是doBegin方法啦

  1. @Override
  2.     protected void doBegin(Object transaction, TransactionDefinition definition) {
  3.         DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
  4.         Connection con = null;
  6.         try {
  7.             if (!txObject.hasConnectionHolder() ||
  8.                     txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
  9.                 Connection newCon = obtainDataSource().getConnection();
  10.                 if (logger.isDebugEnabled()) {
  11.                     logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
  12.                 }
  13.                 txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
  14.             }
  16.             txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
  17.             //这里获取一个数据库连接
  18.             con = txObject.getConnectionHolder().getConnection();
  20.             Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
  21.             txObject.setPreviousIsolationLevel(previousIsolationLevel);
  22.             txObject.setReadOnly(definition.isReadOnly());
  24.             // Switch to manual commit if necessary. This is very expensive in some JDBC drivers,
  25.             // so we don't want to do it unnecessarily (for example if we've explicitly
  26.             // configured the connection pool to set it already).
  27.             if (con.getAutoCommit()) {
  28.                 txObject.setMustRestoreAutoCommit(true);
  29.                 if (logger.isDebugEnabled()) {
  30.                     logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
  31.                 }
  32.                 //关闭自动提交：这里就开启了事务，事务的提交由我们自己控制
  33.                 con.setAutoCommit(false);
  34.             }
  36.             prepareTransactionalConnection(con, definition);
  37.             txObject.getConnectionHolder().setTransactionActive(true);
  39.             int timeout = determineTimeout(definition);
  40.             if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
  41.                 txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
  42.             }
  44.             // Bind the connection holder to the thread.
  45.             if (txObject.isNewConnectionHolder()) {
  46.                 TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
  47.             }
  48.         }
  50.         catch (Throwable ex) {
  51.             if (txObject.isNewConnectionHolder()) {
  52.                 DataSourceUtils.releaseConnection(con, obtainDataSource());
  53.                 txObject.setConnectionHolder(null, false);
  54.             }
  55.             throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
  56.         }
  57.     }

找到了我们的关键性代码

  1. con.setAutoCommit(false);

关闭了自动提交。好了这里如果需要事务的话就开启，我们这里开启了再回到我们之前的代码

  1. if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
  2.             // Standard transaction demarcation with getTransaction and commit/rollback calls.
  3.             TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
  5.             Object retVal;
  6.             try {
  7.                 // This is an around advice: Invoke the next interceptor in the chain.
  8.                 // This will normally result in a target object being invoked.
  9.                 //这里会先去执行我们的业务逻辑方法
  10.                 retVal = invocation.proceedWithInvocation();
  11.             }
  12.             catch (Throwable ex) {
  13.                 //如果有异常，则会到这里去回滚
  14.                 // target invocation exception
  15.                 completeTransactionAfterThrowing(txInfo, ex);
  16.                 throw ex;
  17.             }
  18.             finally {
  19.                 cleanupTransactionInfo(txInfo);
  20.             }
  22.             if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
  23.                 // Set rollback-only in case of Vavr failure matching our rollback rules...
  24.                 TransactionStatus status = txInfo.getTransactionStatus();
  25.                 if (status != null && txAttr != null) {
  26.                     retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
  27.                 }
  28.             }
  30.             commitTransactionAfterReturning(txInfo);
  31.             return retVal;
  32.         }

开启事务后，这里会先执行业务逻辑方法

  1. retVal = invocation.proceedWithInvocation();

这行代码会回到我们再AOP里面熟悉的

  1.   if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {
  2.             return invokeJoinpoint();
  3.         }

而 invokeJoinpoint()会执行具体的业务逻辑。

如果有抛出异常，这里会被catch方法捕获到

我们看下回滚的逻辑

  1. protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) {
  2.         if (txInfo != null && txInfo.getTransactionStatus() != null) {
  3.             if (logger.isTraceEnabled()) {
  4.                 logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() +
  5.                         "] after exception: " + ex);
  6.             }
  7.             //这里需要检查是否需要回滚
  8.             if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) {
  9.                 try {
  10.                     txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus());
  11.                 }
  12.                 catch (TransactionSystemException ex2) {
  13.                     logger.error("Application exception overridden by rollback exception", ex);
  14.                     ex2.initApplicationException(ex);
  15.                     throw ex2;
  16.                 }
  17.                 catch (RuntimeException | Error ex2) {
  18.                     logger.error("Application exception overridden by rollback exception", ex);
  19.                     throw ex2;
  20.                 }
  21.             }
  22.             else {
  23.                 // We don't roll back on this exception.
  24.                 // Will still roll back if TransactionStatus.isRollbackOnly() is true.
  25.                 try {
  26.                     //如果不是运行期异常，我们还是会回滚的
  27.                     txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
  28.                 }
  29.                 catch (TransactionSystemException ex2) {
  30.                     logger.error("Application exception overridden by commit exception", ex);
  31.                     ex2.initApplicationException(ex);
  32.                     throw ex2;
  33.                 }
  34.                 catch (RuntimeException | Error ex2) {
  35.                     logger.error("Application exception overridden by commit exception", ex);
  36.                     throw ex2;
  37.                 }
  38.             }
  39.         }
  40.     }

我们先判断了一下这个异常是否是

  1. @Override
  2.     public boolean rollbackOn(Throwable ex) {
  3.         return (ex instanceof RuntimeException || ex instanceof Error);
  4.     }

RuntimeException，如果是运行期异常，如果是才会回滚，否则就直接提交。

如果没有抛出异常就会执行到

  1. commitTransactionAfterReturning(txInfo);

很明显就是提交事务啦

  1. /**
  2.      * Execute after successful completion of call, but not after an exception was handled.
  3.      * Do nothing if we didn't create a transaction.
  4.      * @param txInfo information about the current transaction
  5.      */
  6.     protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) {
  7.         if (txInfo != null && txInfo.getTransactionStatus() != null) {
  8.             if (logger.isTraceEnabled()) {
  9.                 logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "]");
  10.             }
  11.             txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
  12.         }
  13.     }

好了！其实总结一句，看AOP还是事务，还是别的什么组件，主要是看Advisor里面的advice。

 281

啊！这个可能是世界上最丑的留言输入框功能~


当然，也是最丑的留言列表

有疑问发邮件到 : suibibk@qq.com 侵权立删
Copyright : 个人随笔   备案号 : 粤ICP备18099399号-2