Dubbo源码解析之客户端初始化及服务调用

一时失言乱红尘 2022-04-22 11:02 157阅读 0赞

# 准备 #

`dubbo` 版本：2.5.4

# 客户端初始化过程 #

## 初始化过程 ##

先上时序图，帮助理解客户端初始化过程。  
![Dubbo客户端初始化][Dubbo]

`ReferenceBean` 是客户端初始化入口，其实现 `InitializingBean` 接口，在 `bean` 初始化过程中会调用其 `afterPropertiesSet` 方法，进而调用 `getObject()` \-> `get()` \-> `init()` ，之后再调用 `ReferenceConfig` 的 `createProxy()` 方法。

ReferenceConfig

private T createProxy(Map<String, String> map) { 
        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
            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); // 从注册中心上获得相应的协议url地址
                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)));
                    }
                }
                if (urls == null || urls.size() == 0) { 
                    throw new IllegalStateException("No such any registry to reference " + interfaceName  + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");
                }
            }
    
            if (urls.size() == 1) { 
                // refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
                // refprotocol -> Protocol$Adaptive
                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())) { 
                        registryURL = url; // 用了最后一个registry url
                    }
                }
                // 注册中心地址URL不为空
                if (registryURL != null) {  
                    // 对有注册中心的Cluster只用AvailableCluster
                    URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
                    // 返回的是MockClusterInvoker(FailoverClusterInvoker)
                    invoker = cluster.join(new StaticDirectory(u, invokers));
                }  else { 
                    invoker = cluster.join(new StaticDirectory(invokers));
                }
            }
        }
    
        Boolean c = check;
        if (c == null && consumer != null) { 
            c = consumer.isCheck();
        }
        if (c == null) { 
            c = true; // default true
        }
        if (c && ! invoker.isAvailable()) { 
            throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
        }
        if (logger.isInfoEnabled()) { 
            logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
        }
        // 创建服务代理
        return (T) proxyFactory.getProxy(invoker);
    }

Protocol$Adpative

public class Protocol$Adpative implements com.alibaba.dubbo.rpc.Protocol { 
        public void destroy() { 
            throw new UnsupportedOperationException("method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
        }
    
        public int getDefaultPort() { 
            throw new UnsupportedOperationException("method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
        }
    
        public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException { 
            if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
            if (arg0.getUrl() == null)
                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");
            com.alibaba.dubbo.common.URL url = arg0.getUrl();
            String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
            if (extName == null)
                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
            com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
            return extension.export(arg0);
        }
    
        public com.alibaba.dubbo.rpc.Invoker refer(Class arg0, com.alibaba.dubbo.common.URL arg1) throws com.alibaba.dubbo.rpc.RpcException { 
            if (arg1 == null) throw new IllegalArgumentException("url == null");
            com.alibaba.dubbo.common.URL url = arg1;
            String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
            if (extName == null)
                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
            // extName -> registry
            // extension -> RegistryProtocol
            com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
            return extension.refer(arg0, arg1);
        }
    }

RegistryProtocol

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { 
        url = url.setProtocol(url.getParameter("registry", "dubbo")).removeParameter("registry");
        Registry registry = this.registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) { 
            return this.proxyFactory.getInvoker(registry, type, url);
        } else { 
            Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded("refer"));
            String group = (String)qs.get("group");
            return group == null || group.length() <= 0 || Constants.COMMA_SPLIT_PATTERN.split(group).length <= 1 && !"*".equals(group) ? this.doRefer(this.cluster, registry, type, url) : this.doRefer(this.getMergeableCluster(), registry, type, url);
        }
    }

`RegistryProtocol.doRefer`方法存在一个参数 `Cluster` ，而 `Cluster` 是一个扩展点，存在加在方法级别上的 `@Adaptive` 注解，说明会动态生成自适应适配器( `Cluster$Adaptive` )。在 `RegistryProtocol` 中存在 `Cluster` 扩展点成员变量及 `setter` 方法，说明是一个自动注入的扩展点。

@SPI("failover")
    public interface Cluster { 
        @Adaptive
        <T> Invoker<T> join(Directory<T> var1) throws RpcException;
    }

// RegistryProtocol
    private Cluster cluster;
    
    public void setCluster(Cluster cluster) { 
        this.cluster = cluster;
    }

Cluster$Adpative

public class Cluster$Adpative implements com.alibaba.dubbo.rpc.cluster.Cluster { 
        
        public com.alibaba.dubbo.rpc.Invoker join(com.alibaba.dubbo.rpc.cluster.Directory arg0) throws com.alibaba.dubbo.rpc.RpcException { 
            if (arg0 == null)
                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument == null");
            if (arg0.getUrl() == null)
                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument getUrl() == null");
            com.alibaba.dubbo.common.URL url = arg0.getUrl();
            String extName = url.getParameter("cluster", "failover");
            if (extName == null)
                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.cluster.Cluster) name from url(" + url.toString() + ") use keys([cluster])");
            com.alibaba.dubbo.rpc.cluster.Cluster extension = (com.alibaba.dubbo.rpc.cluster.Cluster) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.cluster.Cluster.class).getExtension(extName);
            return extension.join(arg0);
        }
    }

