Seata 源码篇之AT模式启动流程 - 上 - 02

    @Bean
    @DependsOn({BEAN_NAME_SPRING_APPLICATION_CONTEXT_PROVIDER, BEAN_NAME_FAILURE_HANDLER})
    @ConditionalOnMissingBean(GlobalTransactionScanner.class)
    public static GlobalTransactionScanner globalTransactionScanner(SeataProperties seataProperties, FailureHandler failureHandler,
            ConfigurableListableBeanFactory beanFactory,
            @Autowired(required = false) List<ScannerChecker> scannerCheckers) {
        ... 
        // 1.获取IOC引用 
        GlobalTransactionScanner.setBeanFactory(beanFactory);
        // 2.从各个来源添加bean过滤器: SPI,IOC,属性配置   
        GlobalTransactionScanner.addScannerCheckers(EnhancedServiceLoader.loadAll(ScannerChecker.class));
        GlobalTransactionScanner.addScannerCheckers(scannerCheckers);
        GlobalTransactionScanner.addScannablePackages(seataProperties.getScanPackages());
        GlobalTransactionScanner.addScannerExcludeBeanNames(seataProperties.getExcludesForScanning());
        ...
        // 3.完成Seata目标Bean扫描器的创建
        return new GlobalTransactionScanner(seataProperties.getApplicationId(), seataProperties.getTxServiceGroup(), failureHandler);
    }

    @Override
    protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
        // 1. 过滤掉Seata不感兴趣的Bean
        if (!doCheckers(bean, beanName)) {
            return bean;
        }

        try {
            synchronized (PROXYED_SET) {
                // 2. bean如果已经代理过了,则跳过代理
                if (PROXYED_SET.contains(beanName)) {
                    return bean;
                }
                interceptor = null;
                // 3. Seata提供的Advice,通过Bean所开启的模式不同,会返回AT模式对应的增强器或者TCC模式对应的增强器
                ProxyInvocationHandler proxyInvocationHandler = DefaultInterfaceParser.get().parserInterfaceToProxy(bean);
                // 4. 返回空,说明当前bean无需代理
                if (proxyInvocationHandler == null) {
                    return bean;
                }
                // 5. Advice最终都需要被适配为MethodInterceptor,要么第三方提供的就是MethodInterceptor类型的拦截器,否则需要额外向Spring中注册一个适配器,用于将自定义的Advice转化为MethodInterceptor
                interceptor = new AdapterSpringSeataInterceptor(proxyInvocationHandler);
                LOGGER.info("Bean [{}] with name [{}] would use interceptor [{}]", bean.getClass().getName(), beanName, interceptor.toString());
                // 6. 如果当前Bean还没有被代理过,那么直接调用父类现成逻辑进行代理
                if (!AopUtils.isAopProxy(bean)) {
                    bean = super.wrapIfNecessary(bean, beanName, cacheKey);
                } else {
                // 7. 如果当前Bean已经被代理过了,那就只需要在当前代理对象已有的拦截器链中再加入Seata提供的interceptor即可
                    // 7.1 AdvisedSupport是Spring中所有代理对象都会继承的接口,同时代理对象内部持有一个AdvisedSupport对象实例,该对象持有当前被代理对象的所有上下文信息
                    AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
                    // 7.2 构建Seata需要额外应用到当前代理对象上的增强器链,Seata重写了getAdvicesAndAdvisorsForBean方法,该方法重写后只会返回Seata提供的interceptor
                    Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
                    int pos;
                    // 7.3 将Seata提供的增强器插入到当前Bean已有增强器链指定位置
                    for (Advisor avr : advisor) {
                        pos = findAddSeataAdvisorPosition(advised, avr);
                        advised.addAdvisor(pos, avr);
                    }
                }
                PROXYED_SET.add(beanName);
                return bean;
            }
        } catch (Exception exx) {
            throw new RuntimeException(exx);
        }
    }

   private boolean doCheckers(Object bean, String beanName) {
        // 1. 已经代理过的Bean跳过代理 / 排除集合中包含的bean不进行代理 / 工厂Bean不进行代理
        if (PROXYED_SET.contains(beanName) || EXCLUDE_BEAN_NAME_SET.contains(beanName)
                || FactoryBean.class.isAssignableFrom(bean.getClass())) {
            return false;
        }
        // 2. 借助Bean扫描器过滤掉那些Seata不感兴趣的Bean
        if (!SCANNER_CHECKER_SET.isEmpty()) {
            for (ScannerChecker checker : SCANNER_CHECKER_SET) {
                try {
                    if (!checker.check(bean, beanName, beanFactory)) {
                        // failed check, do not scan this bean
                        return false;
                    }
                } catch (Exception e) {
                    LOGGER.error("Do check failed: beanName={}, checker={}",
                            beanName, checker.getClass().getSimpleName(), e);
                }
            }
        }

        return true;
    }

	protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
		// 一些Spring内部Bean无需代理 /  当前代理创建器会缓存其已经代理过的对象,避免重复代理
		. . .

		// 由子类实现,返回能应用到当前Bean上所有的增强器
		// 常用实现: 从IOC得到所有类型为Advisor的Bean,依次判断是否能应用到当前Bean上
		Object[] specificInterceptors = getAdvicesAndAdvisorsForBean(bean.getClass(), beanName, null);
		// 如果返回空,说明当前Bean无需代理,因为没有Advisor能应用到当前Bean上
		// 如果不为空,则对当前Bean进行代理
		if (specificInterceptors != DO_NOT_PROXY) {
			...
			// 为当前Bean创建代理对象
			Object proxy = createProxy(
					bean.getClass(), beanName, specificInterceptors, new SingletonTargetSource(bean));
			...
			// 返回代理对象
			return proxy;
		}
		...
		// 无需代理,返回原样Bean
		return bean;
	}

    @Override
    protected Object[] getAdvicesAndAdvisorsForBean(Class beanClass, String beanName, TargetSource customTargetSource)
            throws BeansException {
        return new Object[]{interceptor};
    }

    @Override
    public void afterPropertiesSet() {
        ...
        // 每个Client都需要去初始化 RM 和 TM
        if (initialized.compareAndSet(false, true)) {
            initClient();
        }
    }

    private void initClient() {
        ...
        //init TM
        TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
        ...
        //init RM
        RMClient.init(applicationId, txServiceGroup);
        ...
        registerSpringShutdownHook();
    }

    public static void init(String applicationId, String transactionServiceGroup, String accessKey, String secretKey) {
        // 1. 创建TM对应的Netty客户端
        TmNettyRemotingClient tmNettyRemotingClient = TmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup, accessKey, secretKey);
        // 2. 初始化Netty客户端相关资源
        tmNettyRemotingClient.init();
    }

    @Override
    public void init() {
        // 注册消息类型处理器,当从Netty接收到不同类型msg后,会分发给不同的消息处理器处理
        registerProcessor();
        if (initialized.compareAndSet(false, true)) {
            // 负责启动Netty客户端和相关线程池,用于定时发送心跳消息和处理消息
            super.init();
            // 尝试与TC建立连接
            if (io.seata.common.util.StringUtils.isNotBlank(transactionServiceGroup)) {
                getClientChannelManager().reconnect(transactionServiceGroup);
            }
        }
    }

    public static void init(String applicationId, String transactionServiceGroup) {
        // 1. 创建RM对应的Netty客户端
        RmNettyRemotingClient rmNettyRemotingClient = RmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup);
        rmNettyRemotingClient.setResourceManager(DefaultResourceManager.get());
        rmNettyRemotingClient.setTransactionMessageHandler(DefaultRMHandler.get());
        // 2. 初始化Netty客户端相关资源
        rmNettyRemotingClient.init();
    }

