Tars | 第4篇 Subset路由规则业务分析与源码探索

作者: 多氯环己烷 | 来源:发表于2021-09-08 21:21 被阅读0次

前言

通过中期汇报交流会，笔者对Subset业务流程有了一个较为深刻的了解；同时也对前期的一些误区有了认识。本篇为更新Subset业务分析，以及纠正误区。

1. Subset不是负载均衡

简单描述前期工作的误区；

1.1 任务需求

在项目开展之初，笔者只知道Subset路由规则是建立在原有负载均衡逻辑之上，因此花了大量时间在负债均衡上：

任务需求

1.2 负载均衡源码结构图

通过源码分析（详情参照往期文章），可以得到TarsJava里负载均衡的的源码结构图，（基于TarsJava SpringBoot）：

@EnableTarsServer注解：表明这是一个Tars服务；

@Import(TarsServerConfiguration.class)：引入Tars服务相关配置文件；
- Communcator：通信器；
  - getServantProxyFactory()：获取代理工厂管理者；
  - getObjectProxyFactory()：获取对象代理工厂；
    - createLoadBalance()：创建客户端负载均衡调用器；
      - select()：选择负载均衡调用器（有四种模式可以选择）；
        
        invoker：调用器；
        
        invoke()：具体的执行方法；
        
        doInvokeServant()：最底层的执行方法；
      - refresh()：更新负载均衡调用器；
    - createProtocolInvoker()：创建协议调用器；

1.3 负载均衡四种调用器

其中负载均衡跟流量分配与路由强相关，而在TarsJava里，负载均衡有四种调用器可供选择：

ConsistentHashLoadBalance：一致hash选择器；
HashLoadBalance：hash选择器；
RoundRobinLoadBalance: 轮询选择器；
DefaultLoadBalance：默认的选择器（由源码可知先ConsistentHashLoadBalance，HashLoadBalance，RoundRobinLoadBalance）；

1.4 新增两种负载均衡调用器

结合需求文档，笔者以为Subset就是增加两个负载均衡调用器：

ProportionLoadBalance：按比例路由；
DyeLoadBalance：按染色路由；

新的业务流程是是：

首先判断是否为按比例 / 染色路由，并调用对应负载均衡调用器；
接着进行原负载均衡逻辑；
将路由结果封装到status里；

1.5 Subset应该是“过滤”节点而不是“选择”节点

这样理解并没有错，因为Subset路由规则就是在负载均衡之前；但准确来说，这样理解其实是有误的，因为Subset不是负载均衡。

subset是set的子集，所以是如果subset字段有设置的话，是在负责均衡之前，需要先根据subset字段类似于set选择活跃节点的那里，根据规则选出subset的活跃节点。

也就是说，Subset更多的起到的作用不是负载均衡那样的选择节点（返回一个），而是更像过滤器那样的过滤节点（返回多个）。

因此有必要重新分析源码，找到客户端获取服务节点的源码位置，并分析理解。

2. 从头开始源码分析

我们需要找到获取服务端节点的地方。

由于有前面的源码基础，我们可以很快定位到源码的这个位置：

@EnableTarsServer注解：表明这是一个Tars服务；

@Import(TarsServerConfiguration.class)：引入Tars服务相关配置文件；
- Communcator：通信器；
  - getServantProxyFactory()：获取代理工厂管理者；
  - getObjectProxyFactory()：获取对象代理工厂；

2.1 getObjectProxyFactory()源码分析

protected ObjectProxyFactory getObjectProxyFactory() {
    return objectProxyFactory;
}

getObjectProxyFactory()方法返回一个ObjectProxyFactory对象代理工厂，我们点进去看看这个工厂干了什么：