又因为 `Cluster` 扩展点实现中存在以扩展点作为参数的构造方法，所以会被 `Wrapper` 装饰，而该装饰器就是 `MockClusterWrapper` 。

public class MockClusterWrapper implements Cluster { 
        private Cluster cluster;
    
        public MockClusterWrapper(Cluster cluster) { 
            this.cluster = cluster;
        }
    
        public <T> Invoker<T> join(Directory<T> directory) throws RpcException { 
            return new MockClusterInvoker(directory, this.cluster.join(directory));
        }
    }

// RegistryProtocol
    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { 
        RegistryDirectory<T> directory = new RegistryDirectory(type, url);
        directory.setRegistry(registry); // registry -> ZookeeperRegistry
        directory.setProtocol(this.protocol); // protocol -> Protocol$Adaptive
        URL subscribeUrl = new URL("consumer", NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
        if (!"*".equals(url.getServiceInterface()) && url.getParameter("register", true)) { 
            // 注册consumer://协议地址到注册中心
            registry.register(subscribeUrl.addParameters(new String[]{ "category", "consumers", "check", String.valueOf(false)}));
        }
    	// 注册zookeeper地址变更
        directory.subscribe(subscribeUrl.addParameter("category", "providers,configurators,routers"));
        // 返回一个MockClusterInvoker(FailoverClusterInvoker)
        return cluster.join(directory);
    }

MockClusterWrapper

public <T> Invoker<T> join(Directory<T> directory) throws RpcException { 
        return new MockClusterInvoker(directory, this.cluster.join(directory));
    }

FailoverCluster

public <T> Invoker<T> join(Directory<T> directory) throws RpcException { 
        return new FailoverClusterInvoker(directory);
    }

回到ReferenceConfig.createProxy中的proxyFactory.getProxy()

private T createProxy(Map<String, String> map) { 
        ......
    	// 创建服务代理
        // proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
    	// proxyFactory -> ProxyFactory$Adpative
    	return (T) proxyFactory.getProxy(invoker);
    }

ProxyFactory$Adpative

public class ProxyFactory$Adpative implements com.alibaba.dubbo.rpc.ProxyFactory { 
        public java.lang.Object getProxy(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException { 
            if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
            if (arg0.getUrl() == null)
                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");
            com.alibaba.dubbo.common.URL url = arg0.getUrl();
            String extName = url.getParameter("proxy", "javassist");
            if (extName == null)
                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
            // extName -> javassist
            // extension -> StubProxyFactoryWrapper(JavassistProxyFactory)
            com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
            return extension.getProxy(arg0);
        }
    	......
    }

StubProxyFactoryWrapper

public <T> T getProxy(Invoker<T> invoker) throws RpcException { 
        T proxy = this.proxyFactory.getProxy(invoker);
        if (GenericService.class != invoker.getInterface()) { 
            String stub = invoker.getUrl().getParameter("stub", invoker.getUrl().getParameter("local"));
            if (ConfigUtils.isNotEmpty(stub)) { 
                Class<?> serviceType = invoker.getInterface();
                if (ConfigUtils.isDefault(stub)) { 
                    if (invoker.getUrl().hasParameter("stub")) { 
                        stub = serviceType.getName() + "Stub";
                    } else { 
                        stub = serviceType.getName() + "Local";
                    }
                }
    
                try { 
                    Class<?> stubClass = ReflectUtils.forName(stub);
                    if (!serviceType.isAssignableFrom(stubClass)) { 
                        throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + serviceType.getName());
                    }
    
                    try { 
                        Constructor<?> constructor = ReflectUtils.findConstructor(stubClass, serviceType);
                        proxy = constructor.newInstance(proxy);
                        URL url = invoker.getUrl();
                        if (url.getParameter("dubbo.stub.event", false)) { 
                            url = url.addParameter("dubbo.stub.event.methods", StringUtils.join(Wrapper.getWrapper(proxy.getClass()).getDeclaredMethodNames(), ","));
                            url = url.addParameter("isserver", Boolean.FALSE.toString());
    
                            try { 
                                this.export(proxy, invoker.getInterface(), url);
                            } catch (Exception var9) { 
                                LOGGER.error("export a stub service error.", var9);
                            }
                        }
                    } catch (NoSuchMethodException var10) { 
                        throw new IllegalStateException("No such constructor \"public " + stubClass.getSimpleName() + "(" + serviceType.getName() + ")\" in stub implemention class " + stubClass.getName(), var10);
                    }
                } catch (Throwable var11) { 
                    LOGGER.error("Failed to create stub implemention class " + stub + " in consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", cause: " + var11.getMessage(), var11);
                }
            }
        }
    
        return proxy;
    }

JavassistProxyFactory

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

## 客户端与服务端创建连接过程 ##

客户端与服务端创建连接时序图。  
![Dubbo客户端与服务端创建连接过程][Dubbo 1]

RegistryProtocol

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

RegistryDirectory

public void subscribe(URL url) { 
        setConsumerUrl(url);
        registry.subscribe(url, this);
    }

FailbackRegistry

public void subscribe(URL url, NotifyListener listener) { 
        super.subscribe(url, listener);
        removeFailedSubscribed(url, listener);
        try { 
            // 向服务器端发送订阅请求
            doSubscribe(url, listener);
        } catch (Exception e) { 
            Throwable t = e;
    
            List<URL> urls = getCacheUrls(url);
            if (urls != null && urls.size() > 0) { 
                notify(url, listener, urls);
                logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + getUrl().getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/dubbo-registry-" + url.getHost() + ".cache") + ", cause: " + t.getMessage(), t);
            } else { 
                // 如果开启了启动时检测，则直接抛出异常
                boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                        && url.getParameter(Constants.CHECK_KEY, true);
                boolean skipFailback = t instanceof SkipFailbackWrapperException;
                if (check || skipFailback) { 
                    if(skipFailback) { 
                        t = t.getCause();
                    }
                    throw new IllegalStateException("Failed to subscribe " + url + ", cause: " + t.getMessage(), t);
                } else { 
                    logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
                }
            }
    
            // 将失败的订阅请求记录到失败列表，定时重试
            addFailedSubscribed(url, listener);
        }
    }

ZookeeperRegistry

protected void doSubscribe(final URL url, final NotifyListener listener) { 
        try { 
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) { 
                String root = toRootPath();
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                if (listeners == null) { 
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                    listeners = zkListeners.get(url);
                }
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) { 
                    listeners.putIfAbsent(listener, new ChildListener() { 
                        public void childChanged(String parentPath, List<String> currentChilds) { 
                            for (String child : currentChilds) { 
                child = URL.decode(child);
                                if (! anyServices.contains(child)) { 
                                    anyServices.add(child);
                                    subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child, 
                                            Constants.CHECK_KEY, String.valueOf(false)), listener);
                                }
                            }
                        }
                    });
                    zkListener = listeners.get(listener);
                }
                zkClient.create(root, false);
                List<String> services = zkClient.addChildListener(root, zkListener);
                if (services != null && services.size() > 0) { 
                    for (String service : services) { 
          service = URL.decode(service);
          anyServices.add(service);
                        subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service, 
                                Constants.CHECK_KEY, String.valueOf(false)), listener);
                    }
                }
            } else { 
                List<URL> urls = new ArrayList<URL>();
                for (String path : toCategoriesPath(url)) { 
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    if (listeners == null) { 
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                        listeners = zkListeners.get(url);
                    }
                    ChildListener zkListener = listeners.get(listener);
                    if (zkListener == null) { 
                        listeners.putIfAbsent(listener, new ChildListener() { 
                            public void childChanged(String parentPath, List<String> currentChilds) { 
                               ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                            }
                        });
                        zkListener = listeners.get(listener);
                    }
                    zkClient.create(path, false);
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) { 
                       urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                notify(url, listener, urls);
            }
        } catch (Throwable e) { 
            throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

FailbackRegistry

protected void notify(URL url, NotifyListener listener, List<URL> urls) { 
        if (url == null) { 
            throw new IllegalArgumentException("notify url == null");
        }
        if (listener == null) { 
            throw new IllegalArgumentException("notify listener == null");
        }
        try { 
           doNotify(url, listener, urls);
        } catch (Exception t) { 
            // 将失败的通知请求记录到失败列表，定时重试
            Map<NotifyListener, List<URL>> listeners = failedNotified.get(url);
            if (listeners == null) { 
                failedNotified.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, List<URL>>());
                listeners = failedNotified.get(url);
            }
            listeners.put(listener, urls);
            logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
        }
    }
    
    protected void doNotify(URL url, NotifyListener listener, List<URL> urls) { 
        super.notify(url, listener, urls);
    }

AbstractRegistry

protected void notify(URL url, NotifyListener listener, List<URL> urls) { 
        if (url == null) { 
            throw new IllegalArgumentException("notify url == null");
        }
        if (listener == null) { 
            throw new IllegalArgumentException("notify listener == null");
        }
        if ((urls == null || urls.size() == 0) 
                && ! Constants.ANY_VALUE.equals(url.getServiceInterface())) { 
            logger.warn("Ignore empty notify urls for subscribe url " + url);
            return;
        }
        if (logger.isInfoEnabled()) { 
            logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
        }
        Map<String, List<URL>> result = new HashMap<String, List<URL>>();
        for (URL u : urls) { 
            if (UrlUtils.isMatch(url, u)) { 
               String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
               List<URL> categoryList = result.get(category);
               if (categoryList == null) { 
                  categoryList = new ArrayList<URL>();
                  result.put(category, categoryList);
               }
               categoryList.add(u);
            }
        }
        if (result.size() == 0) { 
            return;
        }
        Map<String, List<URL>> categoryNotified = notified.get(url);
        if (categoryNotified == null) { 
            notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
            categoryNotified = notified.get(url);
        }
        // 对providers、configurators、routers路径下的进行更新
        for (Map.Entry<String, List<URL>> entry : result.entrySet()) { 
            String category = entry.getKey();
            List<URL> categoryList = entry.getValue();
            categoryNotified.put(category, categoryList);
            saveProperties(url);
            listener.notify(categoryList);
        }
    }