@SpringBootApplication(scanBasePackages = {"io.seata"})
public class ServerApplication {
    public static void main(String[] args) throws IOException {
        // run the spring-boot application
        SpringApplication.run(ServerApplication.class, args);
    }
}

    @Override
    public void run(String... args) {
          ...
          Server.start(args);
          ... 
    }

    public static void start(String[] args) {
        ...
        // 1. 创建Netty服务端
        NettyRemotingServer nettyRemotingServer = new NettyRemotingServer(workingThreads);
        XID.setPort(nettyRemotingServer.getListenPort());
        // 2. UUIDGenerator: 用于生成全局唯一的全局事务ID和分支事务ID,采用雪花算法实现 
        UUIDGenerator.init(parameterParser.getServerNode());
        // 3. SessionHodler负责事务日志（状态）的持久化存储
        SessionHolder.init();
        // 4. 锁管理器工厂初始化
        LockerManagerFactory.init();
        // 5. 事务协调器的核心,通过RpcServer与远程TM,RM通信来实现分支事务的提交,回滚等。
        DefaultCoordinator coordinator = DefaultCoordinator.getInstance(nettyRemotingServer);
        coordinator.init();
        nettyRemotingServer.setHandler(coordinator);
        ...
        // 6. 启动Netty服务端
        nettyRemotingServer.init();
    }

    @Override
    public Object invoke(@Nonnull MethodInvocation invocation) throws Throwable {
        // 简单封装了一层,便于获取当前代理对象的某些信息
        AdapterInvocationWrapper adapterInvocationWrapper = new AdapterInvocationWrapper(invocation);
        // 核心拦截逻辑还是汇聚在proxyInvocationHandler中
        Object result = proxyInvocationHandler.invoke(adapterInvocationWrapper);
        return result;
    }

    @Override
    protected Object doInvoke(InvocationWrapper invocation) throws Throwable {
        // 1. 获取目标对象类型
        Class<?> targetClass = invocation.getTarget().getClass();
        // 2. 获取当前拦截的目标方法调用
        Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
        if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
            // 3. 获取当前方法上标注的目标注解
            final GlobalTransactional globalTransactionalAnnotation = getAnnotation(specificMethod, targetClass, GlobalTransactional.class);
            final GlobalLock globalLockAnnotation = getAnnotation(specificMethod, targetClass, GlobalLock.class);
            boolean localDisable = disable || (ATOMIC_DEGRADE_CHECK.get() && degradeNum >= degradeCheckAllowTimes);
            if (!localDisable) {
                // 4. 优先处理GlobalTransactional注解情况
                if (globalTransactionalAnnotation != null || this.aspectTransactional != null) {
                    AspectTransactional transactional;
                    if (globalTransactionalAnnotation != null) {
                        // 4.1 填充全局事务上下文信息(超时时间,传播级别,重试次数,回滚异常)
                        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;
                    }
                    // 4.2 处理标注了GlobalTransactional注解的全局事务
                    return handleGlobalTransaction(invocation, transactional);
                } else if (globalLockAnnotation != null) {
                    // 5. 处理GlobalLock注解
                    return handleGlobalLock(invocation, globalLockAnnotation);
                }
            }
        }
        // 6. 直接执行目标方法,无需全局事务支持
        return invocation.proceed();
    }

    Object handleGlobalTransaction(final InvocationWrapper methodInvocation,
                                   final AspectTransactional aspectTransactional) throws Throwable {
        boolean succeed = true;
        try {
            // transactionalTemplate 使用了典型的模版方法加回调的设计模式
            // TransactionalExecutor 负责提供当前事务相关处理接口
            return transactionalTemplate.execute(new TransactionalExecutor() {
                // 执行当前分支事务的本地事务
                @Override
                public Object execute() throws Throwable {
                    return methodInvocation.proceed();
                }
                ...
                // 获取当前事务的上下文信息 -- 这里事务上下文信息由aspectTransactional 转换而来
                @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());
                    transactionInfo.setPropagation(aspectTransactional.getPropagation());
                    transactionInfo.setLockRetryInterval(aspectTransactional.getLockRetryInterval());
                    transactionInfo.setLockRetryTimes(aspectTransactional.getLockRetryTimes());
                    transactionInfo.setLockStrategyMode(aspectTransactional.getLockStrategyMode());
                    Set<RollbackRule> rollbackRules = new LinkedHashSet<>();
                    for (Class<?> rbRule : aspectTransactional.getRollbackFor()) {
                        rollbackRules.add(new RollbackRule(rbRule));
                    }
                    for (String rbRule : aspectTransactional.getRollbackForClassName()) {
                        rollbackRules.add(new RollbackRule(rbRule));
                    }
                    for (Class<?> rbRule : aspectTransactional.getNoRollbackFor()) {
                        rollbackRules.add(new NoRollbackRule(rbRule));
                    }
                    for (String rbRule : aspectTransactional.getNoRollbackForClassName()) {
                        rollbackRules.add(new NoRollbackRule(rbRule));
                    }
                    transactionInfo.setRollbackRules(rollbackRules);
                    return transactionInfo;
                }
            });
        } catch (TransactionalExecutor.ExecutionException e) {
            ....
        }
        ...
    }

    public Object execute(TransactionalExecutor business) throws Throwable {
        // 1. 获取当前全局事务的上下文信息
        TransactionInfo txInfo = business.getTransactionInfo();
        ...
        // 1.1 判断当前是否已经存在一个全局事务
        GlobalTransaction tx = GlobalTransactionContext.getCurrent();

        // 1.2 根据不同的全局事务传播行为进行处理
        Propagation propagation = txInfo.getPropagation();
        SuspendedResourcesHolder suspendedResourcesHolder = null;
        try {
            switch (propagation) {
                // 如果当前存在全局事务,则将当前全局事务挂起,然后直接无事务执行,执行完毕后,再恢复先前挂起的全局事务
                case NOT_SUPPORTED:
                    if (existingTransaction(tx)) {
                        suspendedResourcesHolder = tx.suspend(false);
                    }
                    return business.execute();
                // 如果当前存在全局事务,则将当前全局事务挂起,然后自己新创建一个全局事务执行,执行完毕后,再恢复先前挂起的全局事务
                case REQUIRES_NEW:
                    if (existingTransaction(tx)) {
                        suspendedResourcesHolder = tx.suspend(false);
                    }
                    tx = GlobalTransactionContext.createNew();
                    break;
                // 如果当前存在全局事务,则加入当前全局事务,否则无事务执行
                case SUPPORTS:
                    if (notExistingTransaction(tx)) {
                        return business.execute();
                    }
                    break;
                // 如果当前存在全局事务,则加入当前全局事务,否则新创建一个全局事务
                case REQUIRED:
                    tx = GlobalTransactionContext.getCurrentOrCreate();
                    break;
                // 如果当前存在全局事务,则抛出异常,否则无事务执行
                case NEVER:
                    if (existingTransaction(tx)) {
                        throw new TransactionException(
                                String.format("Existing transaction found for transaction marked with propagation 'never', xid = %s"
                                        , tx.getXid()));
                    } else {
                        return business.execute();
                    }
                // 如果当前不存在全局事务,则抛出异常,否则加入当前全局事务
                case MANDATORY:
                    if (notExistingTransaction(tx)) {
                        throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
                    }
                    break;
                default:
                    throw new TransactionException("Not Supported Propagation:" + propagation);
            }

            // 1.3 将当前全局锁配置设置到本地线程缓存中,然后返回先前的配置
            GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);

            if (tx.getGlobalTransactionRole() == GlobalTransactionRole.Participant) {
                LOGGER.info("join into a existing global transaction,xid={}", tx.getXid());
            }

            try {
                // 2. 如果当前线程是全局事务的发起者,即TM,则给TC发送一个开启全局事务的请求,否则只是简单回调相关钩子方法
                beginTransaction(txInfo, tx);

                Object rs;
                try {
                    // 3. 执行当前分支事务对应的本地事务
                    rs = business.execute();
                } catch (Throwable ex) {
                    // 4. 分支事务执行发生异常,判断对应异常是否需要回滚,如果需要则回滚当前全局事务
                    completeTransactionAfterThrowing(txInfo, tx, ex);
                    throw ex;
                }
                
                // 5. 当前分支事务执行正常,由TM发送提交全局事务的请求
                commitTransaction(tx, txInfo);

                return rs;
            } finally {
                // 6. 资源清理和恢复,同时触发钩子回调
                resumeGlobalLockConfig(previousConfig);
                triggerAfterCompletion();
                cleanUp();
            }
        } finally {
            // 7. 如果存在被挂起的全局事务,则进行恢复
            if (suspendedResourcesHolder != null) {
                tx.resume(suspendedResourcesHolder);
            }
        }
    }

    private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
        // 前后钩子方法调用
        triggerBeforeBegin();
        // 向TC发送开启全局事务的请求
        tx.begin(txInfo.getTimeOut(), txInfo.getName());
        triggerAfterBegin();
        ...
    }

    @Override
    public void begin(int timeout, String name) throws TransactionException {
        this.createTime = System.currentTimeMillis();
        // 如果当前分支事务不是首个发起全局事务的,即是RM,则啥也不干,直接返回
        if (role != GlobalTransactionRole.Launcher) {
            ...
            return;
        }
        // 如果当前分支事务是首个发起全局事务的,即是TM
        assertXIDNull();
        String currentXid = RootContext.getXID();
        if (currentXid != null) {
            throw new IllegalStateException("Global transaction already exists," +
                " can't begin a new global transaction, currentXid = " + currentXid);
        }
        // 向 TC 发送全局事务开启请求
        xid = transactionManager.begin(null, null, name, timeout);
        status = GlobalStatus.Begin;
        // 绑定XID到本地线程空间
        RootContext.bind(xid);
        ...
    }

    @Override
    public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)
        throws TransactionException {
        // 向 TC 发起全局事务开启请求
        GlobalBeginRequest request = new GlobalBeginRequest();
        request.setTransactionName(name);
        request.setTimeout(timeout);
        // 同步等待TC返回结果
        GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
        if (response.getResultCode() == ResultCode.Failed) {
            throw new TmTransactionException(TransactionExceptionCode.BeginFailed, response.getMsg());
        }
        // 从结果中取出本次全局事务ID
        return response.getXid();
    }

    private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
         return (AbstractTransactionResponse) TmNettyRemotingClient.getInstance().sendSyncRequest(request);
         ...
    }

    @Bean(BEAN_NAME_SEATA_AUTO_DATA_SOURCE_PROXY_CREATOR)
    @ConditionalOnMissingBean(SeataAutoDataSourceProxyCreator.class)
    public static SeataAutoDataSourceProxyCreator seataAutoDataSourceProxyCreator(SeataProperties seataProperties) {
        return new SeataAutoDataSourceProxyCreator(seataProperties.isUseJdkProxy(),
            seataProperties.getExcludesForAutoProxying(), seataProperties.getDataSourceProxyMode());
    }

