Fescar源码学习--服务协调器TC

妖狐艹你老母 2022-04-23 05:40 143阅读 0赞

之前我们已经用两篇博客分别介绍了Fescar中的TM和RM两个角色的相关操作，这篇博客我们来介绍一下TC

[《Fescar源码学习--事物管理者TM（服务调用方）》][Fescar_--_TM]

[《Fescar源码学习--资源管理者RM（服务提供方）》][Fescar_--_RM]

### 一、简介 ###

TC（Fescar Server）作为全局事务协调器主要做了以下操作。

（1）TM和RM启动时会注册到TC，TC会将注册信息持久化。

（2）TM在begin事务会首先向TC申请获取全局事务xid。

（3）RM在执行数据库事务之前首先向TC申请获取分片事务branchId。

（4）TM进行事务commit或rollback时会将根据全局事务xid提交到TC，TC根据全局事务xid分别通知分片事务RM，TC 调度 XID 下管辖的全部分支事务完成提交或回滚请求。

###  二、源码分析 ###

**1、接收请求**

Fescar提供 了RpcServer用来接收所有的请求并进行处理。

（1）服务注册请求由ServerMessageListener的onRegRmMessage方法处理

（2）与事务管理相关的请求由ServerMessageListener.onTrxMessage方法处理

@Override
        public void dispatch(long msgId, ChannelHandlerContext ctx, Object msg) {
            if (msg instanceof RegisterRMRequest) {
                //服务注册请求
                serverMessageListener.onRegRmMessage(msgId, ctx, (RegisterRMRequest)msg, this,
                    checkAuthHandler);
            } else {
                if (ChannelManager.isRegistered(ctx.channel())) {
                    //事务管理请求
                    serverMessageListener.onTrxMessage(msgId, ctx, msg, this);
                } else {
                    try {
                        closeChannelHandlerContext(ctx);
                    } catch (Exception exx) {
                        LOGGER.error(exx.getMessage());
                    }
                    if (LOGGER.isInfoEnabled()) {
                        LOGGER.info(String.format("close a unhandled connection! [%s]", ctx.channel().toString()));
                    }
                }
            }
        }

**2、服务注册处理**

服务注册由ServerMessageListener的实现类DefaultServerMessageListenerImpl的onRegRmMessage或registerTMChannel方法处理，最终RM和TM信息都添加到Map中。

RM：dbkey+appname+ip port context

/**
         * dbkey+appname+ip port context
         */
        private static final ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer,
            RpcContext>>>>
            RM_CHANNELS
            = new ConcurrentHashMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer,
            RpcContext>>>>();

TM：ip+appname,port

/**
         * ip+appname,port
         */
        private static final ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> TM_CHANNELS
            = new ConcurrentHashMap<String, ConcurrentMap<Integer, RpcContext>>();

服务注册还是比较简单的，就是将Channel信息添加到Map中

/**
         * Register tm channel.
         *
         * @param request the request
         * @param channel the channel
         * @throws IncompatibleVersionException the incompatible version exception
         */
        public static void registerTMChannel(RegisterTMRequest request, Channel channel)
            throws IncompatibleVersionException {
            Version.checkVersion(request.getVersion());
            RpcContext rpcContext = buildChannelHolder(TransactionRole.TMROLE, request.getVersion(),
                request.getApplicationId(),
                request.getTransactionServiceGroup(),
                null, channel);
            rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
            String clientIdentified = rpcContext.getApplicationId() + Constants.CLIENT_ID_SPLIT_CHAR
                + getClientIpFromChannel(channel);
            TM_CHANNELS.putIfAbsent(clientIdentified, new ConcurrentHashMap<Integer, RpcContext>());
            ConcurrentMap<Integer, RpcContext> clientIdentifiedMap = TM_CHANNELS.get(clientIdentified);
            rpcContext.holdInClientChannels(clientIdentifiedMap);
        }
    
        /**
         * Register rm channel.
         *
         * @param resourceManagerRequest the resource manager request
         * @param channel                the channel
         * @throws IncompatibleVersionException the incompatible  version exception
         */
        public static void registerRMChannel(RegisterRMRequest resourceManagerRequest, Channel channel)
            throws IncompatibleVersionException {
            Version.checkVersion(resourceManagerRequest.getVersion());
            Set<String> dbkeySet = dbKeytoSet(resourceManagerRequest.getResourceIds());
            RpcContext rpcContext;
            if (!IDENTIFIED_CHANNELS.containsKey(channel)) {
                rpcContext = buildChannelHolder(TransactionRole.RMROLE, resourceManagerRequest.getVersion(),
                    resourceManagerRequest.getApplicationId(), resourceManagerRequest.getTransactionServiceGroup(),
                    resourceManagerRequest.getResourceIds(), channel);
                rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
            } else {
                rpcContext = IDENTIFIED_CHANNELS.get(channel);
                rpcContext.addResources(dbkeySet);
            }
            if (null == dbkeySet || dbkeySet.isEmpty()) { return; }
            for (String resourceId : dbkeySet) {
                RM_CHANNELS.putIfAbsent(resourceId,
                    new ConcurrentHashMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>>());
                ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> applicationIdMap
                    = RM_CHANNELS.get(resourceId);
                applicationIdMap.putIfAbsent(resourceManagerRequest.getApplicationId(),
                    new ConcurrentHashMap<String, ConcurrentMap<Integer, RpcContext>>());
                ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> clientIpMap = applicationIdMap.get(
                    resourceManagerRequest.getApplicationId());
                String clientIp = getClientIpFromChannel(channel);
                clientIpMap.putIfAbsent(clientIp, new ConcurrentHashMap<Integer, RpcContext>());
                ConcurrentMap<Integer, RpcContext> portMap = clientIpMap.get(clientIp);
                rpcContext.holdInResourceManagerChannels(resourceId, portMap);
                updateChannelsResource(resourceId, clientIp, resourceManagerRequest.getApplicationId());
            }
    
        }