RegistryDirectory

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
        // 刷新Invoker>>
        refreshInvoker(invokerUrls);
    }

/** * 根据invokerURL列表转换为invoker列表。转换规则如下： * 1.如果url已经被转换为invoker，则不在重新引用，直接从缓存中获取，注意如果url中任何一个参数变更也会重新引用 * 2.如果传入的invoker列表不为空，则表示最新的invoker列表 * 3.如果传入的invokerUrl列表是空，则表示只是下发的override规则或route规则，需要重新交叉对比，决定是否需要重新引用。 * @param invokerUrls 传入的参数不能为null */
    // RegistryDirectory
    private void refreshInvoker(List<URL> invokerUrls){ 
        if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
                && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) { 
            this.forbidden = true; // 禁止访问
            this.methodInvokerMap = null; // 置空列表
            destroyAllInvokers(); // 关闭所有Invoker
        } else { 
            this.forbidden = false; // 允许访问
            Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
            if (invokerUrls.size() == 0 && this.cachedInvokerUrls != null){ 
                invokerUrls.addAll(this.cachedInvokerUrls);
            } else { 
                this.cachedInvokerUrls = new HashSet<URL>();
                this.cachedInvokerUrls.addAll(invokerUrls);//缓存invokerUrls列表，便于交叉对比
            }
            if (invokerUrls.size() ==0 ){ 
               return;
            }
            Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表
            Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // 换方法名映射Invoker列表
            // state change
            //如果计算错误，则不进行处理.
            if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0 ){ 
                logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :"+invokerUrls.size() + ", invoker.size :0. urls :"+invokerUrls.toString()));
                return ;
            }
            this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
            this.urlInvokerMap = newUrlInvokerMap;
            try{ 
                destroyUnusedInvokers(oldUrlInvokerMap,newUrlInvokerMap); // 关闭未使用的Invoker
            }catch (Exception e) { 
                logger.warn("destroyUnusedInvokers error. ", e);
            }
        }
    }

// RegistryDirectory
    private Map<String, Invoker<T>> toInvokers(List<URL> urls) { 
        Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
        if(urls == null || urls.size() == 0){ 
            return newUrlInvokerMap;
        }
        Set<String> keys = new HashSet<String>();
        String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
        for (URL providerUrl : urls) { 
           //如果reference端配置了protocol，则只选择匹配的protocol
           if (queryProtocols != null && queryProtocols.length() >0) { 
              boolean accept = false;
              String[] acceptProtocols = queryProtocols.split(",");
              for (String acceptProtocol : acceptProtocols) { 
                 if (providerUrl.getProtocol().equals(acceptProtocol)) { 
                    accept = true;
                    break;
                 }
              }
              if (!accept) { 
                 continue;
              }
           }
            if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) { 
                continue;
            }
            if (! ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) { 
                logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() + " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost() 
                        + ", supported protocol: "+ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
                continue;
            }
            URL url = mergeUrl(providerUrl);
            
            String key = url.toFullString(); // URL参数是排序的
            if (keys.contains(key)) {  // 重复URL
                continue;
            }
            keys.add(key);
            // 缓存key为没有合并消费端参数的URL，不管消费端如何合并参数，如果服务端URL发生变化，则重新refer
            Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
            Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
            if (invoker == null) {  // 缓存中没有，重新refer
                try { 
                   boolean enabled = true;
                   if (url.hasParameter(Constants.DISABLED_KEY)) { 
                      enabled = ! url.getParameter(Constants.DISABLED_KEY, false);
                   } else { 
                      enabled = url.getParameter(Constants.ENABLED_KEY, true);
                   }
                   if (enabled) { 
                      /** * url -> dubbo://...... * providerUrl -> dubbo://...... * protocol -> Protocol$Adaptive * extension -> getExtension("dubbo") -> ProtocolFilterWrapper(ProtocolListenerWrapper(DubboProtocol)) * 最后调用DubboProtocol的refer方法 */
                      invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);
                   }
                } catch (Throwable t) { 
                    logger.error("Failed to refer invoker for interface:"+serviceType+",url:("+url+")" + t.getMessage(), t);
                }
                if (invoker != null) {  // 将新的引用放入缓存
                    newUrlInvokerMap.put(key, invoker);
                }
            }else { 
                newUrlInvokerMap.put(key, invoker);
            }
        }
        keys.clear();
        return newUrlInvokerMap;
    }