public interface DynamicIntroductionAdvice extends Advice {
    // 当代理对象方法调用被拦截时,通过调用增强器该方法判断方法所属接口是否是增强器提供的
    // 如果是,则将方法调用请求转发给增强器内部持有的实例对象
	boolean implementsInterface(Class<?> intf);
}

public interface IntroductionInfo {
	Class<?>[] getInterfaces();
}

	@Override
	public Object invoke(MethodInvocation mi) throws Throwable {
	    // 1. 当前被调用的目标方法是否是Introduction增强器提供接口中包含的
		if (isMethodOnIntroducedInterface(mi)) {
		    // 1.1 调用接口实现对象的指定方法,获取返回值
		    Object retVal = AopUtils.invokeJoinpointUsingReflection(this.delegate, mi.getMethod(), mi.getArguments());
		    // 1.2 感觉类似一个trick,如果实现返回的是delegate对象,那么这里会替换为返回目标对象
			if (retVal == this.delegate && mi instanceof ProxyMethodInvocation) {
				Object proxy = ((ProxyMethodInvocation) mi).getProxy();
				// 1.3 再确保当前方法返回值兼容目标对象类型时,该trick才会生效
				if (mi.getMethod().getReturnType().isInstance(proxy)) {
					retVal = proxy;
				}
			}
			return retVal;
		}
        // 2. 当前被调用的方法不是Introduction增强器提供接口中包含的,走正常流程
		return doProceed(mi);
	}

public class TestMain {
    public static void main(String[] args) {
        // 1. 准备被代理的目标对象
        People peo = new People() {
            @Override
            public void eat() {
                System.out.println("吃饭");
            }

            @Override
            public void drink() {
                System.out.println("喝水");
            }
        };
        // 2. 准备代理工厂
        ProxyFactory pf = new ProxyFactory();
        // 3. 准备introduction增强器,增强器持有需要额外添加的接口Developer和Developer接口的实现类
        DelegatingIntroductionInterceptor dii=new DelegatingIntroductionInterceptor((Developer) () -> System.out.println("编码"));
        // 4. 添加增强器和代理对象需要继承的接口
        pf.addAdvice(dii);
        pf.addInterface(People.class);
        // 5. 设置被代理对象
        pf.setTarget(peo);
        // 6. 代理对象继承了People接口,Developer接口和Spring AOP内部接口
        peo = (People)pf.getProxy();
        peo.drink();
        peo.eat();
        // 7. 强制转换为Developer接口,实际方法调用会被introduction增强器拦截,调用请求转发给了增强器内部持有的Developer接口实现类
        Developer developer=(Developer) peo;
        developer.code();
    }

    public interface People {
        void eat();
        void drink();
    }

    public interface Developer{
        void code();
    }
}

	@Override
	public Object invoke(MethodInvocation mi) throws Throwable {
	    // 1. 当前被调用的目标方法是否是Introduction增强器提供接口中包含的
		if (isMethodOnIntroducedInterface(mi)) {
		    // 1.1 获取增强器内部持有的接口实现对象 
			Object delegate = getIntroductionDelegateFor(mi.getThis());
			// 1.2 调用实现对象对应的方法,并获取其返回值
			Object retVal = AopUtils.invokeJoinpointUsingReflection(delegate, mi.getMethod(), mi.getArguments());
			// 1.3 感觉类似一个trick,如果实现返回的是delegate对象,那么这里会替换为返回目标对象
			if (retVal == delegate && mi instanceof ProxyMethodInvocation) {
				retVal = ((ProxyMethodInvocation) mi).getProxy();
			}
			return retVal;
		}
        // 2. 当前被调用的方法不是Introduction增强器提供接口中包含的,走正常流程
		return doProceed(mi);
	}

	private Object getIntroductionDelegateFor(@Nullable Object targetObject) {
		synchronized (this.delegateMap) {
			if (this.delegateMap.containsKey(targetObject)) {
				return this.delegateMap.get(targetObject);
			}
			else {
				Object delegate = createNewDelegate();
				this.delegateMap.put(targetObject, delegate);
				return delegate;
			}
		}
	}

