碼上快樂

相關內容    簡體    繁體

Tars | 第4篇 Subset路由規則業務分析與源碼探索

本文轉載自   查看原文   2021-09-10 19:28   97    騰訊犀牛鳥開源人才培養計划 | Tars學習筆記/ 學習筆記

目錄

前言

通過中期匯報交流會，筆者對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：按染色路由；

新的業務流程是是：

  1. 首先判斷是否為按比例 / 染色路由，並調用對應負載均衡調用器；
  2. 接着進行原負載均衡邏輯；
  3. 將路由結果封裝到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規則過濾節點，並且進行節點列表的格式化操作。


最后

新人制作，如有錯誤，歡迎指出，感激不盡！
歡迎關注公眾號，會分享一些更日常的東西！
如需轉載，請標注出處！


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 Tars-Java客戶端源碼分析 CoreCLR源碼探索(八) JIT的工作原理(詳解篇) 第二篇:API的URL規則設計、帶參數的路由 Ansible源碼分析【第四篇】： Ansible源碼分析之API tars framework 源碼解讀(三) servant部分章節。客戶端部分 1。完整的tars調用流程詳解 lua源碼學習篇二：語法分析 業務邏輯-規則引擎RulesEngine 深耕業務 ---- 探索復雜前端業務的開發與設計 鴻蒙內核源碼分析(源碼結構篇) | 關於內核這些問題你知道多少? Ansible源碼分析【第一篇】： Ansible之獲取源碼
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM