本文基于seata 1.3.0版本

在《Seata解析-seata核心类NettyRemotingServer详解》中介绍了RegTmProcessor和RegRmProcessor，这两个处理器用于处理TM和RM注册，本文将详细介绍服务端如何注册TM和RM。

一、TM注册

先来介绍TM的注册流程。
服务端在收到TM的注册请求后，会将请求转化为对象RegisterTMRequest，然后将对象转发给RegTmProcessor的onRegTmMessage方法处理。
下面是onRegTmMessage方法，其中的代码有删减，只展示了核心逻辑：

private void onRegTmMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
        //得到TM注册请求消息
        RegisterTMRequest message = (RegisterTMRequest) rpcMessage.getBody();
        ...
        try {
            if (null == checkAuthHandler || checkAuthHandler.regTransactionManagerCheckAuth(message)) {
                //TM信息注册到服务器端
                ChannelManager.registerTMChannel(message, ctx.channel());
                ...
            }
        } catch (Exception exx) {
            ...
        }
        //异步发送响应消息
        RegisterTMResponse response = new RegisterTMResponse(isSuccess);
        remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), response);
    }

onRegTmMessage方法最后将TM的注册请求转发给了ChannelManager.registerTMChannel处理。registerTMChannel的处理流程也非常简单分为以下四步：

校验客户端版本号；
构建RpcContext对象；
在一个全局Map属性中保存客户端链接通道与RpcContext对象的对应关系；
将客户端的应用名、IP、端口与RpcContext对象的对应关系也保存到一个全局Map对象中；

下面来看一下registerTMChannel方法的具体实现：

public static void registerTMChannel(RegisterTMRequest request, Channel channel) throws IncompatibleVersionException {
        //校验请求方的版本信息，版本必须大于等于0.7.1
        Version.checkVersion(request.getVersion());
        //构建RpcContext对象
        RpcContext rpcContext = buildChannelHolder(NettyPoolKey.TransactionRole.TMROLE, request.getVersion(),
            request.getApplicationId(),
            request.getTransactionServiceGroup(),
            null, channel);
        //将当前连接通道channel与rpcContext之间的对应关系添加到IDENTIFIED_CHANNELS中
        //另外每个rpcContext也都会持有IDENTIFIED_CHANNELS
        //IDENTIFIED_CHANNELS维护全局连接通道channel与RpcCcontext之间的关系，所以通过RpcContext对象可以查看到所有的连接以及对应的RpcContext
        rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
        //clientIdentified=客户端应用名+客户端IP
        String clientIdentified = rpcContext.getApplicationId() + Constants.CLIENT_ID_SPLIT_CHAR
            + ChannelUtil.getClientIpFromChannel(channel);
        //TM_CHANNELS记录所有的TM连接信息，类型是ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>
        //TM_CHANNELS的key=客户端应用名+客户端IP
     	//TM_CHANNELS的value中的key是客户端连接服务端使用的端口号，value中的value是对应的RpcContext对象
        TM_CHANNELS.putIfAbsent(clientIdentified, new ConcurrentHashMap<Integer, RpcContext>());
        //clientIdentifiedMap的key表示是客户端链接的端口号，value是对应的RpcContext对象
        ConcurrentMap<Integer, RpcContext> clientIdentifiedMap = TM_CHANNELS.get(clientIdentified);
        rpcContext.holdInClientChannels(clientIdentifiedMap);
    }

registerTMChannel方法调用了RpcContext中的holdInIdentifiedChannels和holdInClientChannels方法，下面来一一看每个方法的具体实现。
首先看一下创建RpcContext对象的方法buildChannelHolder：

private static RpcContext buildChannelHolder(NettyPoolKey.TransactionRole clientRole, String version, String applicationId,
                                                 String txServiceGroup, String dbkeys, Channel channel) {
        RpcContext holder = new RpcContext();
        //客户端的角色，可以是TM或者RM、SERVER，角色由TransactionRole描述
        holder.setClientRole(clientRole);
        //客户端的版本
        holder.setVersion(version);
        //clientId=客户端的应用名+IP+端口
        holder.setClientId(buildClientId(applicationId, channel));
        //applicationId是应用名，通过spring.application.name设置
        holder.setApplicationId(applicationId);
        //事务分组，可以通过seata.tx-service-group设置，
        //事务分组是为查找TM使用，事务分组以后会单独做介绍
        holder.setTransactionServiceGroup(txServiceGroup);
        //resources表示的是客户端连接的数据库资源，保存了客户端连接数据库的URL
        holder.addResources(dbKeytoSet(dbkeys));
        //客户端连接
        holder.setChannel(channel);
        return holder;
    }

registerTMChannel调用buildChannelHolder创建出RpcContext对象后，继续调用holdInIdentifiedChannels和holdInClientChannels以完成TM的注册：

public void holdInIdentifiedChannels(ConcurrentMap<Channel, RpcContext> clientIDHolderMap) {
        if (this.clientIDHolderMap != null) {
            throw new IllegalStateException();
        }
        this.clientIDHolderMap = clientIDHolderMap;
        this.clientIDHolderMap.put(channel, this);
    }
    public void holdInClientChannels(ConcurrentMap<Integer, RpcContext> clientTMHolderMap) {
        if (this.clientTMHolderMap != null) {
            throw new IllegalStateException();
        }
        this.clientTMHolderMap = clientTMHolderMap;
        //获取连接的端口号
        Integer clientPort = ChannelUtil.getClientPortFromChannel(channel);
        this.clientTMHolderMap.put(clientPort, this);
    }

