5、分布式事务Seata源码学习

public class GlobalTransactionScanner extends AbstractAutoProxyCreator
        implements CachedConfigurationChangeListener, InitializingBean, ApplicationContextAware, DisposableBean {

@Override
public void afterPropertiesSet() {
    if (disableGlobalTransaction) {
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Global transaction is disabled.");
        }
        ConfigurationFactory.getInstance().addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (CachedConfigurationChangeListener) this);
        return;
    }
    if (initialized.compareAndSet(false, true)) {
        initClient();   //初始化客户端，什么客户端呢？
    }

    this.findBusinessBeanNamesNeededEnhancement();
}

protected void initClient() {
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Initializing Global Transaction Clients ... ");
    }
    if (DEFAULT_TX_GROUP_OLD.equals(txServiceGroup)) {
        LOGGER.warn("the default value of seata.tx-service-group: {} has already changed to {} since Seata 1.5, " +
                    "please change your default configuration as soon as possible " +
                    "and we don't recommend you to use default tx-service-group's value provided by seata",
                    DEFAULT_TX_GROUP_OLD, DEFAULT_TX_GROUP);
    }
    if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
        throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
    }
    //init TM
    TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
    }
    //init RM
    RMClient.init(applicationId, txServiceGroup);
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
    }

    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Global Transaction Clients are initialized. ");
    }
    registerSpringShutdownHook();

}

public class RMClient {

    /**
     * Init.
     *
     * @param applicationId           the application id
     * @param transactionServiceGroup the transaction service group
     */
    public static void init(String applicationId, String transactionServiceGroup) {
        RmNettyRemotingClient rmNettyRemotingClient = RmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup);
        rmNettyRemotingClient.setResourceManager(DefaultResourceManager.get());
        rmNettyRemotingClient.setTransactionMessageHandler(DefaultRMHandler.get());
        rmNettyRemotingClient.init();
    }
}

@Override
public void init() {
    // registry processor
    registerProcessor();
    if (initialized.compareAndSet(false, true)) {
        super.init();

        // Found one or more resources that were registered before initialization
        if (resourceManager != null
            && !resourceManager.getManagedResources().isEmpty()
            && StringUtils.isNotBlank(transactionServiceGroup)) {
            boolean failFast = ConfigurationFactory.getInstance().getBoolean(
                ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST,
                DefaultValues.DEFAULT_CLIENT_CHANNEL_CHECK_FAIL_FAST);
            getClientChannelManager().initReconnect(transactionServiceGroup, failFast);
        }
    }
}

