前言

通過中期彙報交流會，筆者對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()原始碼分析

  1. protected ObjectProxyFactory getObjectProxyFactory() {
  2.     return objectProxyFactory;
  3. }

getObjectProxyFactory()方法返回一個ObjectProxyFactory物件代理工廠，我們點進去看看這個工廠幹了什麼：

  1. public <T> ObjectProxy<T> getObjectProxy(Class<T> api, String objName, String setDivision, ServantProxyConfig servantProxyConfig,
  2.                                          LoadBalance<T> loadBalance, ProtocolInvoker<T> protocolInvoker) throws ClientException {
  3.     //服務代理配置
  4.     if (servantProxyConfig == null) {
  5.         servantProxyConfig = createServantProxyConfig(objName, setDivision);
  6.     } else {
  7.         servantProxyConfig.setCommunicatorId(communicator.getId());
  8.         servantProxyConfig.setModuleName(communicator.getCommunicatorConfig().getModuleName(), communicator.getCommunicatorConfig().isEnableSet(), communicator.getCommunicatorConfig().getSetDivision());
  9.         servantProxyConfig.setLocator(communicator.getCommunicatorConfig().getLocator());
  10.         addSetDivisionInfo(servantProxyConfig, setDivision);
  11.         servantProxyConfig.setRefreshInterval(communicator.getCommunicatorConfig().getRefreshEndpointInterval());
  12.         servantProxyConfig.setReportInterval(communicator.getCommunicatorConfig().getReportInterval());
  13.     }
  15.     //更新服務端節點
  16.     updateServantEndpoints(servantProxyConfig);
  18.     //建立負載均衡
  19.     if (loadBalance == null) {
  20.         loadBalance = createLoadBalance(servantProxyConfig);
  21.     }
  23.     //建立協議呼叫
  24.     if (protocolInvoker == null) {
  25.         protocolInvoker = createProtocolInvoker(api, servantProxyConfig);
  26.     }
  27.     return new ObjectProxy<T>(api, servantProxyConfig, loadBalance, protocolInvoker, communicator);
  28. }

工廠的核心作用是生成代理物件，在這裡，先是進行服務配置，更新服務端節點，然後建立負載均衡與協議呼叫，最後將配置好的代理物件返回。

2.2 updateServantEndpoints()更新服務端節點原始碼分析

我們需要關注和的地方就在updateServantEndpoints()更新服務端節點方法裡，我們找到這個方法的原始碼如下：

  1. private void updateServantEndpoints(ServantProxyConfig cfg) {
  2.     CommunicatorConfig communicatorConfig = communicator.getCommunicatorConfig();
  4.     String endpoints = null;
  5.     if (!ParseTools.hasServerNode(cfg.getObjectName()) && !cfg.isDirectConnection() && !communicatorConfig.getLocator().startsWith(cfg.getSimpleObjectName())) {
  6.         try {
  7.             /** 從登錄檔伺服器查詢伺服器節點 */
  8.             if (RegisterManager.getInstance().getHandler() != null) {
  9.                 //解析出服務端節點，用“:”隔離
  10.                 endpoints = ParseTools.parse(RegisterManager.getInstance().getHandler().query(cfg.getSimpleObjectName()),
  11.                         cfg.getSimpleObjectName());
  12.             } else {
  13.                 endpoints = communicator.getQueryHelper().getServerNodes(cfg);
  14.             }
  15.             if (StringUtils.isEmpty(endpoints)) {
  16.                 throw new CommunicatorConfigException(cfg.getSimpleObjectName(), "servant node is empty on get by registry! communicator id=" + communicator.getId());
  17.             }
  18.             ServantCacheManager.getInstance().save(communicator.getId(), cfg.getSimpleObjectName(), endpoints, communicatorConfig.getDataPath());
  20.         } catch (CommunicatorConfigException e) {
  21.             /** 如果失敗，將其從本地快取檔案中取出 */
  22.             endpoints = ServantCacheManager.getInstance().get(communicator.getId(), cfg.getSimpleObjectName(), communicatorConfig.getDataPath());
  23.             logger.error(cfg.getSimpleObjectName() + " error occurred on get by registry, use by local cache=" + endpoints + "|" + e.getLocalizedMessage(), e);
  24.         }
  26.         if (StringUtils.isEmpty(endpoints)) {
  27.             throw new CommunicatorConfigException(cfg.getSimpleObjectName(), "error occurred on create proxy, servant endpoint is empty! locator =" + communicatorConfig.getLocator() + "|communicator id=" + communicator.getId());
  28.         }
  30.         //將服務端節點資訊儲存進CommunicatorConfig配置項的ObjectName屬性裡
  31.         cfg.setObjectName(endpoints);
  32.     }
  34.     if (StringUtils.isEmpty(cfg.getObjectName())) {
  35.         throw new CommunicatorConfigException(cfg.getSimpleObjectName(), "error occurred on create proxy, servant endpoint is empty!");
  36.     }
  37. }