DubboProtocol

public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException { 
        // create rpc invoker.
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        invokers.add(invoker);
        return invoker;
    }

// DubboProtocol
    private ExchangeClient[] getClients(URL url){ 
        //是否共享连接
        boolean service_share_connect = false;
        int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
        //如果connections不配置，则共享连接，否则每服务每连接
        if (connections == 0){ 
            service_share_connect = true;
            connections = 1;
        }
        
        ExchangeClient[] clients = new ExchangeClient[connections];
        for (int i = 0; i < clients.length; i++) { 
            if (service_share_connect){ 
                clients[i] = getSharedClient(url);
            } else { 
                // 初始化客户端
                clients[i] = initClient(url);
            }
        }
        return clients;
    }

// DubboProtocol
    private ExchangeClient initClient(URL url) { 
        
        // client type setting.
        String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
    
        String version = url.getParameter(Constants.DUBBO_VERSION_KEY);
        boolean compatible = (version != null && version.startsWith("1.0."));
        url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() && compatible ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
        //默认开启heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
        
        // BIO存在严重性能问题，暂时不允许使用
        if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { 
            throw new RpcException("Unsupported client type: " + str + "," +
                    " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
        }
        
        ExchangeClient client ;
        try { 
            //设置连接应该是lazy的 
            if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)){ 
                client = new LazyConnectExchangeClient(url ,requestHandler);
            } else { 
                // HeaderExchanger
                client = Exchangers.connect(url ,requestHandler);
            }
        } catch (RemotingException e) { 
            throw new RpcException("Fail to create remoting client for service(" + url
                    + "): " + e.getMessage(), e);
        }
        return client;
    }

HeaderExchanger

public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { 
        return new HeaderExchangeClient(Transporters.connect(url, new ChannelHandler[]{ new DecodeHandler(new HeaderExchangeHandler(handler))}));
    }

NettyTransporter

public Client connect(URL url, ChannelHandler listener) throws RemotingException { 
        // 创建netty客户端
        return new NettyClient(url, listener);
    }

客户端初始化过程完成。

# 服务调用过程 #

服务调用过程时序图。  
![Dubbo服务调用过程][Dubbo 2]

客户端生成 `Proxy0` 代理

*  client：`Proxy0`
 *  server：`JavassistProxyFactory`

Proxy0

public java.lang.String sayHello(new InvokerInvocationHandler(invoker)) { 
        Object[] args = new Object[1];
        args[0] = ($w) $1;
        Object ret = handler.invoke(this, methods[0], args);
        return (java.lang.String) ret;
    }

InvokerInvocationHandler

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        if (method.getDeclaringClass() == Object.class) { 
            return method.invoke(this.invoker, args);
        } else if ("toString".equals(methodName) && parameterTypes.length == 0) { 
            return this.invoker.toString();
        } else if ("hashCode".equals(methodName) && parameterTypes.length == 0) { 
            return this.invoker.hashCode();
        } else { 
            // invoker -> MockClusterInvoker
            return "equals".equals(methodName) && parameterTypes.length == 1 ? this.invoker.equals(args[0]) : this.invoker.invoke(new RpcInvocation(method, args)).recreate();
        }
    }

MockClusterInvoker

public Result invoke(Invocation invocation) throws RpcException { 
        Result result = null;
        String value = this.directory.getUrl().getMethodParameter(invocation.getMethodName(), "mock", Boolean.FALSE.toString()).trim();
        if (value.length() != 0 && !value.equalsIgnoreCase("false")) { 
            if (value.startsWith("force")) { 
                if (logger.isWarnEnabled()) { 
                    logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + this.directory.getUrl());
                }
    			// 直接调用mock
                result = this.doMockInvoke(invocation, (RpcException)null);
            } else { 
                try { 
                    result = this.invoker.invoke(invocation);
                } catch (RpcException var5) { 
                    if (var5.isBiz()) { 
                        throw var5;
                    }
    
                    if (logger.isWarnEnabled()) { 
                        logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + this.directory.getUrl(), var5);
                    }
    				// 调用异常执行mock方法
                    result = this.doMockInvoke(invocation, var5);
                }
            }
        } else { 
            // 调用执行
            // invoker -> FailoverClusterInvoker invoke方法在其父类AbstractClusterInvoker中
            result = this.invoker.invoke(invocation);
        }
    
        return result;
    }

AbstractClusterInvoker

