Spring AOP实现声明式事务机制源码解析

@GlobalTransactional
public boolean purchase(long accountId, long stockId, long quantity) {
    String xid = RootContext.getXID();
    LOGGER.info("New Transaction Begins: " + xid);
    boolean stockResult = reduceAccount(accountId,stockId, quantity);
    if (!stockResult) {
        throw new RuntimeException("账号服务调用失败,事务回滚!");
    }
    Long orderId = createOrder(accountId, stockId, quantity);
    if (orderId == null || orderId <= 0) {
        throw new RuntimeException("订单服务调用失败,事务回滚!");
    }
    return true;
}

GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
try {
    tx.begin(xxx);
    ...
    purchase(xxx)//给purchase增加全局事务处理能力
    ...
    tx.commit();
} catch (Exception exx) {
    tx.rollback();
    throw exx;
}

createProxy:443, AbstractAutoProxyCreator (org.springframework.aop.framework.autoproxy)
wrapIfNecessary:344, AbstractAutoProxyCreator (org.springframework.aop.framework.autoproxy)
wrapIfNecessary:307, GlobalTransactionScanner (io.seata.spring.annotation)
postProcessAfterInitialization:293, AbstractAutoProxyCreator (org.springframework.aop.framework.autoproxy)
applyBeanPostProcessorsAfterInitialization:455, AbstractAutowireCapableBeanFactory (org.springframework.beans.factory.support)
initializeBean:1808, AbstractAutowireCapableBeanFactory (org.springframework.beans.factory.support)
doCreateBean:620, AbstractAutowireCapableBeanFactory (org.springframework.beans.factory.support)
createBean:542, AbstractAutowireCapableBeanFactory (org.springframework.beans.factory.support)

protected Object createProxy(Class<?> beanClass, @Nullable String beanName,
        @Nullable Object[] specificInterceptors, TargetSource targetSource) {
    if (this.beanFactory instanceof ConfigurableListableBeanFactory) {
        // 为目标 Bean 的 BeanDefinition 对象设置一个属性
        // org.springframework.aop.framework.autoproxy.AutoProxyUtils.originalTargetClass -> 目标 Bean 的 Class 对象
        AutoProxyUtils.exposeTargetClass((ConfigurableListableBeanFactory) this.beanFactory, beanName, beanClass);
    }
    // <1> 创建一个代理工厂
    ProxyFactory proxyFactory = new ProxyFactory();
    // <2> 复制当前 ProxyConfig 的一些属性（例如 proxyTargetClass、exposeProxy）
    proxyFactory.copyFrom(this);
    /**
     * <3> 判断是否类代理，也就是是否开启 CGLIB 代理
     * 默认配置下为 `false`，参考 {@link org.springframework.context.annotation.EnableAspectJAutoProxy}
     */
    if (!proxyFactory.isProxyTargetClass()) {
        /*
         * <3.1> 如果这个 Bean 配置了进行类代理，则设置为 `proxyTargetClass` 为 `true`
         */
        if (shouldProxyTargetClass(beanClass, beanName)) {
            proxyFactory.setProxyTargetClass(true);
        }
        else {
            /*
             * <3.2> 检测当前 Bean 实现的接口是否包含可代理的接口
             * 如没有实现，则将 `proxyTargetClass` 设为 `true`，表示需要进行 CGLIB 提升
             */
            evaluateProxyInterfaces(beanClass, proxyFactory);
        }
    }
    /*
     * <4> 对入参的 Advisor 进一步处理，因为其中可能还存在 Advice 类型，需要将他们包装成 DefaultPointcutAdvisor 对象
     * 如果配置了 `interceptorNames` 拦截器，也会添加进来
     */
    Advisor[] advisors = buildAdvisors(beanName, specificInterceptors);
    // <5> 代理工厂添加 Advisor 数组
    proxyFactory.addAdvisors(advisors);
    // <6> 代理工厂设置 TargetSource 对象
    proxyFactory.setTargetSource(targetSource);
    // <7> 对 ProxyFactory 进行加工处理，抽象方法，目前没有子类实现
    customizeProxyFactory(proxyFactory);
    proxyFactory.setFrozen(this.freezeProxy);
    // <8> 是否这个 AdvisedSupport 配置管理器已经过滤过目标类（默认为 false）
    if (advisorsPreFiltered()) {
        // 设置 `preFiltered` 为 `true`
        // 这样 Advisor 们就不会根据 ClassFilter 进行过滤了，而直接通过 MethodMatcher 判断是否处理被拦截方法
        proxyFactory.setPreFiltered(true);
    }
    // Use original ClassLoader if bean class not locally loaded in overriding class loader
    ClassLoader classLoader = getProxyClassLoader();
    if (classLoader instanceof SmartClassLoader && classLoader != beanClass.getClassLoader()) {
       classLoader = ((SmartClassLoader) classLoader).getOriginalClassLoader();
    }
    // <9> 通过 ProxyFactory 代理工厂创建代理对象
    return proxyFactory.getProxy(getProxyClassLoader());
}

