Dubbo 路由及负载均衡性能优化

public class RouterChain<T> {
 // 注册中心最后一次推送的服务列表 private List<Invoker<T>> invokers = Collections.emptyList();
 // 所有路由,包括原生Dubbo基于注册中心的路由规则如“route://” urls . private volatile List<Router> routers = Collections.emptyList();
 // 初始化自动激活的路由 private List<Router> builtinRouters = Collections.emptyList();  private RouterChain(URL url) { //通过ExtensionLoader加载可自动激活的RouterFactory List<RouterFactory> extensionFactories = ExtensionLoader.getExtensionLoader(RouterFactory.class) .getActivateExtension(url, ROUTER_KEY); // 由工厂类生成自动激活的路由策略 List<Router> routers = extensionFactories.stream() .map(factory -> factory.getRouter(url)) .collect(Collectors.toList());
 initWithRouters(routers); }  // 添加额外路由 public void addRouters(List<Router> routers) { List<Router> newRouters = new ArrayList<>(); newRouters.addAll(builtinRouters); newRouters.addAll(routers); Collections.sort(newRouters, comparator); this.routers = newRouters; }  public List<Invoker<T>> route(URL url, Invocation invocation) { List<Invoker<T>> finalInvokers = invokers; // 遍历全部的Router对象，执行路由规则 for (Router router : routers) { finalInvokers = router.route(finalInvokers, url, invocation); } return finalInvokers; }}

@SPIpublic interface RouterFactory {
 @Adaptive("protocol") Router getRouter(URL url);}

public interface Router extends Comparable<Router> {
  /** * * @param invokers 带过滤实例列表 * @param url 消费方url * @param invocation 会话信息 * @return routed invokers * @throws RpcException */ <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;

 /** * 当注册中心的服务列表发现变化，或有动态配置变更会触发实例信息的变化 * 当时2.7.x的Dubbo并没有真正使用这个方法，可基于此方法进行路由缓存 * @param invokers invoker list * @param <T> invoker's type */ default <T> void notify(List<Invoker<T>> invokers) {
 }
}