**3、begin操作**

TM在执行dubbo远程调用之前会调用begin方法向TC申请一个全局事务xid，TC由GlobalBeginRequest的handle方法进行处理

@Override
        public AbstractTransactionResponse handle(RpcContext rpcContext) {
            return handler.handle(this, rpcContext);
        }

在AbstractTCInboundHandler中调用handle处理操作。

@Override
        public GlobalBeginResponse handle(GlobalBeginRequest request, final RpcContext rpcContext) {
            GlobalBeginResponse response = new GlobalBeginResponse();
            exceptionHandleTemplate(new Callback<GlobalBeginRequest, GlobalBeginResponse>() {
                @Override
                public void execute(GlobalBeginRequest request, GlobalBeginResponse response) throws TransactionException {
                    doGlobalBegin(request, response, rpcContext);
                }
            }, request, response);
            return response;
        }

在DefaultCoordinator中执行doGlobalBegin操作，最终调用DefaultCore的begin方法

@Override
        protected void doGlobalBegin(GlobalBeginRequest request, GlobalBeginResponse response, RpcContext rpcContext) throws TransactionException {
            response.setXid(core.begin(rpcContext.getApplicationId(), rpcContext.getTransactionServiceGroup(), request.getTransactionName(), request.getTimeout()));
        }

在DefaultCore的begin方法中完成全局事务xid的生成

@Override
        public String begin(String applicationId, String transactionServiceGroup, String name, int timeout) throws TransactionException {
            GlobalSession session = GlobalSession.createGlobalSession(
                    applicationId, transactionServiceGroup, name, timeout);
            session.addSessionLifecycleListener(SessionHolder.getRootSessionManager());
            //全局xid会进行持久化操作
            session.begin();
    
            return XID.generateXID(session.getTransactionId());
        }

这样就生成全局事务xid了。

**4、commit操作**

TM在进行commit操作后最终也是有TC的DefaultCore的commit方法进行操作，根据xid找到GlobalSession，然后依次调用分片事务进行commit操作。

@Override
        public GlobalStatus commit(String xid) throws TransactionException {
            GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid));
            if (globalSession == null) {
                return GlobalStatus.Finished;
            }
            GlobalStatus status = globalSession.getStatus();
    
            globalSession.closeAndClean(); // Highlight: Firstly, close the session, then no more branch can be registered.
    
            if (status == GlobalStatus.Begin) {
                if (globalSession.canBeCommittedAsync()) {
                    asyncCommit(globalSession);
                } else {
                    doGlobalCommit(globalSession, false);
                }
    
            }
            return globalSession.getStatus();
        }
    
        @Override
        public void doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException {
            //查找所有的分片事务，依次进行commit操作
            for (BranchSession branchSession : globalSession.getSortedBranches()) {
                BranchStatus currentStatus = branchSession.getStatus();
                if (currentStatus == BranchStatus.PhaseOne_Failed) {
                    continue;
                }
                try {
                    BranchStatus branchStatus = resourceManagerInbound.branchCommit(XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(),
                            branchSession.getResourceId(), branchSession.getApplicationData());
    
                    switch (branchStatus) {
                        case PhaseTwo_Committed:
                            globalSession.removeBranch(branchSession);
                            continue;
                        case PhaseTwo_CommitFailed_Unretriable:
                            if (globalSession.canBeCommittedAsync()) {
                                LOGGER.error("By [" + branchStatus + "], failed to commit branch " + branchSession);
                                continue;
                            } else {
                                globalSession.changeStatus(GlobalStatus.CommitFailed);
                                globalSession.end();
                                LOGGER.error("Finally, failed to commit global[" + globalSession.getTransactionId() + "] since branch[" + branchSession.getBranchId() + "] commit failed");
                                return;
                            }
                        default:
                            if (!retrying) {
                                queueToRetryCommit(globalSession);
                                return;
                            }
                            if (globalSession.canBeCommittedAsync()) {
                                LOGGER.error("By [" + branchStatus + "], failed to commit branch " + branchSession);
                                continue;
                            } else {
                                LOGGER.error("Failed to commit global[" + globalSession.getTransactionId() + "] since branch[" + branchSession.getBranchId() + "] commit failed, will retry later.");
                                return;
                            }
    
                    }
    
                } catch (Exception ex) {
                    LOGGER.info("Exception committing branch " + branchSession, ex);
                    if (!retrying) {
                        queueToRetryCommit(globalSession);
                        if (ex instanceof TransactionException) {
                            throw (TransactionException) ex;
                        } else {
                            throw new TransactionException(ex);
                        }
                    }
    
                }
    
            }
            if (globalSession.hasBranch()) {
                LOGGER.info("Global[" + globalSession.getTransactionId() + "] committing is NOT done.");
                return;
            }
            globalSession.changeStatus(GlobalStatus.Committed);
            globalSession.end();
            LOGGER.info("Global[" + globalSession.getTransactionId() + "] committing is successfully done.");
        }

**5、rollback操作**

rollback与commit的操作几乎类似，也是依次调用分片事务的rollback操作

@Override
        public GlobalStatus rollback(String xid) throws TransactionException {
            GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid));
            if (globalSession == null) {
                return GlobalStatus.Finished;
            }
            GlobalStatus status = globalSession.getStatus();
    
            globalSession.close(); // Highlight: Firstly, close the session, then no more branch can be registered.
    
            if (status == GlobalStatus.Begin) {
                globalSession.changeStatus(GlobalStatus.Rollbacking);
                doGlobalRollback(globalSession, false);
    
            }
            return globalSession.getStatus();
        }
    
        @Override
        public void doGlobalRollback(GlobalSession globalSession, boolean retrying) throws TransactionException {
            for (BranchSession branchSession : globalSession.getReverseSortedBranches()) {
                BranchStatus currentBranchStatus = branchSession.getStatus();
                if (currentBranchStatus == BranchStatus.PhaseOne_Failed) {
                    continue;
                }
                try {
                    BranchStatus branchStatus = resourceManagerInbound.branchRollback(XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(),
                            branchSession.getResourceId(), branchSession.getApplicationData());
    
                    switch (branchStatus) {
                        case PhaseTwo_Rollbacked:
                            globalSession.removeBranch(branchSession);
                            LOGGER.error("Successfully rolled back branch " + branchSession);
                            continue;
                        case PhaseTwo_RollbackFailed_Unretriable:
                            GlobalStatus currentStatus = globalSession.getStatus();
                            if (currentStatus.name().startsWith("Timeout")) {
                                globalSession.changeStatus(GlobalStatus.TimeoutRollbackFailed);
                            } else {
                                globalSession.changeStatus(GlobalStatus.RollbackFailed);
                            }
                            globalSession.end();
                            LOGGER.error("Failed to rollback global[" + globalSession.getTransactionId() + "] since branch[" + branchSession.getBranchId() + "] rollback failed");
                            return;
                        default:
                            LOGGER.info("Failed to rollback branch " + branchSession);
                            if (!retrying) {
                                queueToRetryRollback(globalSession);
                            }
                            return;
    
                    }
                } catch (Exception ex) {
                    LOGGER.info("Exception rollbacking branch " + branchSession, ex);
                    if (!retrying) {
                        queueToRetryRollback(globalSession);
                        if (ex instanceof TransactionException) {
                            throw (TransactionException) ex;
                        } else {
                            throw new TransactionException(ex);
                        }
                    }
    
                }
    
            }
            GlobalStatus currentStatus = globalSession.getStatus();
            if (currentStatus.name().startsWith("Timeout")) {
                globalSession.changeStatus(GlobalStatus.TimeoutRollbacked);
            } else {
                globalSession.changeStatus(GlobalStatus.Rollbacked);
            }
            globalSession.end();
        }

总结：

（1）TC作为一个注册中心，保存了TM和RM服务信息。

（2）TC提供全局事务xid和分片事务branchId的创建操作。

（3）TC转发TM的commit或rollback相关操作。

[Fescar_--_TM]: https://blog.csdn.net/qq924862077/article/details/86623537
[Fescar_--_RM]: https://blog.csdn.net/qq924862077/article/details/86623666