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

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

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

《Fescar源码学习—事物管理者TM（服务调用方）》

《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方法处理

  1. @Override
  2.     public void dispatch(long msgId, ChannelHandlerContext ctx, Object msg) {
  3.         if (msg instanceof RegisterRMRequest) {
  4.             //服务注册请求
  5.             serverMessageListener.onRegRmMessage(msgId, ctx, (RegisterRMRequest)msg, this,
  6.                 checkAuthHandler);
  7.         } else {
  8.             if (ChannelManager.isRegistered(ctx.channel())) {
  9.                 //事务管理请求
  10.                 serverMessageListener.onTrxMessage(msgId, ctx, msg, this);
  11.             } else {
  12.                 try {
  13.                     closeChannelHandlerContext(ctx);
  14.                 } catch (Exception exx) {
  15.                     LOGGER.error(exx.getMessage());
  16.                 }
  17.                 if (LOGGER.isInfoEnabled()) {
  18.                     LOGGER.info(String.format("close a unhandled connection! [%s]", ctx.channel().toString()));
  19.                 }
  20.             }
  21.         }
  22.     }

2、服务注册处理

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

RM：dbkey+appname+ip port context

  1. /**
  2.      * dbkey+appname+ip port context
  3.      */
  4.     private static final ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer,
  5.         RpcContext>>>>
  6.         RM_CHANNELS
  7.         = new ConcurrentHashMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer,
  8.         RpcContext>>>>();

TM：ip+appname,port

  1. /**
  2.      * ip+appname,port
  3.      */
  4.     private static final ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> TM_CHANNELS
  5.         = new ConcurrentHashMap<String, ConcurrentMap<Integer, RpcContext>>();

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

  1. /**
  2.      * Register tm channel.
  3.      *
  4.      * @param request the request
  5.      * @param channel the channel
  6.      * @throws IncompatibleVersionException the incompatible version exception
  7.      */
  8.     public static void registerTMChannel(RegisterTMRequest request, Channel channel)
  9.         throws IncompatibleVersionException {
  10.         Version.checkVersion(request.getVersion());
  11.         RpcContext rpcContext = buildChannelHolder(TransactionRole.TMROLE, request.getVersion(),
  12.             request.getApplicationId(),
  13.             request.getTransactionServiceGroup(),
  14.             null, channel);
  15.         rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
  16.         String clientIdentified = rpcContext.getApplicationId() + Constants.CLIENT_ID_SPLIT_CHAR
  17.             + getClientIpFromChannel(channel);
  18.         TM_CHANNELS.putIfAbsent(clientIdentified, new ConcurrentHashMap<Integer, RpcContext>());
  19.         ConcurrentMap<Integer, RpcContext> clientIdentifiedMap = TM_CHANNELS.get(clientIdentified);
  20.         rpcContext.holdInClientChannels(clientIdentifiedMap);
  21.     }
  23.     /**
  24.      * Register rm channel.
  25.      *
  26.      * @param resourceManagerRequest the resource manager request
  27.      * @param channel                the channel
  28.      * @throws IncompatibleVersionException the incompatible  version exception
  29.      */
  30.     public static void registerRMChannel(RegisterRMRequest resourceManagerRequest, Channel channel)
  31.         throws IncompatibleVersionException {
  32.         Version.checkVersion(resourceManagerRequest.getVersion());
  33.         Set<String> dbkeySet = dbKeytoSet(resourceManagerRequest.getResourceIds());
  34.         RpcContext rpcContext;
  35.         if (!IDENTIFIED_CHANNELS.containsKey(channel)) {
  36.             rpcContext = buildChannelHolder(TransactionRole.RMROLE, resourceManagerRequest.getVersion(),
  37.                 resourceManagerRequest.getApplicationId(), resourceManagerRequest.getTransactionServiceGroup(),
  38.                 resourceManagerRequest.getResourceIds(), channel);
  39.             rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
  40.         } else {
  41.             rpcContext = IDENTIFIED_CHANNELS.get(channel);
  42.             rpcContext.addResources(dbkeySet);
  43.         }
  44.         if (null == dbkeySet || dbkeySet.isEmpty()) { return; }
  45.         for (String resourceId : dbkeySet) {
  46.             RM_CHANNELS.putIfAbsent(resourceId,
  47.                 new ConcurrentHashMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>>());
  48.             ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> applicationIdMap
  49.                 = RM_CHANNELS.get(resourceId);
  50.             applicationIdMap.putIfAbsent(resourceManagerRequest.getApplicationId(),
  51.                 new ConcurrentHashMap<String, ConcurrentMap<Integer, RpcContext>>());
  52.             ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> clientIpMap = applicationIdMap.get(
  53.                 resourceManagerRequest.getApplicationId());
  54.             String clientIp = getClientIpFromChannel(channel);
  55.             clientIpMap.putIfAbsent(clientIp, new ConcurrentHashMap<Integer, RpcContext>());
  56.             ConcurrentMap<Integer, RpcContext> portMap = clientIpMap.get(clientIp);
  57.             rpcContext.holdInResourceManagerChannels(resourceId, portMap);
  58.             updateChannelsResource(resourceId, clientIp, resourceManagerRequest.getApplicationId());
  59.         }
  61.     }

