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

public class RouterChainT> {
    // 注册中心最后一次推送的服务列表    private List> invokers = Collections.emptyList();
    // 所有路由,包括原生Dubbo基于注册中心的路由规则如“route://” urls .    private volatile List routers = Collections.emptyList();
    // 初始化自动激活的路由    private List builtinRouters = Collections.emptyList();     private RouterChain(URL url) {    //通过ExtensionLoader加载可自动激活的RouterFactory        List extensionFactories = ExtensionLoader.getExtensionLoader(RouterFactory.class)                .getActivateExtension(url, ROUTER_KEY);    // 由工厂类生成自动激活的路由策略        List routers = extensionFactories.stream()                .map(factory -> factory.getRouter(url))                .collect(Collectors.toList());
        initWithRouters(routers);    }      // 添加额外路由    public void addRouters(List routers) {        List newRouters = new ArrayList();        newRouters.addAll(builtinRouters);        newRouters.addAll(routers);        Collections.sort(newRouters, comparator);        this.routers = newRouters;    }     public List> route(URL url, Invocation invocation) {        List> 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 ComparableRouter> {
       /**     *     * @param invokers   带过滤实例列表     * @param url        消费方url     * @param invocation 会话信息     * @return routed invokers     * @throws RpcException     */     List> route(List> invokers, URL url, Invocation invocation) throws RpcException;

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

public  List> route(List> 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> result = new ArrayList>();        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                     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")     Invoker select(List> 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                        weight = calculateWarmupWeight((int)uptime, warmup, weight);                   }               }           }       }       return Math.max(weight, 0);   }
@Override    protected  Invoker doSelect(List> 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             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                 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"})

"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> invokers) {    // 创建带epoch的BitList        this.invokers = new BitList>(invokers == null ? Collections.emptyList() : invokers,createBitListEpoch());        routers.forEach(router -> router.notify(this.invokers));    }

public  List> route(List> invokers, URL consumerUrl, Invocation invocation) throws RpcException {        …………//省略非核心代码        BitList> bitList = (BitList>) invokers;    //获取路由结果        BitList> result = getNearestInvokersWithCache(bitList);        if (result.size() > 0) {            if (fallback) {                // 开启服务降级，available.ratio = 当前机房可用服务节点数量 ／ 集群可用服务节点数量                int curAvailableRatio = (int) Math.floor(result.size() * 100.0d / invokers.size());                if (curAvailableRatio                     return invokers;                }            }
            return result;        } else if (force) {            return result;        } else {            return invokers;        }
    }      /**     * 获取缓存列表     * @param invokers     * @param      * @return     */    private  BitList> getNearestInvokersWithCache(BitList> invokers) {        ValueWrapper valueWrapper = get服务器托管网Cache(getSystemProperty(LOC));        // 是否存在缓存        if (valueWrapper != null) {            BitList> invokerBitList = (BitList>) valueWrapper.get();            // 缓存的epoch与源列表是否一致            if (invokers.isSameEpoch(invokerBitList)) {                BitList> tmp = invokers.clone();                // 结果取交集                return tmp.and(invokerBitList);            }        }        // 缓存不存在 实时计算放回        return getNearestInvokers(invokers);    }    /**     * 新服务列表通知     * @param invokers     * @param      */ @Override    public  void notify(List> invokers) {        clear();        if (invokers != null && invokers instanceof BitList) {            BitList> bitList = (BitList>) 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      * @return     */    public  List> doGroup(List> invokers, int groupNum) {        int listLength = invokers.size() / groupNum;        List> result = new ArrayList(listLength);        int random = ThreadLocalRandom.current().nextInt(groupNum);        for (int i = random; i             result.add(invokers.get(i));        }        return result;    }