@Override
public void init() {
    timerExecutor.scheduleAtFixedRate(() -> {
        try {
            clientChannelManager.reconnect(getTransactionServiceGroup());
        } catch (Exception ex) {
            LOGGER.warn("reconnect server failed. {}", ex.getMessage());
        }
    }, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
    if (this.isEnableClientBatchSendRequest()) {
        mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
                                                          MAX_MERGE_SEND_THREAD,
                                                          KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
                                                          new LinkedBlockingQueue<>(),
                                                          new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
        mergeSendExecutorService.submit(new MergedSendRunnable());
    }
    super.init();
    clientBootstrap.start();
}

void doReconnect(String transactionServiceGroup, boolean failFast) {
    List<String> availList;
    try {
        availList = getAvailServerList(transactionServiceGroup);
    } catch (Exception e) {
        LOGGER.error("Failed to get available servers: {}", e.getMessage(), e);
        throwFailFastException(failFast, "Failed to get available servers");
        return;
    }
    if (CollectionUtils.isEmpty(availList)) {
        RegistryService registryService = RegistryFactory.getInstance();
        String clusterName = registryService.getServiceGroup(transactionServiceGroup);

        if (StringUtils.isBlank(clusterName)) {
            LOGGER.error("can not get cluster name in registry config '{}{}', please make sure registry config correct",
                         ConfigurationKeys.SERVICE_GROUP_MAPPING_PREFIX,
                         transactionServiceGroup);
            throwFailFastException(failFast, "can not get cluster name in registry config.");
            return;
        }

        if (!(registryService instanceof FileRegistryServiceImpl)) {
            LOGGER.error("no available service found in cluster '{}', please make sure registry config correct and keep your seata server running", clusterName);
        }
        throwFailFastException(failFast, "no available service found in cluster.");
        return;
    }
    try {
        doReconnect(availList, transactionServiceGroup);
    } catch (Exception e) {
        if (failFast) {
            throw e;
        }
        LOGGER.error("connect server failed. {}", e.getMessage(), e);
    }
}

void doReconnect(List<String> availList, String transactionServiceGroup) {
    Set<String> channelAddress = new HashSet<>(availList.size());
    Map<String, Exception> failedMap = new HashMap<>();
    try {
        for (String serverAddress : availList) {
            try {
                acquireChannel(serverAddress);
                channelAddress.add(serverAddress);
            } catch (Exception e) {
                failedMap.put(serverAddress, e);
            }
        }
        if (failedMap.size() > 0) {
            if (LOGGER.isInfoEnabled()) {
                LOGGER.error("{} can not connect to {} cause:{}", FrameworkErrorCode.NetConnect.getErrCode(),
                             failedMap.keySet(),
                             failedMap.values().stream().map(Throwable::getMessage).collect(Collectors.toSet()));
            } else if (LOGGER.isDebugEnabled()) {
                failedMap.forEach((key, value) -> {
                    LOGGER.error("{} can not connect to {} cause:{} trace information:",
                                 FrameworkErrorCode.NetConnect.getErrCode(), key, value.getMessage(), value);
                });
            }
        }
        if (availList.size() == failedMap.size()) {
            String invalidAddress = StringUtils.join(failedMap.keySet().iterator(), ", ");
            throw new FrameworkException("can not connect to [" + invalidAddress + "]");
        }
    } finally {
        if (CollectionUtils.isNotEmpty(channelAddress)) {
            List<InetSocketAddress> aliveAddress = new ArrayList<>(channelAddress.size());
            for (String address : channelAddress) {
                String[] array = NetUtil.splitIPPortStr(address);
                aliveAddress.add(new InetSocketAddress(array[0], Integer.parseInt(array[1])));
            }
            RegistryFactory.getInstance().refreshAliveLookup(transactionServiceGroup, aliveAddress);
        } else {
            RegistryFactory.getInstance().refreshAliveLookup(transactionServiceGroup, Collections.emptyList());
        }
    }
}

Channel acquireChannel(String serverAddress) {
    Channel channelToServer = channels.get(serverAddress);
    if (channelToServer != null) {
        channelToServer = getExistAliveChannel(channelToServer, serverAddress);
        if (channelToServer != null) {
            return channelToServer;
        }
    }
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("will connect to {}", serverAddress);
    }
    Object lockObj = CollectionUtils.computeIfAbsent(channelLocks, serverAddress, key -> new Object());
    synchronized (lockObj) {
        return doConnect(serverAddress);
    }
}

public class AspectTransactionalInterceptor implements MethodInterceptor {

@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
    Class<?> targetClass = invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null;
    Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
    InvocationWrapper invocationWrapper = new DefaultInvocationWrapper(null, invocation.getThis(), specificMethod, invocation.getArguments());
    return this.globalTransactionalInterceptorHandler.invoke(invocationWrapper);
}

