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

揭秘vivo互联网海量服务的研发之路 报名中👇

（1）RouterChain

RouterChain是路由链的入口，其核心字段有

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;
    }
}

@SPI
public 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));
    }

路由的主要消耗是在route方法：

同机房优先路由

这些方法都有一个特点，那就是遍历执行。如负载均衡，针对每一个invoker都需要通过getWeight方法进行权重的计算；就近路由的router方法对于每一个invoker都需要通过url获取及机房信息进行匹配计算。

我们分析一下getWeight及router时间复杂度，发现是O(n)的时间复杂度，而且路由是由路由链组成的，每次每个 Router的route方法调用逻辑都会遍历实例列表，那么当实例列表数量过大时，每次匹配的计算的逻辑过大，那么就会造成大量的计算成本，导致占用大量cpu，同时也导致路由负载均衡效率低下。

综上所述，罪恶的的根源就是遍历导致的，当服务提供方数量越多，影响越大。

dubbo.consumer.router=-tag

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

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

每次路由目前都是进行实时计算，但是在大多数情况下，我们的实例列表是稳定不变的，只有在发布窗口或配置变更窗口内实例列表才会发生变更，那我们是否可以考虑缓存呢。如就近路由，可以以机房为key进行机房实例的全量缓存。针对接口级标签路由可以缓存不同标签值指定的实例信息。

我们知道路由的执行过程是责任链模式，每一个Router的实例列表入参实际上是一个Router的结果，可参考公式：target = rn(…r3(r2(r1(src))))。那么所有的路由可以基于注册中心推送的原始服务列表进行路由计算并缓存，然后不同的路由结果相互取交集就能得到最终的结果，当实例信息发生变更时，缓存失效并重新计算。

进入具体路由方法时，先判断是否存在缓存的路由值，且缓存值的epoch必须与上一个路由的epoch需一致，此时缓存才生效，然后缓存值与上个Router的结果取交集。

如果不存在缓存或epoch不一致则重新进行实时的路由计算。

上文我们说到，不同的路由策略之间的结果是取交集的，然后最终的结果才送入负载均衡流程。那如何在缓存的同时，加快交集的计算呢。答案就是基于位图：BitMap。

BitMap的基本原理就是用一个bit位来存放某种状态，适用于大规模数据的查找及位运算操作。如在路由场景，先基于全量的推送数据进行计算缓存。如果某个实例被路由选中，则其值为1，若两个路由的结果要取交集，那直接对BitMap进行"&"运行即可。

全量缓存示意图：

 /**
     * 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);
   }

遍历是罪恶的源泉，而实例的数量决定这罪恶的深浅，我们有什么办法减少负载均衡过程中的遍历呢。一是根据group及version划分不同的集群，但是这需要涉及到业务方代码或配置层面的改动，会带来额外的成本。所以我们放弃了。

二是没有什么是加一层解决不了的问题，为了尽量减少进入负载均衡的节点数量，考虑新增一个垫底的路由策略，在走完所有的路由策略后，若节点数量>自定义数量后，进行虚拟分组，虚拟分组的策略也可进行自定义，然后随机筛选一组进入负载均衡。此时进入负载均衡的实例数量就会有倍数的下降。

需要注意的是分组路由必须保证是在路由链的最后一环，否则会导致其他路由计算错误。

  /**
     * 
     * @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;
    }

参考资料：