public Result invoke(Invocation invocation) throws RpcException { 
        this.checkWhetherDestroyed();
        // 获取所有invoker列表>>
        List<Invoker<T>> invokers = this.list(invocation);
        LoadBalance loadbalance;
        if (invokers != null && invokers.size() > 0) { 
            loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(((Invoker)invokers.get(0)).getUrl().getMethodParameter(invocation.getMethodName(), "loadbalance", "random"));
        } else { 
            loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension("random");
        }
    
        RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
        // 执行调用
        return this.doInvoke(invocation, invokers, loadbalance);
    }
    
    // 获取所有invoker列表
    protected List<Invoker<T>> list(Invocation invocation) throws RpcException { 
        List<Invoker<T>> invokers = this.directory.list(invocation);
        return invokers;
    }

AbstractDirectory

public List<Invoker<T>> list(Invocation invocation) throws RpcException { 
        if (this.destroyed) { 
            throw new RpcException("Directory already destroyed .url: " + this.getUrl());
        } else { 
            // 执行获取所有invoker列表
            // invoker -> ProtocolFilterWrapper
            List<Invoker<T>> invokers = this.doList(invocation);
            List<Router> localRouters = this.routers;
            if (localRouters != null && localRouters.size() > 0) { 
                Iterator var4 = localRouters.iterator();
    
                while(var4.hasNext()) { 
                    Router router = (Router)var4.next();
    
                    try { 
                        if (router.getUrl() == null || router.getUrl().getParameter("runtime", true)) { 
                            invokers = router.route(invokers, this.getConsumerUrl(), invocation);
                        }
                    } catch (Throwable var7) { 
                        logger.error("Failed to execute router: " + this.getUrl() + ", cause: " + var7.getMessage(), var7);
                    }
                }
            }
    
            return invokers;
        }
    }

RegistryDirectory

public List<Invoker<T>> doList(Invocation invocation) { 
        if (this.forbidden) { 
            throw new RpcException(4, "No provider available from registry " + this.getUrl().getAddress() + " for service " + this.getConsumerUrl().getServiceKey() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", may be providers disabled or not registered ?");
        } else { 
            List<Invoker<T>> invokers = null;
            Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;
            if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) { 
                String methodName = RpcUtils.getMethodName(invocation);
                Object[] args = RpcUtils.getArguments(invocation);
                if (args != null && args.length > 0 && args[0] != null && (args[0] instanceof String || args[0].getClass().isEnum())) { 
                    invokers = (List)localMethodInvokerMap.get(methodName + "." + args[0]);
                }
    
                if (invokers == null) { 
                    invokers = (List)localMethodInvokerMap.get(methodName);
                }
    
                if (invokers == null) { 
                    invokers = (List)localMethodInvokerMap.get("*");
                }
    
                if (invokers == null) { 
                    Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
                    if (iterator.hasNext()) { 
                        invokers = (List)iterator.next();
                    }
                }
            }
    
            return (List)(invokers == null ? new ArrayList(0) : invokers);
        }
    }

回到 `AbstractClusterInvoker` 的 `invoke` 方法，最后会调用 `FailoverClusterInvoker` 的 `doInvoke` 方法

`FailoverClusterInvoker`

public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
        List<Invoker<T>> copyinvokers = invokers;
        this.checkInvokers(invokers, invocation);
        int len = this.getUrl().getMethodParameter(invocation.getMethodName(), "retries", 2) + 1;
        if (len <= 0) { 
            len = 1;
        }
    
        RpcException le = null;
        List<Invoker<T>> invoked = new ArrayList(invokers.size());
        Set<String> providers = new HashSet(len);
    
        for(int i = 0; i < len; ++i) { 
            if (i > 0) { 
                this.checkWhetherDestroyed();
                copyinvokers = this.list(invocation);
                this.checkInvokers(copyinvokers, invocation);
            }
    		// 选择调用invoker>>
            Invoker<T> invoker = this.select(loadbalance, invocation, copyinvokers, invoked);
            invoked.add(invoker);
            RpcContext.getContext().setInvokers(invoked);
    
            try { 
                // 执行调用
                Result result = invoker.invoke(invocation);
                if (le != null && logger.isWarnEnabled()) { 
                    logger.warn("Although retry the method " + invocation.getMethodName() + " in the service " + this.getInterface().getName() + " was successful by the provider " + invoker.getUrl().getAddress() + ", but there have been failed providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + this.directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + le.getMessage(), le);
                }
    
                Result var12 = result;
                return var12;
            } catch (RpcException var17) { 
                if (var17.isBiz()) { 
                    throw var17;
                }
    
                le = var17;
            } catch (Throwable var18) { 
                le = new RpcException(var18.getMessage(), var18);
            } finally { 
                providers.add(invoker.getUrl().getAddress());
            }
        }
    
        throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method " + invocation.getMethodName() + " in the service " + this.getInterface().getName() + ". Tried " + len + " times of the providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + this.directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + (le != null ? le.getMessage() : ""), (Throwable)(le != null && le.getCause() != null ? le.getCause() : le));
    }