方法的核心功能在try語句那裡，就是去獲取服務端的所有結點，獲取的邏輯是：

  • 如果伺服器沒有例項化，就從CommunicatorConfig通訊器配置項中通過getServerNodes()方法獲取服務節點列表；
  • 如果伺服器已經例項化，就根據掛載的服務名獲取服務節點列表；
  • 如果上述操作失敗，就從快取中獲取服務節點列表；

2.3 getServerNodes()獲取服務端節點原始碼分析

可以看出獲取服務端節點的核心方法是getServerNodes()，原始碼如下：

  1. public String getServerNodes(ServantProxyConfig config) {
  2.     QueryFPrx queryProxy = getPrx();
  3.     String name = config.getSimpleObjectName();
  4.     //存活的節點
  5.     Holder<List<EndpointF>> activeEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
  6.     //掛掉的節點
  7.     Holder<List<EndpointF>> inactiveEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
  8.     int ret = TarsHelper.SERVERSUCCESS;
  9.     //判斷是否為啟用集
  10.     if (config.isEnableSet()) {
  11.         ret = queryProxy.findObjectByIdInSameSet(name, config.getSetDivision(), activeEp, inactiveEp);
  12.     } else {
  13.         ret = queryProxy.findObjectByIdInSameGroup(name, activeEp, inactiveEp);
  14.     }
  16.     if (ret != TarsHelper.SERVERSUCCESS) {
  17.         return null;
  18.     }
  19.     Collections.sort(activeEp.getValue());
  20.     //value就是最後的節點引數
  22.     //將獲取到的節點列表格式化為一個字串格式
  23.     StringBuilder value = new StringBuilder();
  24.     if (activeEp.value != null && !activeEp.value.isEmpty()) {
  25.         for (EndpointF endpointF : activeEp.value) {
  26.             if (value.length() > 0) {
  27.                 value.append(":");
  28.             }
  29.             value.append(ParseTools.toFormatString(endpointF, true));
  30.         }
  31.     }
  33.     //個格式化後的字串加上Tars的服務名
  34.     if (value.length() < 1) {
  35.         return null;
  36.     }
  37.     value.insert(0, Constants.TARS_AT);
  38.     value.insert(0, name);
  39.     return value.toString();
  40. }

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

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

  1. public String getServerNodes(ServantProxyConfig config) {
  2.     QueryFPrx queryProxy = getPrx();
  3.     String name = config.getSimpleObjectName();
  4.     //存活的節點
  5.     Holder<List<EndpointF>> activeEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
  6.     //掛掉的節點
  7.     Holder<List<EndpointF>> inactiveEp = new Holder<List<EndpointF>>(new ArrayList<EndpointF>());
  8.     int ret = TarsHelper.SERVERSUCCESS;
  9.     //判斷是否為啟用集
  10.     if (config.isEnableSet()) {
  11.         ret = queryProxy.findObjectByIdInSameSet(name, config.getSetDivision(), activeEp, inactiveEp);
  12.     } else {
  13.         ret = queryProxy.findObjectByIdInSameGroup(name, activeEp, inactiveEp);
  14.     }
  16.     if (ret != TarsHelper.SERVERSUCCESS) {
  17.         return null;
  18.     }
  19.     Collections.sort(activeEp.getValue());
  20.     //value就是最後的節點引數
  22. //        //將獲取到的節點列表格式化為一個字串格式
  23. //        StringBuilder value = new StringBuilder();
  24. //        if (activeEp.value != null && !activeEp.value.isEmpty()) {
  25. //            for (EndpointF endpointF : activeEp.value) {
  26. //                if (value.length() > 0) {
  27. //                    value.append(":");
  28. //                }
  29. //                value.append(ParseTools.toFormatString(endpointF, true));
  30. //            }
  31. //        }
  33.     //對上述註釋程式碼做抽取，增加按subset規則過濾節點
  34.     StringBuilder value = filterEndpointsBySubset(activeEp, config);
  36.     //個格式化後的字串加上Tars的服務名
  37.     if (value.length() < 1) {
  38.         return null;
  39.     }
  40.     value.insert(0, Constants.TARS_AT);
  41.     value.insert(0, name);
  42.     return value.toString();
  43. }

