Dubbo源碼分析—消費者引用服務

瞭解服務引用

消費者引用一個服務的主過程，以下圖所示：

首先ReferenceConfig類的init方法調用Protocol的refer方法生成Invoker 實例(如上圖中的紅色部分)，這是服務消費的關鍵。接下來把Invoker轉換爲客戶端須要的接口(如：HelloWorld)。

源碼分析

入口分析——ReferenceBean

ReferenceBean類實現了FactoryBean接口，經過getBean方法返回的不是FactoryBean自己，而是FactoryBean#getObject方法所返回的對象。

getObject方法返回ReferenceBean類的實例變量ref，其源碼以下：

public Object getObject() throws Exception {
    return get();
}

public synchronized T get() {
    if (destroyed){
        throw new IllegalStateException("Already destroyed!");
    }
    if (ref == null) {
        init();
    }
    return ref;
}
複製代碼

init方法對ref變量進行賦值，源碼以下：

ref = createProxy(map);
複製代碼

接下來咱們一塊兒看看createProxy方法，源碼以下：

private T createProxy(Map<String, String> map) {
    // 一、根據referenceBean屬性構建tmpUrl
    URL tmpUrl = new URL("temp", "localhost", 0, map);

    final boolean isJvmRefer;
    if (isInjvm() == null) {
        if (url != null && url.length() > 0) { //指定URL的狀況下，不作本地引用
            isJvmRefer = false;
        } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
            //默認狀況下若是本地有服務暴露，則引用本地服務.
            isJvmRefer = true;
        } else {
            isJvmRefer = false;
        }
    } else {
        isJvmRefer = isInjvm().booleanValue();
    }

    // 本地引用
    if (isJvmRefer) {
        URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
        invoker = refprotocol.refer(interfaceClass, url);
        if (logger.isInfoEnabled()) {
            logger.info("Using injvm service " + interfaceClass.getName());
        }
    } else {
        // 用戶指定URL，指定的URL多是對點對直連地址，也多是註冊中心URL
        if (url != null && url.length() > 0) { 
            String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
            if (us != null && us.length > 0) {
                for (String u : us) {
                    URL url = URL.valueOf(u);
                    if (url.getPath() == null || url.getPath().length() == 0) {
                        url = url.setPath(interfaceName);
                    }
                    if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                        urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                    } else {
                        urls.add(ClusterUtils.mergeUrl(url, map));
                    }
                }
            }
        } else { 
            // 經過註冊中心配置拼裝URL
            List<URL> us = loadRegistries(false);
            if (us != null && us.size() > 0) {
                for (URL u : us) {
                    URL monitorUrl = loadMonitor(u);
                    if (monitorUrl != null) {
                        map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                    }
                    urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                }
            }
        }
        // 二、調用procotol獲取invoker對象
        if (urls.size() == 1) {  // 只存在一個註冊中心
            invoker = refprotocol.refer(interfaceClass, urls.get(0));
        } else {
            List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
            URL registryURL = null;
            for (URL url : urls) {
                invokers.add(refprotocol.refer(interfaceClass, url));
                if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { // 用了最後一個registry url
                    registryURL = url; 
                }
            }
            if (registryURL != null) { // 有 註冊中心協議的URL
                // 對有註冊中心的Cluster 只用 AvailableCluster
                URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME); 
                invoker = cluster.join(new StaticDirectory(u, invokers));
            }  else { // 不是 註冊中心的URL
                invoker = cluster.join(new StaticDirectory(invokers));
            }
        }
    }

    // 三、建立invoker的代理對象
    return (T) proxyFactory.getProxy(invoker);
}
複製代碼

createProxy方法的過程以下：

一、根據ReferenceBean的屬性拼接URL；

二、調用refprotocol.refer()方法獲取invoker對象；

三、建立invoker對象的代理對象。

獲取invoker對象的過程

refprotocol.refer方法的時序圖以下：

refprotocol.refer方法主要邏輯由doRefer方法完成的，向註冊中心註冊消費者的url，而後訂閱引用接口的信息，最後經過cluster.join(directory)獲取invoke對象（這個過程比較簡單，直接構造invoker對象）。doRefer方法源碼以下：

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
    if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
            && url.getParameter(Constants.REGISTER_KEY, true)) {
        // 向註冊中心註冊消費者的url 
        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));
    return cluster.join(directory);
}
複製代碼

註冊消費者url和發佈服務的註冊過程相似，都是建立zookeeper臨時節點。咱們重點關注其訂閱過程：

// this = doRefer()中directory
registry.subscribe(url, this);            
複製代碼

經過debug發現會訂閱下面三個zookeeper目錄：

/dubbo/com.alibaba.dubbo.demo.DemoService/providers /dubbo/com.alibaba.dubbo.demo.DemoService/configurators /dubbo/com.alibaba.dubbo.demo.DemoService/routers

當這三個目錄有子節點變化時，都會出發點RegisterDirectory的notity方法。源碼以下：

public synchronized void notify(List<URL> urls) {
    List<URL> invokerUrls = new ArrayList<URL>();
    List<URL> routerUrls = new ArrayList<URL>();
    List<URL> configuratorUrls = new ArrayList<URL>();
    for (URL url : urls) {
        String protocol = url.getProtocol();
        String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
        if (Constants.ROUTERS_CATEGORY.equals(category) 
                || Constants.ROUTE_PROTOCOL.equals(protocol)) {
            routerUrls.add(url);
        } else if (Constants.CONFIGURATORS_CATEGORY.equals(category) 
                || Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
            configuratorUrls.add(url);
        } else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
            invokerUrls.add(url);
        } else {
            logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
        }
    }
    // configurators 
    if (configuratorUrls != null && configuratorUrls.size() >0 ){
        this.configurators = toConfigurators(configuratorUrls);
    }
    // routers
    if (routerUrls != null && routerUrls.size() >0 ){
        List<Router> routers = toRouters(routerUrls);
        if(routers != null){ // null - do nothing
            setRouters(routers);
        }
    }
    List<Configurator> localConfigurators = this.configurators; // local reference
    // 合併override參數
    this.overrideDirectoryUrl = directoryUrl;
    if (localConfigurators != null && localConfigurators.size() > 0) {
        for (Configurator configurator : localConfigurators) {
            this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
        }
    }
    // providers
    refreshInvoker(invokerUrls);
}
複製代碼

建立invoker的代理對象

建立invoker的代理對象由JavassistProxyFactory.getProxy方法完成的，源碼以下：

public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
    return (T) Proxy.getProxy(interfaces)
        .newInstance(new InvokerInvocationHandler(invoker));
}
複製代碼

生成的代理對象其源碼以下：

public class proxy0 implements DC, EchoService, DemoService {
    public static Method[] methods;
    private InvocationHandler handler;

    public String sayHello(String var1) {
        Object[] var2 = new Object[]{var1};
        Object var3 = this.handler.invoke(this, methods[0], var2);
        return (String)var3;
    }

    public Object $echo(Object var1) {
        Object[] var2 = new Object[]{var1};
        Object var3 = this.handler.invoke(this, methods[1], var2);
        return (Object)var3;
    }

    public proxy0() {
    }

    public proxy0(InvocationHandler var1) {
        this.handler = var1;
    }
}
複製代碼