public <T> ObjectProxy<T> getObjectProxy(Class<T> api, String objName, String setDivision, ServantProxyConfig servantProxyConfig,
                                         LoadBalance<T> loadBalance, ProtocolInvoker<T> protocolInvoker) throws ClientException {
    //服务代理配置
    if (servantProxyConfig == null) {
        servantProxyConfig = createServantProxyConfig(objName, setDivision);
    } else {
        servantProxyConfig.setCommunicatorId(communicator.getId());
        servantProxyConfig.setModuleName(communicator.getCommunicatorConfig().getModuleName(), communicator.getCommunicatorConfig().isEnableSet(), communicator.getCommunicatorConfig().getSetDivision());
        servantProxyConfig.setLocator(communicator.getCommunicatorConfig().getLocator());
        addSetDivisionInfo(servantProxyConfig, setDivision);
        servantProxyConfig.setRefreshInterval(communicator.getCommunicatorConfig().getRefreshEndpointInterval());
        servantProxyConfig.setReportInterval(communicator.getCommunicatorConfig().getReportInterval());
    }

    //更新服务端节点
    updateServantEndpoints(servantProxyConfig);

    //创建负载均衡
    if (loadBalance == null) {
        loadBalance = createLoadBalance(servantProxyConfig);
    }

    //创建协议调用
    if (protocolInvoker == null) {
        protocolInvoker = createProtocolInvoker(api, servantProxyConfig);
    }
    return new ObjectProxy<T>(api, servantProxyConfig, loadBalance, protocolInvoker, communicator);
}

工厂的核心作用是生成代理对象，在这里，先是进行服务配置，更新服务端节点，然后创建负载均衡与协议调用，最后将配置好的代理对象返回。

2.2 updateServantEndpoints()更新服务端节点源码分析

我们需要关注和的地方就在updateServantEndpoints()更新服务端节点方法里，我们找到这个方法的源码如下：

private void updateServantEndpoints(ServantProxyConfig cfg) {
    CommunicatorConfig communicatorConfig = communicator.getCommunicatorConfig();

    String endpoints = null;
    if (!ParseTools.hasServerNode(cfg.getObjectName()) && !cfg.isDirectConnection() && !communicatorConfig.getLocator().startsWith(cfg.getSimpleObjectName())) {
        try {
            /** 从注册表服务器查询服务器节点 */
            if (RegisterManager.getInstance().getHandler() != null) {
                //解析出服务端节点，用“:”隔离
                endpoints = ParseTools.parse(RegisterManager.getInstance().getHandler().query(cfg.getSimpleObjectName()),
                        cfg.getSimpleObjectName());
            } else {
                endpoints = communicator.getQueryHelper().getServerNodes(cfg);
            }
            if (StringUtils.isEmpty(endpoints)) {
                throw new CommunicatorConfigException(cfg.getSimpleObjectName(), "servant node is empty on get by registry! communicator id=" + communicator.getId());
            }
            ServantCacheManager.getInstance().save(communicator.getId(), cfg.getSimpleObjectName(), endpoints, communicatorConfig.getDataPath());

        } catch (CommunicatorConfigException e) {
            /** 如果失败，将其从本地缓存文件中取出 */
            endpoints = ServantCacheManager.getInstance().get(communicator.getId(), cfg.getSimpleObjectName(), communicatorConfig.getDataPath());
            logger.error(cfg.getSimpleObjectName() + " error occurred on get by registry, use by local cache=" + endpoints + "|" + e.getLocalizedMessage(), e);
        }

        if (StringUtils.isEmpty(endpoints)) {
            throw new CommunicatorConfigException(cfg.getSimpleObjectName(), "error occurred on create proxy, servant endpoint is empty! locator =" + communicatorConfig.getLocator() + "|communicator id=" + communicator.getId());
        }

        //将服务端节点信息保存进CommunicatorConfig配置项的ObjectName属性里
        cfg.setObjectName(endpoints);
    }

    if (StringUtils.isEmpty(cfg.getObjectName())) {
        throw new CommunicatorConfigException(cfg.getSimpleObjectName(), "error occurred on create proxy, servant endpoint is empty!");
    }
}

方法的核心功能在try语句那里，就是去获取服务端的所有结点，获取的逻辑是：

