Tars | 第4篇 Subset路由规则业务分析与源码探索
目录
前言
通过中期汇报交流会，笔者对Subset业务流程有了一个较为深刻的了解；同时也对前期的一些误区有了认识。本篇为更新Subset业务分析，以及纠正误区。
1. Subset不是负载均衡
简单描述前期工作的误区；
1.1 任务需求
在项目开展之初，笔者只知道Subset路由规则是建立在原有负载均衡逻辑之上，因此花了大量时间在负债均衡上：
1.2 负载均衡源码结构图
通过源码分析（详情参照往期文章），可以得到TarsJava里负载均衡的的源码结构图，（基于TarsJava SpringBoot）：
@EnableTarsServer注解：表明这是一个Tars服务；
@Import(TarsServerConfiguration.class)：引入Tars服务相关配置文件；
getServantProxyFactory()：获取代理工厂管理者；
getObjectProxyFactory()：获取对象代理工厂；
select()：选择负载均衡调用器（有四种模式可以选择）；
refresh()：更新负载均衡调用器；
invoke()：具体的执行方法；
doInvokeServant()：最底层的执行方法；
invoker：调用器；
createLoadBalance()：创建客户端负载均衡调用器；
createProtocolInvoker()：创建协议调用器；
Communcator：通信器；
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服务相关配置文件；
getServantProxyFactory()：获取代理工厂管理者；
getObjectProxyFactory()：获取对象代理工厂；
Communcator：通信器；
2.1 getObjectProxyFactory()源码分析
protected ObjectProxyFactory getObjectProxyFactory() {
return objectProxyFactory;
}
getObjectProxyFactory()方法返回一个ObjectProxyFactory对象代理工厂，我们点进去看看这个工厂干了什么：
publicObjectProxygetObjectProxy(Classapi, String objName, String setDivision, ServantProxyConfig servantProxyConfig,
LoadBalanceloadBalance, ProtocolInvokerprotocolInvoker) 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(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> activeEp = new Holder<list>(new ArrayList());
//挂掉的节点
Holder<list> inactiveEp = new Holder<list>(new ArrayList());
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
3. Subset应该添加在哪
Subset应该在节点列表格式化之前。
3.1 获取服务端节点的源码结构图
通过上述分析，我们可得出获取服务端节点getServerNodes()的源码结构图：
@EnableTarsServer注解：表明这是一个Tars服务；
@Import(TarsServerConfiguration.class)：引入Tars服务相关配置文件；
getServantProxyFactory()：获取代理工厂管理者；
getObjectProxyFactory()：获取对象代理工厂；
getServerNodes()：获取服务节点列表；
updateServantEndpoints()： 更新服务端节点；
Communcator：通信器；
3.2 修改getServerNodes()方法
由上述分析，我们可以知道：getServerNodes()的处理逻辑是：
首先创建两个Holder对象；
接着获取到存活与不存活节点列表的值封装进
activeEp与inactiveEp里；将获取到的节点列表格式化为一个字符串格式
value；进行后续格式化操作；
我们要在数据格式化前将列表里的节点进行过滤，不然如果先格式化字符串再过滤，将会涉及字符串的操作，当服务的节点过多是，这部分字符串的校验与判断将会十分消耗性能，因此要在格式化前通过Subset规则过滤节点，修改后的getServerNodes()方法如下：
public String getServerNodes(ServantProxyConfig config) {
QueryFPrx queryProxy = getPrx();
String name = config.getSimpleObjectName();
//存活的节点
Holder<list> activeEp = new Holder<list>(new ArrayList());
//挂掉的节点
Holder<list> inactiveEp = new Holder<list>(new ArrayList());
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> activeEp, ServantProxyConfig config){
StringBuilder value = new StringBuilder();
//config的非空判断
if(config == null){
return null;
}
String ruleType = config.getRuleType();
MapruleData = 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.Entryentry : 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服务相关配置文件；
getServantProxyFactory()：获取代理工厂管理者；
getObjectProxyFactory()：获取对象代理工厂；
getServerNodes()：获取服务节点列表；
updateServantEndpoints()： 更新服务端节点；
Communcator：通信器；
4.3 核心在filterEndpointsBySubset()方法
该方法的主要作用为根据Subset规则过滤节点，并且进行节点列表的格式化操作。