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

前言

通過中期彙報交流會，筆者對Subset業務流程有了一個較為深刻的了解；同時也對前期的一些誤區有了認識。本篇為更新Subset業務分析，以及糾正誤區。

1. Subset不是負載均衡

簡單描述前期工作的誤區；

1.1 任務需求

在項目開展之初，筆者隻知道Subset路由規則是建立在原有負載均衡邏輯之上，是以花了大量時間在負債均衡上：

Tars | 第4篇 Subset路由規則業務分析與源碼探索前言1. Subset不是負載均衡2. 從頭開始源碼分析3. Subset應該添加在哪4. 總結最後

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的格式

通過以下測試方法我們可以知道格式化後是字元串格式如下：

[email protected] -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

Tars | 第4篇 Subset路由規則業務分析與源碼探索前言1. Subset不是負載均衡2. 從頭開始源碼分析3. Subset應該添加在哪4. 總結最後

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規則過濾節點，并且進行節點清單的格式化操作。

最後

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

Tars | 第4篇 Subset路由規則業務分析與源碼探索前言1. Subset不是負載均衡2. 從頭開始源碼分析3. Subset應該添加在哪4. 總結最後

Tars | 第4篇 Subset路由規則業務分析與源碼探索前言1. Subset不是負載均衡2. 從頭開始源碼分析3. Subset應該添加在哪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()方法

最後

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method