如果服务器没有实例化，就从CommunicatorConfig通信器配置项中通过getServerNodes()方法获取服务节点列表；
如果服务器已经实例化，就根据挂载的服务名获取服务节点列表；
如果上述操作失败，就从缓存中获取服务节点列表；

2.3 getServerNodes()获取服务端节点源码分析

可以看出获取服务端节点的核心方法是getServerNodes()，源码如下：

public String getServerNodes(ServantProxyConfig config) {
    QueryFPrx queryProxy = getPrx();
    String name = config.getSimpleObjectName();
    //存活的节点
    Holder<List<EndpointF>> activeEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
    //挂掉的节点
    Holder<List<EndpointF>> inactiveEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
    int ret = TarsHelper.SERVERSUCCESS;
    //判断是否为启用集
    if (config.isEnableSet()) {
        ret = queryProxy.findObjectByIdInSameSet(name, config.getSetDivision(), activeEp, inactiveEp);
    } else {
        ret = queryProxy.findObjectByIdInSameGroup(name, activeEp, inactiveEp);
    }

    if (ret != TarsHelper.SERVERSUCCESS) {
        return null;
    }
    Collections.sort(activeEp.getValue());
    //value就是最后的节点参数

    //将获取到的节点列表格式化为一个字符串格式
    StringBuilder value = new StringBuilder();
    if (activeEp.value != null && !activeEp.value.isEmpty()) {
        for (EndpointF endpointF : activeEp.value) {
            if (value.length() > 0) {
                value.append(":");
            }
            value.append(ParseTools.toFormatString(endpointF, true));
        }
    }
    
    //个格式化后的字符串加上Tars的服务名
    if (value.length() < 1) {
        return null;
    }
    value.insert(0, Constants.TARS_AT);
    value.insert(0, name);
    return value.toString();
}

getServerNodes()的处理逻辑是：

getServerNodes()首先创建两个Holder对象，用来保存存活节点列表activeEp与不存活节点列表inactiveEp的值；
接着判断是否为启用集，使用对象代理的方式，调用findObjectByIdInSameSet()或findObjectByIdInSameGroup()方法获取到存活与不存活节点列表的值封装进activeEp与inactiveEp里；
将获取到的节点列表格式化为一个字符串格式value；
进行后续格式化操作；

2.4 endpoints的格式

通过以下测试方法我们可以知道格式化后是字符串格式如下：

abc@tcp -h host1 -p 1 -t 3000 -a 1 -g 4 -s setId1 -v 10 -w 9:tcp -h host2 -p 1 -t 3000 -a 1 -g 4 -s setId2 -v 10 -w 9

endpoints格式

3. Subset应该添加在哪

Subset应该在节点列表格式化之前。

3.1 获取服务端节点的源码结构图

通过上述分析，我们可得出获取服务端节点getServerNodes()的源码结构图：

@EnableTarsServer注解：表明这是一个Tars服务；

@Import(TarsServerConfiguration.class)：引入Tars服务相关配置文件；
- Communcator：通信器；
  - getServantProxyFactory()：获取代理工厂管理者；
  - getObjectProxyFactory()：获取对象代理工厂；
    - updateServantEndpoints()：更新服务端节点；
      - getServerNodes()：获取服务节点列表；

3.2 修改getServerNodes()方法

由上述分析，我们可以知道：getServerNodes()的处理逻辑是：

首先创建两个Holder对象；
接着获取到存活与不存活节点列表的值封装进activeEp与inactiveEp里；
将获取到的节点列表格式化为一个字符串格式value；
进行后续格式化操作；

我们要在数据格式化前将列表里的节点进行过滤，不然如果先格式化字符串再过滤，将会涉及字符串的操作，当服务的节点过多是，这部分字符串的校验与判断将会十分消耗性能，因此要在格式化前通过Subset规则过滤节点，修改后的getServerNodes()方法如下：

public String getServerNodes(ServantProxyConfig config) {
    QueryFPrx queryProxy = getPrx();
    String name = config.getSimpleObjectName();
    //存活的节点
    Holder<List<EndpointF>> activeEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
    //挂掉的节点
    Holder<List<EndpointF>> inactiveEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
    int ret = TarsHelper.SERVERSUCCESS;
    //判断是否为启用集
    if (config.isEnableSet()) {
        ret = queryProxy.findObjectByIdInSameSet(name, config.getSetDivision(), activeEp, inactiveEp);
    } else {
        ret = queryProxy.findObjectByIdInSameGroup(name, activeEp, inactiveEp);
    }

    if (ret != TarsHelper.SERVERSUCCESS) {
        return null;
    }
    Collections.sort(activeEp.getValue());
    //value就是最后的节点参数

//        //将获取到的节点列表格式化为一个字符串格式
//        StringBuilder value = new StringBuilder();
//        if (activeEp.value != null && !activeEp.value.isEmpty()) {
//            for (EndpointF endpointF : activeEp.value) {
//                if (value.length() > 0) {
//                    value.append(":");
//                }
//                value.append(ParseTools.toFormatString(endpointF, true));
//            }
//        }

    //对上述注释代码做抽取，增加按subset规则过滤节点
    StringBuilder value = filterEndpointsBySubset(activeEp, config);

    //个格式化后的字符串加上Tars的服务名
    if (value.length() < 1) {
        return null;
    }
    value.insert(0, Constants.TARS_AT);
    value.insert(0, name);
    return value.toString();
}

修改的逻辑是：

抽取将节点列表格式化为一个字符串格式value的代码；
添加filterEndpointsBySubset(activeEp, config)根据Subset规则过滤节点方法；
- 该方法的参数为存活节点列表与路由规则；
- 该方法的逻辑是先进行Subset规则判断，再进行节点列表的数据格式；

3.3 添加的filterEndpointsBySubset()方法

该方法的实现逻辑代码如下：

public StringBuilder filterEndpointsBySubset(Holder<List<EndpointF>> activeEp, ServantProxyConfig config){
    StringBuilder value = new StringBuilder();

    //config的非空判断
    if(config == null){
        return null;
    }
    String ruleType = config.getRuleType();
    Map<String, String> ruleData = config.getRuleData();
    String routeKey = config.getRouteKey();
    if(ruleData.size() < 1 || ruleType == null){
        return null;
    }

    //按比例路由
    if(Constants.TARS_SUBSET_PROPORTION.equals(ruleType)){
        int totalWeight = 0;
        int supWeight = 0;
        String subset = null;
        //获得总权重
        for(String weight : ruleData.values()){
            totalWeight+=Integer.parseInt(weight);
        }
        //获取随机数
        Random random = new Random();
        int r = random.nextInt(totalWeight);
        //根据随机数找到subset
        for (Map.Entry<String, String> entry : ruleData.entrySet()){
            supWeight+=Integer.parseInt(entry.getValue());
            if( r < supWeight){
                subset = entry.getKey();
                break;
            }
        }
        //利用subset过滤不符合条件的节点
        if (activeEp.value != null && !activeEp.value.isEmpty()) {
            for (EndpointF endpointF : activeEp.value) {
                //subset判断
                if(endpointF != null && endpointF.getSubset().equals(subset)){
                    if (value.length() > 0) {
                        value.append(":");
                    }
                    value.append(ParseTools.toFormatString(endpointF, true));
                }

            }
        }
        return value;
    }

    //按请求参数路由
    if(Constants.TARS_SUBSET_PARAMETER.equals(ruleType)){
        //获取将要路由到的路径
        String route = ruleData.get("route");
        if( route == null ){
            return null;
        }

        //判断是否含有键“equal”；“match”，并获取染色Key
        String key;
        if(ruleData.containsKey("equal")){
            //精确路由
            key = ruleData.get("equal");
            //对染色Key做非空校验
            if(key == null || "".equals(key)){
                return null;
            }

            //利用subset过滤不符合条件的节点
            if (activeEp.value != null && !activeEp.value.isEmpty()) {
                for (EndpointF endpointF : activeEp.value) {
                    //subset判断，精确判断
                    if(endpointF != null && routeKey.equals(key) && route.equals(endpointF.getSubset())){
                        if (value.length() > 0) {
                            value.append(":");
                        }
                        value.append(ParseTools.toFormatString(endpointF, true));
                    }
                }
            }
        } else if( ruleData.containsKey("match")){
            //正则路由
            key = ruleData.get("match");
            //对染色Key做非空校验
            if(key == null || "".equals(key)){
                return null;
            }

            //利用subset过滤不符合条件的节点
            if (activeEp.value != null && !activeEp.value.isEmpty()) {
                for (EndpointF endpointF : activeEp.value) {
                    //subset判断，正则规则
                    if(endpointF != null && StringUtils.matches(key, routeKey) && route.equals(endpointF.getSubset())){
                        if (value.length() > 0) {
                            value.append(":");
                        }
                        value.append(ParseTools.toFormatString(endpointF, true));
                    }

                }
            }
        } else {
            //【报错】
            return null;
        }
        return value;
    }
    //无规则路由
    if(Constants.TARS_SUBSET_DEFAULT.equals(ruleType)){
        //获取将要路由到的路径
        String route = ruleData.get("default");
        if( route == null ){
            return null;
        }
        //利用subset过滤不符合条件的节点
        if (activeEp.value != null && !activeEp.value.isEmpty()) {
            for (EndpointF endpointF : activeEp.value) {
                //subset判断
                if(endpointF != null && endpointF.getSubset().equals(route)){
                    if (value.length() > 0) {
                        value.append(":");
                    }
                    value.append(ParseTools.toFormatString(endpointF, true));
                }

            }
        }
        return value;

    }
    return value;
}

