前言
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()方法
最后

前言

通過中期匯報交流會，筆者對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 = https://www.cnblogs.com/dlhjw/archive/2021/09/10/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 = https://www.cnblogs.com/dlhjw/archive/2021/09/10/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 = https://www.cnblogs.com/dlhjw/archive/2021/09/10/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 = https://www.cnblogs.com/dlhjw/archive/2021/09/10/new StringBuilder();

    //config的非空判斷
    if(config == null){
        return null;
    }
    String ruleType = config.getRuleType();
    Map 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 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規則過濾節點，并且進行節點串列的格式化操作，

最后

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

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/299079.html

標籤：其他

上一篇：用pipenv管理python虛擬環境

下一篇：python-openpyxl Excel的單元格樣式設定，包括字體、樣式、寬高等等！

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()方法

最后