3、begin操作

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

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

在AbstractTCInboundHandler中调用handle处理操作。

  1. @Override
  2.     public GlobalBeginResponse handle(GlobalBeginRequest request, final RpcContext rpcContext) {
  3.         GlobalBeginResponse response = new GlobalBeginResponse();
  4.         exceptionHandleTemplate(new Callback<GlobalBeginRequest, GlobalBeginResponse>() {
  5.             @Override
  6.             public void execute(GlobalBeginRequest request, GlobalBeginResponse response) throws TransactionException {
  7.                 doGlobalBegin(request, response, rpcContext);
  8.             }
  9.         }, request, response);
  10.         return response;
  11.     }

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

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

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

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

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

4、commit操作

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

  1. @Override
  2.     public GlobalStatus commit(String xid) throws TransactionException {
  3.         GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid));
  4.         if (globalSession == null) {
  5.             return GlobalStatus.Finished;
  6.         }
  7.         GlobalStatus status = globalSession.getStatus();
  9.         globalSession.closeAndClean(); // Highlight: Firstly, close the session, then no more branch can be registered.
  11.         if (status == GlobalStatus.Begin) {
  12.             if (globalSession.canBeCommittedAsync()) {
  13.                 asyncCommit(globalSession);
  14.             } else {
  15.                 doGlobalCommit(globalSession, false);
  16.             }
  18.         }
  19.         return globalSession.getStatus();
  20.     }
  22.     @Override
  23.     public void doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException {
  24.         //查找所有的分片事务，依次进行commit操作
  25.         for (BranchSession branchSession : globalSession.getSortedBranches()) {
  26.             BranchStatus currentStatus = branchSession.getStatus();
  27.             if (currentStatus == BranchStatus.PhaseOne_Failed) {
  28.                 continue;
  29.             }
  30.             try {
  31.                 BranchStatus branchStatus = resourceManagerInbound.branchCommit(XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(),
  32.                         branchSession.getResourceId(), branchSession.getApplicationData());
  34.                 switch (branchStatus) {
  35.                     case PhaseTwo_Committed:
  36.                         globalSession.removeBranch(branchSession);
  37.                         continue;
  38.                     case PhaseTwo_CommitFailed_Unretriable:
  39.                         if (globalSession.canBeCommittedAsync()) {
  40.                             LOGGER.error("By [" + branchStatus + "], failed to commit branch " + branchSession);
  41.                             continue;
  42.                         } else {
  43.                             globalSession.changeStatus(GlobalStatus.CommitFailed);
  44.                             globalSession.end();
  45.                             LOGGER.error("Finally, failed to commit global[" + globalSession.getTransactionId() + "] since branch[" + branchSession.getBranchId() + "] commit failed");
  46.                             return;
  47.                         }
  48.                     default:
  49.                         if (!retrying) {
  50.                             queueToRetryCommit(globalSession);
  51.                             return;
  52.                         }
  53.                         if (globalSession.canBeCommittedAsync()) {
  54.                             LOGGER.error("By [" + branchStatus + "], failed to commit branch " + branchSession);
  55.                             continue;
  56.                         } else {
  57.                             LOGGER.error("Failed to commit global[" + globalSession.getTransactionId() + "] since branch[" + branchSession.getBranchId() + "] commit failed, will retry later.");
  58.                             return;
  59.                         }
  61.                 }
  63.             } catch (Exception ex) {
  64.                 LOGGER.info("Exception committing branch " + branchSession, ex);
  65.                 if (!retrying) {
  66.                     queueToRetryCommit(globalSession);
  67.                     if (ex instanceof TransactionException) {
  68.                         throw (TransactionException) ex;
  69.                     } else {
  70.                         throw new TransactionException(ex);
  71.                     }
  72.                 }
  74.             }
  76.         }
  77.         if (globalSession.hasBranch()) {
  78.             LOGGER.info("Global[" + globalSession.getTransactionId() + "] committing is NOT done.");
  79.             return;
  80.         }
  81.         globalSession.changeStatus(GlobalStatus.Committed);
  82.         globalSession.end();
  83.         LOGGER.info("Global[" + globalSession.getTransactionId() + "] committing is successfully done.");
  84.     }

5、rollback操作

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

  1. @Override
  2.     public GlobalStatus rollback(String xid) throws TransactionException {
  3.         GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid));
  4.         if (globalSession == null) {
  5.             return GlobalStatus.Finished;
  6.         }
  7.         GlobalStatus status = globalSession.getStatus();
  9.         globalSession.close(); // Highlight: Firstly, close the session, then no more branch can be registered.
  11.         if (status == GlobalStatus.Begin) {
  12.             globalSession.changeStatus(GlobalStatus.Rollbacking);
  13.             doGlobalRollback(globalSession, false);
  15.         }
  16.         return globalSession.getStatus();
  17.     }
  19.     @Override
  20.     public void doGlobalRollback(GlobalSession globalSession, boolean retrying) throws TransactionException {
  21.         for (BranchSession branchSession : globalSession.getReverseSortedBranches()) {
  22.             BranchStatus currentBranchStatus = branchSession.getStatus();
  23.             if (currentBranchStatus == BranchStatus.PhaseOne_Failed) {
  24.                 continue;
  25.             }
  26.             try {
  27.                 BranchStatus branchStatus = resourceManagerInbound.branchRollback(XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(),
  28.                         branchSession.getResourceId(), branchSession.getApplicationData());
  30.                 switch (branchStatus) {
  31.                     case PhaseTwo_Rollbacked:
  32.                         globalSession.removeBranch(branchSession);
  33.                         LOGGER.error("Successfully rolled back branch " + branchSession);
  34.                         continue;
  35.                     case PhaseTwo_RollbackFailed_Unretriable:
  36.                         GlobalStatus currentStatus = globalSession.getStatus();
  37.                         if (currentStatus.name().startsWith("Timeout")) {
  38.                             globalSession.changeStatus(GlobalStatus.TimeoutRollbackFailed);
  39.                         } else {
  40.                             globalSession.changeStatus(GlobalStatus.RollbackFailed);
  41.                         }
  42.                         globalSession.end();
  43.                         LOGGER.error("Failed to rollback global[" + globalSession.getTransactionId() + "] since branch[" + branchSession.getBranchId() + "] rollback failed");
  44.                         return;
  45.                     default:
  46.                         LOGGER.info("Failed to rollback branch " + branchSession);
  47.                         if (!retrying) {
  48.                             queueToRetryRollback(globalSession);
  49.                         }
  50.                         return;
  52.                 }
  53.             } catch (Exception ex) {
  54.                 LOGGER.info("Exception rollbacking branch " + branchSession, ex);
  55.                 if (!retrying) {
  56.                     queueToRetryRollback(globalSession);
  57.                     if (ex instanceof TransactionException) {
  58.                         throw (TransactionException) ex;
  59.                     } else {
  60.                         throw new TransactionException(ex);
  61.                     }
  62.                 }
  64.             }
  66.         }
  67.         GlobalStatus currentStatus = globalSession.getStatus();
  68.         if (currentStatus.name().startsWith("Timeout")) {
  69.             globalSession.changeStatus(GlobalStatus.TimeoutRollbacked);
  70.         } else {
  71.             globalSession.changeStatus(GlobalStatus.Rollbacked);
  72.         }
  73.         globalSession.end();
  74.     }

总结：

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

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

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

发表评论

表情：
评论列表 （有 0 条评论，219人围观）

还没有评论，来说两句吧...

相关阅读

    相关 TCS 贪吃蛇C源码

    linux版：这是我自己独立制作的自认为比较成功的小游戏，贪吃蛇主要运用了链表技术以及排序算法。 原版是2014年的windows版后期我移植到了linux: //

    我不是女神ヾ我不是女神ヾ/ 0 赞/ 289 阅读

    相关 分布式协调服务Zookeeper

    分布式系统介绍 分布式系统的定义 《分布式系统原理和范型》一书中定义：分布式系统是若干独立计算机的集合，这些计算机对于用户来说就像是单个相关系统。 从进程角度看，两

    Myth丶恋晨Myth丶恋晨/ 0 赞/ 435 阅读