@Override
public Object invoke(InvocationWrapper invocation) throws Throwable {
    if (CollectionUtils.isNotEmpty(getMethodsToProxy()) && !getMethodsToProxy().contains(invocation.getMethod().getName())) {
        return invocation.proceed();
    }
    if (nextInvocationHandlerChain != null) {
        invocation = new NestInterceptorHandlerWrapper(nextInvocationHandlerChain, ihttp://blog.shengxiao.tech/imagesnvocation);
    }
    return doInvoke(invocation);
}

@Override
protected Object doInvoke(InvocationWrapper invocation) throws Throwable {
    Class<?> targetClass = invocation.getTarget().getClass();
    Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
    if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
        boolean localDisable = disable || (ATOMIC_DEGRADE_CHECK.get() && degradeNum >= degradeCheckAllowTimes);
        if (!localDisable) {
            final AspectTransactional globalTransactionalAnnotation = getAspectTransactional(specificMethod, targetClass);
            final GlobalLockConfig globalLockAnnotation = getGlobalLockConfig(specificMethod, targetClass);
            if (globalTransactionalAnnotation != null || this.aspectTransactional != null) {
                AspectTransactional transactional;
                if (globalTransactionalAnnotation != null) {
                    transactional = globalTransactionalAnnotation;
                } else {
                    transactional = this.aspectTransactional;
                }
                return handleGlobalTransaction(invocation, transactional);
            } else if (globalLockAnnotation != null) {
                return handleGlobalLock(invocation, globalLockAnnotation);
            }
        }
    }
    return invocation.proceed();
}

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 {
        switch (propagation) {
            case NOT_SUPPORTED:
                // If transaction is existing, suspend it.
                if (existingTransaction(tx)) {
                    suspendedResourcesHolder = tx.suspend(false);
                }
                // Execute without transaction and return.
                return business.execute();
            case REQUIRES_NEW:
                // If transaction is existing, suspend it, and then begin new transaction.
                if (existingTransaction(tx)) {
                    suspendedResourcesHolder = tx.suspend(false);
                }
                tx = GlobalTransactionContext.createNew();
                // Continue and execute with new transaction
                break;
            case SUPPORTS:
                // If transaction is not existing, execute without transaction.
                if (notExistingTransaction(tx)) {
                    return business.execute();
                }
                // Continue and execute with new transaction
                break;
            case REQUIRED:
                // If current transaction is existing, execute with current transaction,else create
                tx = GlobalTransactionContext.getCurrentOrCreate();
                break;
            case NEVER:
                // If transaction is existing, throw exception.
                if (existingTransaction(tx)) {
                    throw new TransactionException(
                        String.format("Existing transaction found for transaction marked with propagation 'never', xid = %s"
                                      , tx.getXid()));
                } else {
                    // Execute without transaction and return.
                    return business.execute();
                }
            case MANDATORY:
                // If transaction is not existing, throw exception.
                if (notExistingTransaction(tx)) {
                    throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
                }
                // Continue and execute with current transaction.
                break;
            default:
                throw new TransactionException("Not Supported Propagation:" + propagation);
        }

        // set current tx config to holder
        GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);

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

        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.
            beginTransaction(txInfo, tx);

            Object rs;
            try {
                // Do Your Business
                rs = business.execute();
            } catch (Throwable ex) {
                // 3. The needed business exception to rollback.
                completeTransactionAfterThrowing(txInfo, tx, ex);
                throw ex;
            }

            // 4. everything is fine, commit.
            commitTransaction(tx, txInfo);

            return rs;
        } finally {
            //5. clear
            resumeGlobalLockConfig(previousConfig);
            triggerAfterCompletion(tx);
            cleanUp(tx);
        }
    } finally {
        // If the transaction is suspended, resume it.
        if (suspendedResourcesHolder != null) {
            tx.resume(suspendedResourcesHolder);
        }
    }
}

@Override
public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)
    throws TransactionException {
    GlobalBeginRequest request = new GlobalBeginRequest();
    request.setTransactionName(name);
    request.setTimeout(timeout);
    GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
    if (response.getResultCode() == ResultCode.Failed) {
        throw new TmTransactionException(TransactionExceptionCode.BeginFailed, response.getMsg());
    }
    return response.getXid();
}

private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
    try {
        return (AbstractTransactionResponse) TmNettyRemotingClient.getInstance().sendSyncRequest(request);
    } catch (TimeoutException toe) {
        throw new TmTransactionException(TransactionExceptionCode.IO, "RPC timeout", toe);
    }
}

@Override
public Object sendSyncRequest(Object msg) throws TimeoutException {
    String serverAddress = loadBalance(getTransactionServiceGroup(), msg);
    long timeoutMillis = this.getRpcRequestTimeout();
    RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);

    // send batch message
    // put message into basketMap, @see MergedSendRunnable
    if (this.isEnableClientBatchSendRequest()) {

        // send batch message is sync request, needs to create messageFuture and put it in futures.
        MessageFuture messageFuture = new MessageFuture();
        messageFuture.setRequestMessage(rpcMessage);
        messageFuture.setTimeout(timeoutMillis);
        futures.put(rpcMessage.getId(), messageFuture);

        // put message into basketMap
        BlockingQueue<RpcMessage> basket = CollectionUtils.computeIfAbsent(basketMap, serverAddress,
                                                                           key -> new LinkedBlockingQueue<>());
        if (!basket.offer(rpcMessage)) {
            LOGGER.error("put message into basketMap offer failed, serverAddress:{},rpcMessage:{}",
                         serverAddress, rpcMessage);
            return null;
        }
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("offer message: {}", rpcMessage.getBody());
        }
        if (!isSending) {
            synchronized (mergeLock) {
                mergeLock.notifyAll();
            }
        }

        try {
            Object response = messageFuture.get(timeoutMillis, TimeUnit.MILLISECONDS);
            return response;
        } catch (Exception exx) {
            LOGGER.error("wait response error:{},ip:{},request:{}", exx.getMessage(), serverAddress, rpcMessage.getBody());
            if (exx instanceof TimeoutException) {
                throw (TimeoutException)exx;
            } else {
                throw new RuntimeException(exx);
            }
        }
    } else {
        Channel channel = clientChannelManager.acquireChannel(serverAddress);
        return super.sendSync(channel, rpcMessage, timeoutMillis);
    }

}

@Override
public void init() {
    timerExecutor.scheduleAtFixedRate(() -> {
        try {
            clientChannelManager.reconnect(getTransactionServiceGroup());
        } catch (Exception ex) {
            LOGGER.warn("reconnect server failed. {}", ex.getMessage());
        }
    }, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
    if (this.isEnableClientBatchSendRequest()) {
        mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
                                                          MAX_MERGE_SEND_THREAD,
                                                          KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
                                                          new LinkedBlockingQueue<>(),
                                                          new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
        mergeSendExecutorService.submit(new MergedSendRunnable());
    }
    super.init();
    clientBootstrap.start();
}

@Override
public void run() {
    while (true) {
        synchronized (mergeLock) {
            try {
                mergeLock.wait(MAX_MERGE_SEND_MILLS);
            } catch (InterruptedException e) {
            }
        }
        isSending = true;
        basketMap.forEach((address, basket) -> {
            if (basket.isEmpty()) {
                return;
            }

            MergedWarpMessage mergeMessage = new MergedWarpMessage();
            while (!basket.isEmpty()) {
                RpcMessage msg = basket.poll();
                mergeMessage.msgs.add((AbstractMessage) msg.getBody());
                mergeMessage.msgIds.add(msg.getId());
            }
            if (mergeMessage.msgIds.size() > 1) {
                printMergeMessageLog(mergeMessage);
            }
            Channel sendChannel = null;
            try {
                // send batch message is sync request, but there is no need to get the return value.
                // Since the messageFuture has been created before the message is placed in basketMap,
                // the return value will be obtained in ClientOnResponseProcessor.
                sendChannel = clientChannelManager.acquireChannel(address);
                AbstractNettyRemotingClient.this.sendAsyncRequest(sendChannel, mergeMessage);
            } catch (FrameworkException e) {
                if (e.getErrcode() == FrameworkErrorCode.ChannelIsNotWritable && sendChannel != null) {
                    destroyChannel(address, sendChannel);
                }
                // fast fail
                for (Integer msgId : mergeMessage.msgIds) {
                    MessageFuture messageFuture = futures.remove(msgId);
                    Integer parentId = childToParentMap.remove(msgId);
                    if (parentId != null) {
                        mergeMsgMap.remove(parentId);
                    }
                    if (messageFuture != null) {
                        messageFuture.setResultMessage(
                            new RuntimeException(String.format("%s is unreachable", address), e));
                    }
                }
                LOGGER.error("client merge call failed: {}", e.getMessage(), e);
            }
        });
        isSending = false;
    }
}

@Override
public void sendAsyncRequest(Channel channel, Object msg) {
    if (channel == null) {
        LOGGER.warn("sendAsyncRequest nothing, caused by null channel.");
        return;
    }
    RpcMessage rpcMessage = buildRequestMessage(msg, msg instanceof HeartbeatMessage
                                                ? ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST
                                                : ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY);
    Object body = rpcMessage.getBody();
    if (body instanceof MergeMessage) {
        Integer parentId = rpcMessage.getId();
        mergeMsgMap.put(parentId, (MergeMessage)rpcMessage.getBody());
        if (body instanceof MergedWarpMessage) {
            for (Integer msgId : ((MergedWarpMessage)rpcMessage.getBody()).msgIds) {
                childToParentMap.put(msgId, parentId);
            }
        }
    }
    super.sendAsync(channel, rpcMessage);
}