public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL consumerUrl, Invocation invocation) throws RpcException { if (!this.enabled) { return invokers; }
  // 获取本地机房信息 String local = getSystemProperty(LOC); if (invokers == null || invokers.size() == 0) { return invokers; } List<Invoker<T>> result = new ArrayList<Invoker<T>>(); for (Invoker invoker: invokers) { // 获取与本地机房一致的invoker并加入列表中 String invokerLoc = getProperty(invoker, invocation, LOC); if (local.equals(invokerLoc)) { result.add(invoker); } }
 if (result.size() > 0) { if (fallback){ // 开启服务降级，available.ratio = 当前机房可用服务节点数量 ／ 集群可用服务节点数量 int curAvailableRatio = (int) Math.floor(result.size() * 100.0d / invokers.size()); if (curAvailableRatio <= availableRatio) { return invokers; } }
 return result; } else if (force) { return result; } else { return invokers; }
 }

@SPI(RandomLoadBalance.NAME)public interface LoadBalance {
 /** * 从服务列表中筛选一个. * * @param invokers invokers. * @param url refer url * @param invocation invocation. * @return selected invoker. */ @Adaptive("loadbalance") <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}

 // 预热过程权重计算 static int calculateWarmupWeight(int uptime, int warmup, int weight) { int ww = (int) (uptime / ((float) warmup / weight)); return ww < 1 ? 1 : (Math.min(ww, weight)); }
int getWeight(Invoker<?> invoker, Invocation invocation) { int weight; URL url = invoker.getUrl(); // 多注册中心场景下的，注册中心权重获取 if (UrlUtils.isRegistryService(url)) { weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT); } else { weight = url.getMethodParameter(invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT); if (weight > 0) { // 获取实例启动时间 long timestamp = invoker.getUrl().getParameter(TIMESTAMP_KEY, 0L); if (timestamp > 0L) { long uptime = System.currentTimeMillis() - timestamp; if (uptime < 0) { return 1; } // 获取预热时间 int warmup = invoker.getUrl().getParameter(WARMUP_KEY, DEFAULT_WARMUP); if (uptime > 0 && uptime < warmup) { weight = calculateWarmupWeight((int)uptime, warmup, weight); } } } } return Math.max(weight, 0); }
@Override protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { // Number of invokers int length = invokers.size(); // Every invoker has the same weight? boolean sameWeight = true; // the weight of every invokers int[] weights = new int[length]; // the first invoker's weight int firstWeight = getWeight(invokers.get(0), invocation); weights[0] = firstWeight; // The sum of weights int totalWeight = firstWeight; for (int i = 1; i < length; i++) { int weight = getWeight(invokers.get(i), invocation); // save for later use weights[i] = weight; // Sum totalWeight += weight; if (sameWeight && weight != firstWeight) { sameWeight = false; } } if (totalWeight > 0 && !sameWeight) { // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight. int offset = ThreadLocalRandom.current().nextInt(totalWeight); // Return a invoker based on the random value. for (int i = 0; i < length; i++) { offset -= weights[i]; if (offset < 0) { return invokers.get(i); } } } // If all invokers have the same weight value or totalWeight=0, return evenly. return invokers.get(ThreadLocalRandom.current().nextInt(length)); }

dubbo.consumer.router=-tag

@DubboReference(parameters = {"router","-tag"})

<dubbo:reference id="demoService" check="false" interface="com.dubbo.study.n.api.DemoService" router="-tag" />

 /** * Notify router chain of the initial addresses from registry at the first time. * Notify whenever addresses in registry change. */ public void setInvokers(List<Invoker<T>> invokers) { // 创建带epoch的BitList this.invokers = new BitList<Invoker<T>>(invokers == null ? Collections.emptyList() : invokers,createBitListEpoch()); routers.forEach(router -> router.notify(this.invokers)); }

public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL consumerUrl, Invocation invocation) throws RpcException { …………//省略非核心代码 BitList<Invoker<T>> bitList = (BitList<Invoker<T>>) invokers; //获取路由结果 BitList<Invoker<T>> result = getNearestInvokersWithCache(bitList); if (result.size() > 0) { if (fallback) { // 开启服务降级，available.ratio = 当前机房可用服务节点数量 ／ 集群可用服务节点数量 int curAvailableRatio = (int) Math.floor(result.size() * 100.0d / invokers.size()); if (curAvailableRatio <= availableRatio) { return invokers; } }
 return result; } else if (force) { return result; } else { return invokers; }
 }  /** * 获取缓存列表 * @param invokers * @param <T> * @return */ private <T> BitList<Invoker<T>> getNearestInvokersWithCache(BitList<Invoker<T>> invokers) { ValueWrapper valueWrapper = getCache(getSystemProperty(LOC)); // 是否存在缓存 if (valueWrapper != null) { BitList<Invoker<T>> invokerBitList = (BitList<Invoker<T>>) valueWrapper.get(); // 缓存的epoch与源列表是否一致 if (invokers.isSameEpoch(invokerBitList)) { BitList<Invoker<T>> tmp = invokers.clone(); // 结果取交集 return tmp.and(invokerBitList); } } // 缓存不存在 实时计算放回 return getNearestInvokers(invokers); }  /** * 新服务列表通知 * @param invokers * @param <T> */ @Override public <T> void notify(List<Invoker<T>> invokers) { clear(); if (invokers != null && invokers instanceof BitList) { BitList<Invoker<T>> bitList = (BitList<Invoker<T>>) invokers; // 设置最后一次更新的服务列表 lastNotify = bitList.clone(); if (!CollectionUtils.isEmpty(invokers) && this.enabled) { // 获取机房相同的服务列表并进行缓存 setCache(getSystemProperty(LOC), getNearestInvokers(lastNotify)); } } }

优化前：//这里主要要用多注册中心场景下，注册中心权重的获取，绝大多数情况下并不会有这个逻辑 if (UrlUtils.isRegistryService(url)) { weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT); } 优化后： if (invoker instanceof ClusterInvoker && UrlUtils.isRegistryService(url)) { weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT); }

 /** *  * @param invokers 待分组实例列表 * @param groupNum 分组数量 * @param <T> * @return */ public <T> List<Invoker<T>> doGroup(List<Invoker<T>> invokers, int groupNum) { int listLength = invokers.size() / groupNum; List<Invoker<T>> result = new ArrayList<>(listLength); int random = ThreadLocalRandom.current().nextInt(groupNum); for (int i = random; i < invokers.size(); i = i + groupNum) { result.add(invokers.get(i)); } return result; }