// AbstractClusterInvoker
    protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException { 
        if (invokers != null && invokers.size() != 0) { 
            String methodName = invocation == null ? "" : invocation.getMethodName();
            boolean sticky = ((Invoker)invokers.get(0)).getUrl().getMethodParameter(methodName, "sticky", false);
            if (this.stickyInvoker != null && !invokers.contains(this.stickyInvoker)) { 
                this.stickyInvoker = null;
            }
    
            if (sticky && this.stickyInvoker != null && (selected == null || !selected.contains(this.stickyInvoker)) && this.availablecheck && this.stickyInvoker.isAvailable()) { 
                return this.stickyInvoker;
            } else { 
                Invoker<T> invoker = this.doselect(loadbalance, invocation, invokers, selected);
                if (sticky) { 
                    this.stickyInvoker = invoker;
                }
    
                return invoker;
            }
        } else { 
            return null;
        }
    }

// AbstractClusterInvoker
    private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException { 
        if (invokers != null && invokers.size() != 0) { 
            if (invokers.size() == 1) { 
                return (Invoker)invokers.get(0);
            } else if (invokers.size() == 2 && selected != null && selected.size() > 0) { 
                return selected.get(0) == invokers.get(0) ? (Invoker)invokers.get(1) : (Invoker)invokers.get(0);
            } else { 
                // 负载均衡策略执行选择
                Invoker<T> invoker = loadbalance.select(invokers, this.getUrl(), invocation);
                if (selected != null && selected.contains(invoker) || !invoker.isAvailable() && this.getUrl() != null && this.availablecheck) { 
                    try { 
                        Invoker<T> rinvoker = this.reselect(loadbalance, invocation, invokers, selected, this.availablecheck);
                        if (rinvoker != null) { 
                            invoker = rinvoker;
                        } else { 
                            int index = invokers.indexOf(invoker);
    
                            try { 
                                invoker = index < invokers.size() - 1 ? (Invoker)invokers.get(index + 1) : invoker;
                            } catch (Exception var9) { 
                                logger.warn(var9.getMessage() + " may because invokers list dynamic change, ignore.", var9);
                            }
                        }
                    } catch (Throwable var10) { 
                        logger.error("clustor relselect fail reason is :" + var10.getMessage() + " if can not slove ,you can set cluster.availablecheck=false in url", var10);
                    }
                }
    
                return invoker;
            }
        } else { 
            return null;
        }
    }

AbstractLoadBalance

public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) { 
        if (invokers != null && invokers.size() != 0) { 
            return invokers.size() == 1 ? (Invoker)invokers.get(0) : this.doSelect(invokers, url, invocation);
        } else { 
            return null;
        }
    }
    // 模板方法
    protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> var1, URL var2, Invocation var3);

RandomLoadBalance

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { 
        int length = invokers.size();
        int totalWeight = 0;
        boolean sameWeight = true;
    
        int offset;
        int i;
        for(offset = 0; offset < length; ++offset) { 
            i = this.getWeight((Invoker)invokers.get(offset), invocation);
            totalWeight += i;
            if (sameWeight && offset > 0 && i != this.getWeight((Invoker)invokers.get(offset - 1), invocation)) { 
                sameWeight = false;
            }
        }
    
        if (totalWeight > 0 && !sameWeight) { 
            offset = this.random.nextInt(totalWeight);
    
            for(i = 0; i < length; ++i) { 
                offset -= this.getWeight((Invoker)invokers.get(i), invocation);
                if (offset < 0) { 
                    return (Invoker)invokers.get(i);
                }
            }
        }
    
        return (Invoker)invokers.get(this.random.nextInt(length));
    }

回到 `FailoverClusterInvoker` 的 `doInvoke` 方法，会继续调用 `DubboProtocol` ，而 `invoke` 方法在其父类 `AbstractInvoker` 中

`AbstractInvoker`