修改的邏輯是：

  • 抽取將節點列表格式化為一個字串格式value的程式碼；
  • 新增filterEndpointsBySubset(activeEp, config)根據Subset規則過濾節點方法；
    • 該方法的引數為存活節點列表與路由規則；
    • 該方法的邏輯是先進行Subset規則判斷，再進行節點列表的資料格式；

3.3 新增的filterEndpointsBySubset()方法

該方法的實現邏輯程式碼如下：

  1. public StringBuilder filterEndpointsBySubset(Holder<List<EndpointF>> activeEp, ServantProxyConfig config){
  2.     StringBuilder value = new StringBuilder();
  4.     //config的非空判斷
  5.     if(config == null){
  6.         return null;
  7.     }
  8.     String ruleType = config.getRuleType();
  9.     Map<String, String> ruleData = config.getRuleData();
  10.     String routeKey = config.getRouteKey();
  11.     if(ruleData.size() < 1 || ruleType == null){
  12.         return null;
  13.     }
  15.     //按比例路由
  16.     if(Constants.TARS_SUBSET_PROPORTION.equals(ruleType)){
  17.         int totalWeight = 0;
  18.         int supWeight = 0;
  19.         String subset = null;
  20.         //獲得總權重
  21.         for(String weight : ruleData.values()){
  22.             totalWeight+=Integer.parseInt(weight);
  23.         }
  24.         //獲取隨機數
  25.         Random random = new Random();
  26.         int r = random.nextInt(totalWeight);
  27.         //根據隨機數找到subset
  28.         for (Map.Entry<String, String> entry : ruleData.entrySet()){
  29.             supWeight+=Integer.parseInt(entry.getValue());
  30.             if( r < supWeight){
  31.                 subset = entry.getKey();
  32.                 break;
  33.             }
  34.         }
  35.         //利用subset過濾不符合條件的節點
  36.         if (activeEp.value != null && !activeEp.value.isEmpty()) {
  37.             for (EndpointF endpointF : activeEp.value) {
  38.                 //subset判斷
  39.                 if(endpointF != null && endpointF.getSubset().equals(subset)){
  40.                     if (value.length() > 0) {
  41.                         value.append(":");
  42.                     }
  43.                     value.append(ParseTools.toFormatString(endpointF, true));
  44.                 }
  46.             }
  47.         }
  48.         return value;
  49.     }
  51.     //按請求引數路由
  52.     if(Constants.TARS_SUBSET_PARAMETER.equals(ruleType)){
  53.         //獲取將要路由到的路徑
  54.         String route = ruleData.get("route");
  55.         if( route == null ){
  56.             return null;
  57.         }
  59.         //判斷是否含有鍵“equal”；“match”，並獲取染色Key
  60.         String key;
  61.         if(ruleData.containsKey("equal")){
  62.             //精確路由
  63.             key = ruleData.get("equal");
  64.             //對染色Key做非空校驗
  65.             if(key == null || "".equals(key)){
  66.                 return null;
  67.             }
  69.             //利用subset過濾不符合條件的節點
  70.             if (activeEp.value != null && !activeEp.value.isEmpty()) {
  71.                 for (EndpointF endpointF : activeEp.value) {
  72.                     //subset判斷，精確判斷
  73.                     if(endpointF != null && routeKey.equals(key) && route.equals(endpointF.getSubset())){
  74.                         if (value.length() > 0) {
  75.                             value.append(":");
  76.                         }
  77.                         value.append(ParseTools.toFormatString(endpointF, true));
  78.                     }
  79.                 }
  80.             }
  81.         } else if( ruleData.containsKey("match")){
  82.             //正則路由
  83.             key = ruleData.get("match");
  84.             //對染色Key做非空校驗
  85.             if(key == null || "".equals(key)){
  86.                 return null;
  87.             }
  89.             //利用subset過濾不符合條件的節點
  90.             if (activeEp.value != null && !activeEp.value.isEmpty()) {
  91.                 for (EndpointF endpointF : activeEp.value) {
  92.                     //subset判斷，正則規則
  93.                     if(endpointF != null && StringUtils.matches(key, routeKey) && route.equals(endpointF.getSubset())){
  94.                         if (value.length() > 0) {
  95.                             value.append(":");
  96.                         }
  97.                         value.append(ParseTools.toFormatString(endpointF, true));
  98.                     }
  100.                 }
  101.             }
  102.         } else {
  103.             //【報錯】
  104.             return null;
  105.         }
  106.         return value;
  107.     }
  108.     //無規則路由
  109.     if(Constants.TARS_SUBSET_DEFAULT.equals(ruleType)){
  110.         //獲取將要路由到的路徑
  111.         String route = ruleData.get("default");
  112.         if( route == null ){
  113.             return null;
  114.         }
  115.         //利用subset過濾不符合條件的節點
  116.         if (activeEp.value != null && !activeEp.value.isEmpty()) {
  117.             for (EndpointF endpointF : activeEp.value) {
  118.                 //subset判斷
  119.                 if(endpointF != null && endpointF.getSubset().equals(route)){
  120.                     if (value.length() > 0) {
  121.                         value.append(":");
  122.                     }
  123.                     value.append(ParseTools.toFormatString(endpointF, true));
  124.                 }
  126.             }
  127.         }
  128.         return value;
  130.     }
  131.     return value;
  132. }

由於方法比較冗餘，但思路沒錯，測試跑的通，後期需要進一步修改簡化、優化。

4. 總結

4.1 Subset不是負載均衡

Subset流量路由應該在負載均衡之前，相當於一個過濾器。

4.2 getServerNodes()的原始碼結構圖

可以知道獲取服務端節點的思想邏輯，獲取服務端節點getServerNodes()的原始碼結構圖：

@EnableTarsServer註解：表明這是一個Tars服務；

  • @Import(TarsServerConfiguration.class)：引入Tars服務相關配置檔案；

    • Communcator：通訊器；

      • getServantProxyFactory()：獲取代理工廠管理者；
      • getObjectProxyFactory()：獲取物件代理工廠；
        • updateServantEndpoints()： 更新服務端節點；

          • getServerNodes()：獲取服務節點列表；

4.3 核心在filterEndpointsBySubset()方法

該方法的主要作用為根據Subset規則過濾節點，並且進行節點列表的格式化操作。

最後

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