由于方法比较冗余，但思路没错，测试跑的通，后期需要进一步修改简化、优化。

4. 总结

4.1 Subset不是负载均衡

Subset流量路由应该在负载均衡之前，相当于一个过滤器。

4.2 getServerNodes()的源码结构图

可以知道获取服务端节点的思想逻辑，获取服务端节点getServerNodes()的源码结构图：

@EnableTarsServer注解：表明这是一个Tars服务；

@Import(TarsServerConfiguration.class)：引入Tars服务相关配置文件；
- Communcator：通信器；
  - getServantProxyFactory()：获取代理工厂管理者；
  - getObjectProxyFactory()：获取对象代理工厂；
    - updateServantEndpoints()：更新服务端节点；
      - getServerNodes()：获取服务节点列表；

4.3 核心在filterEndpointsBySubset()方法

该方法的主要作用为根据Subset规则过滤节点，并且进行节点列表的格式化操作。

最后

$\color{blue}{\rm\small{新人制作，如有错误，欢迎指出，感激不尽！}}$

$\color{blue}{\rm\small{欢迎关注我，并与我交流！}}$

$\color{blue}{\rm\small{如需转载，请标注出处！}}$

Tars | 第4篇 Subset路由规则业务分析与源码探索

前言

1. Subset不是负载均衡

1.1 任务需求

1.2 负载均衡源码结构图

1.3 负载均衡四种调用器

1.4 新增两种负载均衡调用器

1.5 Subset应该是“过滤”节点而不是“选择”节点

2. 从头开始源码分析

2.1 getObjectProxyFactory()源码分析

2.2 updateServantEndpoints()更新服务端节点源码分析

2.3 getServerNodes()获取服务端节点源码分析

2.4 endpoints的格式

3. Subset应该添加在哪

3.1 获取服务端节点的源码结构图

3.2 修改getServerNodes()方法

3.3 添加的filterEndpointsBySubset()方法

4. 总结

4.1 Subset不是负载均衡

4.2 getServerNodes()的源码结构图

4.3 核心在filterEndpointsBySubset()方法

最后

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读