dubbo源碼分析-Directory 和 LoadBalance-筆記

Directory

• 訂閱節點的變化，
  1. 當zookeeper上指定節點發生變化之後，會通知到RegistryDirectory的notify方法
  2. 將url轉化爲invoker對象

調用過程當中invokers的使用

• StaticDirectory: 靜態目錄服務，
  • 它的全部Invoker經過構造函數傳入，
  • 服務消費方引用服務的時候， 服務對多註冊中心的引用，將Invokers集合直接傳入 StaticDirectory構造器，再由Cluster假裝成一個Invoker
  • StaticDirectory的list方法直接返回全部invoker集合；
• RegistryDirectory: 註冊目錄服務，
  • 它的Invoker集合是從註冊中心獲取的，
  • 它實現了NotifyListener接口實現了回調接口notify(List<Url>)

Directory目錄服務的更新過程

• RegistryProtocol.doRefer方法，也就是消費端在初始化的時候，這裏涉及到了RegistryDirectory這個類。而後執行cluster.join(directory)方法。
• 這些代碼在上節課有分析過。
• cluster.join其實就是將Directory中的多個Invoker假裝成一個Invoker, 對上層透明，包含集羣的容錯機制

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);//對多個invoker進行組裝
    directory.setRegistry(registry); //ZookeeperRegistry
    directory.setProtocol(protocol); //protocol=Protocol$Adaptive
    //url=consumer://192.168.111....
    URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
    //會把consumer://192...  註冊到註冊中心
    if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
            && url.getParameter(Constants.REGISTER_KEY, true)) {
        //zkClient.create()
        registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                Constants.CHECK_KEY, String.valueOf(false)));
    }
    directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
            Constants.PROVIDERS_CATEGORY 
            + "," + Constants.CONFIGURATORS_CATEGORY 
            + "," + Constants.ROUTERS_CATEGORY));
    //Cluster$Adaptive
    return cluster.join(directory);
}

directory.subscribe

訂閱節點的變化，

1. 當zookeeper上指定節點發生變化之後，會通知到RegistryDirectory的notify方法
2. 將url轉化爲invoker對象

調用過程當中invokers的使用

• 再調用過程當中，AbstractClusterInvoker.invoke方法中

public Result invoke(final Invocation invocation) throws RpcException {

    checkWhetherDestroyed();

    LoadBalance loadbalance;
    
    List<Invoker<T>> invokers = list(invocation); 
    if (invokers != null && invokers.size() > 0) {
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                .getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
    } else {
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
    }
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    return doInvoke(invocation, invokers, loadbalance);
}

list方法

• 從directory中得到invokers

protected  List<Invoker<T>> list(Invocation invocation) throws RpcException {
   List<Invoker<T>> invokers = directory.list(invocation);
   return invokers;
}

負載均衡LoadBalance

• LoadBalance負載均衡， 負責從多個 Invokers中選出具體的一個Invoker用於本次調用，調用過程當中包含了負載均衡的算法。

負載均衡代碼訪問入口

• 在AbstractClusterInvoker.invoke中代碼以下，經過名稱得到指定的擴展點。RandomLoadBalance

public Result invoke(final Invocation invocation) throws RpcException {

    checkWhetherDestroyed();

    LoadBalance loadbalance;
    
    List<Invoker<T>> invokers = list(invocation);
    if (invokers != null && invokers.size() > 0) {
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                .getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
    } else {
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
    }
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    return doInvoke(invocation, invokers, loadbalance);
}

AbstractClusterInvoker.doselect

• 調用LoadBalance.select方法，將 invokers 按照指定算法進行負載

private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
    if (invokers == null || invokers.size() == 0)
        return null;
    if (invokers.size() == 1)
        return invokers.get(0);
    // 若是隻有兩個invoker，退化成輪循
    if (invokers.size() == 2 && selected != null && selected.size() > 0) {
        return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0);
    }
    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
    
    //若是 selected中包含（優先判斷） 或者 不可用&&availablecheck=true 則重試.
    if( (selected != null && selected.contains(invoker))
            ||(!invoker.isAvailable() && getUrl()!=null && availablecheck)){
        try{
            Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
            if(rinvoker != null){
                invoker =  rinvoker;
            }else{
                //看下第一次選的位置，若是不是最後，選+1位置.
                int index = invokers.indexOf(invoker);
                try{
                    //最後在避免碰撞
                    invoker = index <invokers.size()-1?invokers.get(index+1) :invoker;
                }catch (Exception e) {
                    logger.warn(e.getMessage()+" may because invokers list dynamic change, ignore.",e);
                }
            }
        }catch (Throwable t){
            logger.error("clustor relselect fail reason is :"+t.getMessage() +" if can not slove ,you can set cluster.availablecheck=false in url",t);
        }
    }
    return invoker;
}

• 默認狀況下，LoadBalance使用的是Random算法，可是這個隨機和咱們理解上的隨機仍是不同的,由於他還有個概念叫weight(權重)

RandomLoadBalance

• 假設有四個集羣節點A,B,C,D,對應的權重分別是1,2,3,4,
• 那麼請求到A節點的機率就爲1/(1+2+3+4) = 10%.B,C,D節點依次類推爲20%,30%,40%.

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    int length = invokers.size(); // 總個數
    int totalWeight = 0; // 總權重
    boolean sameWeight = true; // 權重是否都同樣
    for (int i = 0; i < length; i++) {
        int weight = getWeight(invokers.get(i), invocation);
        totalWeight += weight; // 累計總權重
        if (sameWeight && i > 0
                && weight != getWeight(invokers.get(i - 1), invocation)) {
            sameWeight = false; // 計算全部權重是否同樣
        }
    }
    if (totalWeight > 0 && ! sameWeight) {
        // 若是權重不相同且權重大於0則按總權重數隨機
        int offset = random.nextInt(totalWeight);
        // 並肯定隨機值落在哪一個片段上
        for (int i = 0; i < length; i++) {
            offset -= getWeight(invokers.get(i), invocation);
            if (offset < 0) {
                return invokers.get(i);
            }
        }
    }
    // 若是權重相同或權重爲0則均等隨機
    return invokers.get(random.nextInt(length));
}