public class TestMain {
    public static void main(String[] args) {
        // 1. 准备被代理的目标对象
        People peo = new People() {
            @Override
            public void eat() {
                System.out.println("吃饭");
            }

            @Override
            public void drink() {
                System.out.println("喝水");
            }
        };
        // 2. 准备代理工厂
        ProxyFactory pf = new ProxyFactory();
        // 3. 准备introduction增强器
        DelegatePerTargetObjectIntroductionInterceptor dii=new DelegatePerTargetObjectIntroductionInterceptor(DeveloperImpl.class,Developer.class);
        // 4. 添加增强器和代理对象需要继承的接口
        pf.addAdvice(dii);
        pf.addInterface(People.class);
        // 5. 设置被代理对象
        pf.setTarget(peo);
        // 6. 代理对象继承了People接口,Developer接口和Spring AOP内部接口
        peo = (People)pf.getProxy();
        peo.drink();
        peo.eat();
        // 7. 强制转换为Developer接口,实际方法调用会被introduction增强器拦截,调用请求转发给了增强器内部持有的Developer接口实现类
        Developer developer=(Developer) peo;
        developer.code();
    }

    public interface People {
        void eat();
        void drink();
    }

    public interface Developer{
        void code();
    }

    public static class DeveloperImpl implements Developer{
        @Override
        public void code() {
            System.out.println("编码");
        }
    }
}