public Result invoke(Invocation inv) throws RpcException { 
        if (this.destroyed.get()) { 
            throw new RpcException("Rpc invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + " is DESTROYED, can not be invoked any more!");
        } else { 
            RpcInvocation invocation = (RpcInvocation)inv;
            invocation.setInvoker(this);
            if (this.attachment != null && this.attachment.size() > 0) { 
                invocation.addAttachmentsIfAbsent(this.attachment);
            }
    
            Map<String, String> context = RpcContext.getContext().getAttachments();
            if (context != null) { 
                invocation.addAttachmentsIfAbsent(context);
            }
    
            if (this.getUrl().getMethodParameter(invocation.getMethodName(), "async", false)) { 
                invocation.setAttachment("async", Boolean.TRUE.toString());
            }
    
            RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
    
            try { 
                return this.doInvoke(invocation);
            } catch (InvocationTargetException var6) { 
                Throwable te = var6.getTargetException();
                if (te == null) { 
                    return new RpcResult(var6);
                } else { 
                    if (te instanceof RpcException) { 
                        ((RpcException)te).setCode(3);
                    }
    
                    return new RpcResult(te);
                }
            } catch (RpcException var7) { 
                if (var7.isBiz()) { 
                    return new RpcResult(var7);
                } else { 
                    throw var7;
                }
            } catch (Throwable var8) { 
                return new RpcResult(var8);
            }
        }
    }
    
    protected abstract Result doInvoke(Invocation var1) throws Throwable;

DubboInvoker

protected Result doInvoke(Invocation invocation) throws Throwable { 
        RpcInvocation inv = (RpcInvocation)invocation;
        String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment("path", this.getUrl().getPath());
        inv.setAttachment("version", this.version);
        ExchangeClient currentClient;
        if (this.clients.length == 1) { 
            currentClient = this.clients[0];
        } else { 
            currentClient = this.clients[this.index.getAndIncrement() % this.clients.length];
        }
    
        try { 
            boolean isAsync = RpcUtils.isAsync(this.getUrl(), invocation);
            boolean isOneway = RpcUtils.isOneway(this.getUrl(), invocation);
            int timeout = this.getUrl().getMethodParameter(methodName, "timeout", 1000);
            if (isOneway) { 
                boolean isSent = this.getUrl().getMethodParameter(methodName, "sent", false);
                currentClient.send(inv, isSent);
                RpcContext.getContext().setFuture((Future)null);
                return new RpcResult();
            } else if (isAsync) { 
                ResponseFuture future = currentClient.request(inv, timeout);
                RpcContext.getContext().setFuture(new FutureAdapter(future));
                return new RpcResult();
            } else { 
                RpcContext.getContext().setFuture((Future)null);
                // 发送请求>>
                // currentClient -> 当前创建的连接
                return (Result)currentClient.request(inv, timeout).get();
            }
        } catch (TimeoutException var9) { 
            throw new RpcException(2, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + this.getUrl() + ", cause: " + var9.getMessage(), var9);
        } catch (RemotingException var10) { 
            throw new RpcException(1, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + this.getUrl() + ", cause: " + var10.getMessage(), var10);
        }
    }

HeaderExchangeClient

public ResponseFuture request(Object request, int timeout) throws RemotingException { 
        return this.channel.request(request, timeout);
    }

HeaderExchangeChannel

public ResponseFuture request(Object request, int timeout) throws RemotingException { 
        if (this.closed) { 
            throw new RemotingException(this.getLocalAddress(), (InetSocketAddress)null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
        } else { 
            Request req = new Request();
            req.setVersion("2.0.0");
            req.setTwoWay(true);
            req.setData(request);
            DefaultFuture future = new DefaultFuture(this.channel, req, timeout);
    
            try { 
                this.channel.send(req);
                return future;
            } catch (RemotingException var6) { 
                future.cancel();
                throw var6;
            }
        }
    }
    
    public void send(Object message) throws RemotingException { 
        this.send(message, this.getUrl().getParameter("sent", false));
    }
    
    public void send(Object message, boolean sent) throws RemotingException { 
        if (this.closed) { 
            throw new RemotingException(this.getLocalAddress(), (InetSocketAddress)null, "Failed to send message " + message + ", cause: The channel " + this + " is closed!");
        } else { 
            if (!(message instanceof Request) && !(message instanceof Response) && !(message instanceof String)) { 
                Request request = new Request();
                request.setVersion("2.0.0");
                request.setTwoWay(false);
                request.setData(message);
                this.channel.send(request, sent);
            } else { 
                // channel -> NettyChannel
                this.channel.send(message, sent);
            }
        }
    }

NettyChannel

public void send(Object message, boolean sent) throws RemotingException { 
        super.send(message, sent);
        boolean success = true;
        int timeout = 0;
    
        try { 
            ChannelFuture future = this.channel.write(message);
            if (sent) { 
                timeout = this.getUrl().getPositiveParameter("timeout", 1000);
                success = future.await((long)timeout);
            }
    
            Throwable cause = future.getCause();
            if (cause != null) { 
                throw cause;
            }
        } catch (Throwable var7) { 
            throw new RemotingException(this, "Failed to send message " + message + " to " + this.getRemoteAddress() + ", cause: " + var7.getMessage(), var7);
        }
    
        if (!success) { 
            throw new RemotingException(this, "Failed to send message " + message + " to " + this.getRemoteAddress() + "in timeout(" + timeout + "ms) limit");
        }
    }

服务调用完成。

[Dubbo]: /images/20220422/a920125db8ef4880829dd52c8d692f8b.png
[Dubbo 1]: /images/20220422/f76c7987b519477bb29f68a76fd40b38.png
[Dubbo 2]: /images/20220422/350a18692abe47a5b7ca1f6473502eb9.png