这两个方法都比较简单，都是将相关信息直接保存到全局的Map对象中。
TM信息注册成功后，构建RegisterTMResponse对象返回给客户端。
到此TM的注册流程全部结束。
从上面的流程可以看出，TM注册是将TM的应用信息和连接通道保存到全局Map对象中，并创建RpcContext上下文对象，该对象将贯穿对应链接的整个生命周期。

二、RM注册

RM注册与TM注册非常类似，服务端收到请求后，将请求对象转发给RegRmProcessor的onRegRmMessage方法处理，该方法与onRegTmMessage类似，这里不再做介绍。onRegTmMessage再将请求对象转发给ChannelManager.registerRMChannel处理，下面来看一下这个方法：

public static void registerRMChannel(RegisterRMRequest resourceManagerRequest, Channel channel)
        throws IncompatibleVersionException {
        //检查版本信息
        Version.checkVersion(resourceManagerRequest.getVersion());
        //构建RM的资源集合，这里的资源和TM中的资源一样，都是连接数据库的URL
        Set<String> dbkeySet = dbKeytoSet(resourceManagerRequest.getResourceIds());
        RpcContext rpcContext;
        if (!IDENTIFIED_CHANNELS.containsKey(channel)) {
            //如果channel没有注册过，那么创建rpcContext对象
            rpcContext = buildChannelHolder(NettyPoolKey.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 (dbkeySet == null || dbkeySet.isEmpty()) { return; }
        //注册每个数据库资源
        for (String resourceId : dbkeySet) {
            String clientIp;
            ConcurrentMap<Integer, RpcContext> portMap = RM_CHANNELS.computeIfAbsent(resourceId, resourceIdKey -> new ConcurrentHashMap<>())
                    .computeIfAbsent(resourceManagerRequest.getApplicationId(), applicationId -> new ConcurrentHashMap<>())
                    .computeIfAbsent(clientIp = ChannelUtil.getClientIpFromChannel(channel), clientIpKey -> new ConcurrentHashMap<>());
            //将端口与资源的对应关系保存到portMap中
            rpcContext.holdInResourceManagerChannels(resourceId, portMap);
            updateChannelsResource(resourceId, clientIp, resourceManagerRequest.getApplicationId());
        }

    }
    //更新连接资源，也就是更新RM的资源信息
    private static void updateChannelsResource(String resourceId, String clientIp, String applicationId) {
        ConcurrentMap<Integer, RpcContext> sourcePortMap = RM_CHANNELS.get(resourceId).get(applicationId).get(clientIp);
        for (ConcurrentMap.Entry<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>>> rmChannelEntry : RM_CHANNELS.entrySet()) {
            //如果资源已经注册过，则跳过
            if (rmChannelEntry.getKey().equals(resourceId)) {
                continue;
            }
            ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> applicationIdMap = rmChannelEntry.getValue();
            //如果应用不同，则跳过
            if (!applicationIdMap.containsKey(applicationId)) {
                continue;
            }
            ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> clientIpMap = applicationIdMap.get(applicationId);
            //如果IP不同，则跳过
            if (!clientIpMap.containsKey(clientIp)) {
                continue;
            }
            ConcurrentMap<Integer, RpcContext> portMap = clientIpMap.get(clientIp);
            for (ConcurrentMap.Entry<Integer, RpcContext> portMapEntry : portMap.entrySet()) {
                Integer port = portMapEntry.getKey();
                if (!sourcePortMap.containsKey(port)) {
                    RpcContext rpcContext = portMapEntry.getValue();
                    //新增资源注册信息中不包含旧资源的端口，
                    //那么将旧资源的端口与RpcContext保存到sourcePortMap中，也就是保存到新增资源中
                    sourcePortMap.put(port, rpcContext);
                    //同时也将新增资源的数据库URL保存到旧资源的RpcContext中
                    //保存：资源ID->RM连接端口->RpcContext 到Map对象中
                    rpcContext.holdInResourceManagerChannels(resourceId, port);
                }
            }
        }
    }

updateChannelsResource用于更新RM的资源信息，这个逻辑比较繁琐，总起来说作用是：如果新注册的数据库资源与已经注册过的旧资源属于同一个应用，且IP相同，但是端口不同，这说明，当前注册的资源与旧资源所属的应用是同一个，且在一台机器上部署，只是端口不同，那么seata会将旧资源的端口以及RpcContext对应关系注册到新资源里面，同时在旧资源的RpcContext中也增加新资源的信息。
这里就引发了两个问题：

为什么TM注册时，不判断连接是否注册过，而RM需要？
为什么RM最后需要更新资源信息，也就是调用updateChannelsResource方法的作用是什么？

对于第一个问题，待到分析TM和RM时，再解释原因。
第二个问题更新资源的原因是如果与RM的连接断开了，可以使用其他通道与RM进行通讯，因为RM与分支事务有关，比如通知分支事务回滚，而此时与RM的连接断开了，那么seata会选择同一个IP上同一个应用的不同端口的连接进行通知，以此来保证事务的一致性。

三、总结

本文分析了TM和RM在服务端的注册流程，总起来说，两者的注册流程非常相似，首先构建RpcContext对象，然后将该对象与应用信息一起存放到内存的Map对象中。RpcContext对象会贯穿整个连接的生命周期。

Seata解析-TM和RM服务端注册过程详解

一、TM注册

二、RM注册

三、总结

相关文章