public class SeataAutoDataSourceProxyCreator extends AbstractAutoProxyCreator {
    ... 
    private final Object[] advisors;

    public SeataAutoDataSourceProxyCreator(boolean useJdkProxy, String[] excludes, String dataSourceProxyMode) {
        ...
        this.advisors = buildAdvisors(dataSourceProxyMode);
    }

    private Object[] buildAdvisors(String dataSourceProxyMode) {
        Advice advice = new SeataAutoDataSourceProxyAdvice(dataSourceProxyMode);
        return new Object[]{new DefaultIntroductionAdvisor(advice)};
    }

    @Override
    protected Object[] getAdvicesAndAdvisorsForBean(Class<?> beanClass, String beanName, TargetSource customTargetSource) {
        return advisors;
    }
    ...
 }   

    @Override
    protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
        // 1. 只针对类型为DataSource的Bean进行代理
        if (!(bean instanceof DataSource)) {
            return bean;
        }

        // 2. 不对类型为SeataDataSourceProxy的数据源再次进行代理
        if (!(bean instanceof SeataDataSourceProxy)) {
            // 3. 调用抽象自动代理创建器的方法对当前数据源进行代理
            Object enhancer = super.wrapIfNecessary(bean, beanName, cacheKey);
            // 4. 如果当前Bean无需代理,直接返回原本的Bean
            if (bean == enhancer) {
                return bean;
            }
            // 5. 当前数据源对应的代理对象已经添加了SeataAutoDataSourceProxyAdvice
            // 但是真正负责对数据源进行代理的是SeataDataSourceProxy,这一点在SeataAutoDataSourceProxyAdvice的invoke方法中会体现出来
            DataSource origin = (DataSource) bean;
            // 建立好原始数据源和其代理数据源的映射关系
            SeataDataSourceProxy proxy = buildProxy(origin, dataSourceProxyMode);
            DataSourceProxyHolder.put(origin, proxy);
            return enhancer;
        }
        // 3. 如果手动向容器中注入了数据源代理,依旧取出其中原始数据源,进行动态代理,然后返回
        // 这里是为了将自动和手动设置代理数据源的逻辑统一起来,不然可能会产生重复代理的问题
        SeataDataSourceProxy proxy = (SeataDataSourceProxy) bean;
        DataSource origin = proxy.getTargetDataSource();
        Object originEnhancer = super.wrapIfNecessary(origin, beanName, cacheKey);
        if (origin == originEnhancer) {
            return origin;
        }
        DataSourceProxyHolder.put(origin, proxy);
        return bean
    }
    
    // 根据不同的代理模式创建不同的数据源代理对象
    SeataDataSourceProxy buildProxy(DataSource origin, String proxyMode) {
        if (BranchType.AT.name().equalsIgnoreCase(proxyMode)) {
            return new DataSourceProxy(origin);
        }
        if (BranchType.XA.name().equalsIgnoreCase(proxyMode)) {
            return new DataSourceProxyXA(origin);
        }
        throw new IllegalArgumentException("Unknown dataSourceProxyMode: " + proxyMode);
    }

   @Override
    public Object invoke(MethodInvocation invocation) throws Throwable {
        // 1. 检查当前是否存在全局事务,或者是否需要获取全局锁
        if (!inExpectedContext()) {
            return invocation.proceed();
        }

        // 2. 获取当前调用的是数据源的哪个方法
        Method method = invocation.getMethod();
        String name = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();

        Method declared;
        try {
            declared = DataSource.class.getDeclaredMethod(name, parameterTypes);
        } catch (NoSuchMethodException e) {
            return invocation.proceed();
        }

        // 3. 取出当前数据源对应的SeataDataSourceProxy,然后调用代理数据源对应的方法
        DataSource origin = (DataSource) invocation.getThis();
        SeataDataSourceProxy proxy = DataSourceProxyHolder.get(origin);
        Object[] args = invocation.getArguments();
        return declared.invoke(proxy, args);
    }