@Override
public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
    Class<?> targetClass =
        methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null;
    Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
    if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
        final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
        // 获取方法上的@GlobalTransactional注解中的内容
        final GlobalTransactional globalTransactionalAnnotation =
            getAnnotation(method, targetClass, GlobalTransactional.class);
        // 获取方法上的@GlobalLock注解中的内容
        final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);
        //判断是否禁用或者降级状态
        boolean localDisable = disable || (ATOMIC_DEGRADE_CHECK.get() && degradeNum >= degradeCheckAllowTimes);
        if (!localDisable) {
            if (globalTransactionalAnnotation != null || this.aspectTransactional != null) {
                AspectTransactional transactional;
                if (globalTransactionalAnnotation != null) {
                    //构建事务描述信息，这些基础配置信息很重要
                    transactional = new AspectTransactional(globalTransactionalAnnotation.timeoutMills(),
                        globalTransactionalAnnotation.name(), globalTransactionalAnnotation.rollbackFor(),
                        globalTransactionalAnnotation.rollbackForClassName(),
                        globalTransactionalAnnotation.noRollbackFor(),
                        globalTransactionalAnnotation.noRollbackForClassName(),
                        globalTransactionalAnnotation.propagation(),
                        globalTransactionalAnnotation.lockRetryInterval(),
                        globalTransactionalAnnotation.lockRetryTimes(),
                        globalTransactionalAnnotation.lockStrategyMode());
                } else {
                    transactional = this.aspectTransactional;
                }
                //若是@GlobalTransactional
                return handleGlobalTransaction(methodInvocation, transactional);
            } else if (globalLockAnnotation != null) {
                //若是@GlobalLock
                return handleGlobalLock(methodInvocation, globalLockAnnotation);
            }
        }
    }
    return methodInvocation.proceed();
}

Object handleGlobalTransaction(final MethodInvocation methodInvocation,
    final AspectTransactional aspectTransactional) throws Throwable {
    boolean succeed = true;
    try {
        return transactionalTemplate.execute(new TransactionalExecutor() {
            @Override
            public Object execute() throws Throwable {
                return methodInvocation.proceed();
            }
            ...
            @Override
            public TransactionInfo getTransactionInfo() {
                // reset the value of timeout
                int timeout = aspectTransactional.getTimeoutMills();
                if (timeout <= 0 || timeout == DEFAULT_GLOBAL_TRANSACTION_TIMEOUT) {
                    timeout = defaultGlobalTransactionTimeout;
                }
                TransactionInfo transactionInfo = new TransactionInfo();
                transactionInfo.setTimeOut(timeout);
                transactionInfo.setName(name());
                ...
                return transactionInfo;
            }
        });
    } catch (TransactionalExecutor.ExecutionException e) {
        ...
        }
    } finally {
        if (ATOMIC_DEGRADE_CHECK.get()) {
            EVENT_BUS.post(new DegradeCheckEvent(succeed));
        }
    }
}

public Object execute(TransactionalExecutor business) throws Throwable {
    // 1. Get transactionInfo
    TransactionInfo txInfo = business.getTransactionInfo();
    if (txInfo == null) {
        throw new ShouldNeverHappenException("transactionInfo does not exist");
    }
    // 1.1 Get current transaction, if not null, the tx role is 'GlobalTransactionRole.Participant'.
    GlobalTransaction tx = GlobalTransactionContext.getCurrent();
    // 1.2 Handle the transaction propagation.
    Propagation propagation = txInfo.getPropagation();
    SuspendedResourcesHolder suspendedResourcesHolder = null;
    try {
        //...
        // 1.3 If null, create new transaction with role 'GlobalTransactionRole.Launcher'.
        //若全局事务上下文未就绪则new DefaultGlobalTransaction();
        if (tx == null) {
            tx = GlobalTransactionContext.createNew();
        }
        // set current tx config to holder
        GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);
        try {
            // 2. If the tx role is 'GlobalTransactionRole.Launcher', send the request of beginTransaction to TC,
            //    else do nothing. Of course, the hooks will still be triggered.
            //2. 开启全局事务，
            // 2.1 triggerBeforeBegin()
                // 2.2 会跟TC通信获取全局事务ID:xid，
                // 2.3 RootContext.bind(xid);
            // 2.4 triggerAfterBegin()的事件通知调用
            beginTransaction(txInfo, tx);
            Object rs;
            try {
                // 执行我们的事务方法如`purchase`方法
                rs = business.execute();
            } catch (Throwable ex) {
                // 3. 遇到 business exception 则回滚
                completeTransactionAfterThrowing(txInfo, tx, ex);
                throw ex;
            }
            // 4. 提交事务，触发事件回调
            // 4.1 triggerBeforeCommit();
                // 4.2 tx.commit();与TC通信提交事务，内部默认是有5次重试机会
            // 4.3 triggerAfterCommit();
            commitTransaction(tx, txInfo);
            //返回结果
            return rs;
        } finally {
            //5. clear
            resumeGlobalLockConfig(previousConfig);
            //结束后的回调
            triggerAfterCompletion();
            cleanUp();
        }
    } finally {
        // If the transaction is suspended, resume it.
        if (suspendedResourcesHolder != null) {
            tx.resume(suspendedResourcesHolder);
        }
    }
}