dubbo进阶之服务注册消费

soulballad约 7353 字大约 25 分钟

Invoker是什么

从前面的分析来看，服务的发布分三个阶段：

第一个阶段会创造一个invoker
第二个阶段会把经历过一系列处理的invoker（各种包装），在DubboProtocol中保存到exporterMap中
第三个阶段把dubbo协议的url地址注册到注册中心上

前面没有分析Invoker，我们来简单看看Invoker到底是一个啥东西。

Invoker是Dubbo领域模型中非常重要的一个概念，和 ExtensionLoader 的重要性是一样的，如果Invoker没有搞懂，那么不算是看懂了Dubbo的源码。我们继续回到 ServiceConﬁg 中 export 的代码，这段代码是还没有分析过的。以这个作为入口来分析我们前面 export 出去的 nvoker 到底是啥东西

Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));

ProxyFacotory.getInvoker

这个是一个代理工厂，用来生成invoker，从它的定义来看，它是一个自适应扩展点，看到这样的扩展
点，我们几乎可以不假思索的想到它会存在一个动态适配器类

ProxyFactory proxyFactory = 
ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

ProxyFactory

这个方法的简单解读为：它是一个spi扩展点，并且默认的扩展实现是 javassit，这个接口中有三个方法，并且都是加了@Adaptive 的自适应扩展点。所以如果调用 getInvoker 方法，应该会返回一个
ProxyFactory$Adaptive

@SPI("javassist")
public interface ProxyFactory {

    @Adaptive({Constants.PROXY_KEY})
    <T> T getProxy(Invoker<T> invoker) throws RpcException;

    @Adaptive({Constants.PROXY_KEY})
    <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException;

    @Adaptive({Constants.PROXY_KEY})
    <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException;

ProxyFactory$Adaptive

这个自适应扩展点，做了两件事情

通过ExtensionLoader.getExtensionLoader(ProxyFactory.class).getExtension(extName)获取了一个指定名称的扩展点；
在dubbo-rpc-api/resources/META-INF/com.alibaba.dubbo.rpc.ProxyFactory中，定义了
javassis=JavassisProxyFactory
调用JavassisProxyFactory的getInvoker方法

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

    public java.lang.Object getProxy(org.apache.dubbo.rpc.Invoker arg0, boolean 
                                     arg1) throws org.apache.dubbo.rpc.RpcException {
        if (arg0 == null) 
            throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument == null");
        if (arg0.getUrl() == null) 
            throw new IllegalArgumentException("org.apache.dubbo.rpc.Invoker argument getUrl() == null");
        org.apache.dubbo.common.URL url = arg0.getUrl();
        String extName = url.getParameter("proxy", "javassist");
        if(extName == null) throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.ProxyFactory) name from url (" + url.toString() + ") use keys([proxy])");
        org.apache.dubbo.rpc.ProxyFactory extension = 
            (org.apache.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(
            org.apache.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        return extension.getProxy(arg0, arg1);
    }

    public org.apache.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, 
                                                   java.lang.Class arg1, org.apache.dubbo.common.URL arg2) throws org.apache.dubbo.rpc.RpcException {
        if (arg2 == null) 
            throw new IllegalArgumentException("url == null");
        org.apache.dubbo.common.URL url = arg2;
        String extName = url.getParameter("proxy", "javassist");
        if(extName == null) 
            throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.ProxyFactory) name from url (" + url.toString() + ") use keys([proxy])");
        org.apache.dubbo.rpc.ProxyFactory extension = 
            (org.apache.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(
            org.apache.dubbo.rpc.ProxyFactory.class).getExtension(extName);
        return extension.getInvoker(arg0, arg1, arg2);
    }
}

JavassistProxyFactory.getInvoker

javassist 是一个动态类库，用来实现动态代理的。

proxy：接口的实现: com.gupao.springboot.dubbo.provider.SayHelloServiceImpl
type：接口全称 com.gupao.springboot.dubbo.api.SayHelloService
url：协议地址：registry://...

@Override
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
    // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
    final Wrapper wrapper = 
        Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? 
                           proxy.getClass() : type);
    return new AbstractProxyInvoker<T>(proxy, type, url) {
        @Override
        protected Object doInvoke(T proxy, String methodName,
                                  Class<?>[] parameterTypes,
                                  Object[] arguments) throws Throwable {
            return wrapper.invokeMethod(proxy, methodName, parameterTypes, 
                                        arguments);
        }
    };
}

javassist生成的动态代理代码

通过断点的方式（Wrapper.java 258行），在Wrapper.getWrapper 中的 makeWrapper，会创建一个动态代理，核心的方法 invokeMethod 代码如下：

1566662632024

public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws java.lang.reflect.InvocationTargetException {
    com.gupao.springboot.dubbo.provider.SayHelloServiceImpl w;
    try {
        w = ((com.gupao.springboot.dubbo.provider.SayHelloServiceImpl) $1);
    } catch (Throwable e) {
        throw new IllegalArgumentException(e);
    }
    try {
        if ("sayHello".equals($2) && $3.length == 0) {
            return ($w) w.sayHello();
        }
    } catch (Throwable e) {
        throw new java.lang.reflect.InvocationTargetException(e);
    }
    throw new org.apache.dubbo.common.bytecode.NoSuchMethodException("Not found method \"" + $2 + "\" in class com.gupao.springboot.dubbo.provider.SayHelloServiceImpl.");
}

构建好了代理类之后，返回一个 AbstractproxyInvoker，并且它实现了 doInvoke 方法，这个地方似乎看到了dubbo 消费者调用过来的时候触发的影子，因为 wrapper.invokeMethod 本质上就是触发上面动态代理类的方法invokeMethod。

1566662869431

所以，简单总结一下Invoke本质上应该是一个代理，经过层层包装最终进行了发布。当消费者发起请求的时候，会获得这个invoker进行调用。

最终发布出去的invoker, 也不是单纯的一个代理，也是经过多层包装InvokerDelegate(DelegateProviderMetaDataInvoker(AbstractProxyInvoker()))

服务注册流程

关于服务发布这一条线分析完成之后，再来了解一下服务注册的过程，希望大家还记得我们之所以走到这一步，是因为我们在 RegistryProtocol 这个类中，看到了服务发布的流程。

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws 
    RpcException {
    URL registryUrl = getRegistryUrl(originInvoker);
    // url to export locally
    URL providerUrl = getProviderUrl(originInvoker);

    // Subscribe the override data
    // FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call
    //  the same service. Because the subscribed is cached key with the name of the service, it causes the
    //  subscription information to cover.
    final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
    final OverrideListener overrideSubscribeListener = new 
        OverrideListener(overrideSubscribeUrl, originInvoker);
    overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);

    providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
    //export invoker
    final ExporterChangeableWrapper<T> exporter = 
        doLocalExport(originInvoker, providerUrl);

    // url to registry
    final Registry registry = getRegistry(originInvoker);
    final URL registeredProviderUrl = getRegisteredProviderUrl(providerUrl, 
                                                               registryUrl);
    ProviderInvokerWrapper<T> providerInvokerWrapper = 
        ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, 
                                                  registeredProviderUrl);
    //to judge if we need to delay publish
    boolean register = registeredProviderUrl.getParameter("register", true);
    if (register) {
        register(registryUrl, registeredProviderUrl);
        providerInvokerWrapper.setReg(true);
    }

    // Deprecated! Subscribe to override rules in 2.6.x or before.
    registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);

    exporter.setRegisterUrl(registeredProviderUrl);
    exporter.setSubscribeUrl(overrideSubscribeUrl);
    //Ensure that a new exporter instance is returned every time export
    return new DestroyableExporter<>(exporter);
}

服务注册核心代码

从export方法中抽离出来的部分代码，就是服务注册的流程

// url to registry
final Registry registry = getRegistry(originInvoker);
final URL registeredProviderUrl = getRegisteredProviderUrl(providerUrl, registryUrl);
ProviderInvokerWrapper<T> providerInvokerWrapper = 
    ProviderConsumerRegTable.registerProvider(originInvoker,
                                              registryUrl, registeredProviderUrl);
//to judge if we need to delay publish
boolean register = registeredProviderUrl.getParameter("register", true);
if (register) {
    register(registryUrl, registeredProviderUrl);
    providerInvokerWrapper.setReg(true);
}

getRegistry

把url转化为对应配置的注册中心的具体协议
根据具体协议，从registryFactory中获得指定的注册中心实现

那么这个registryFactory具体是怎么赋值的呢？

private Registry getRegistry(final Invoker<?> originInvoker) {
    //把url转化为配置的具体协议，比如zookeeper://ip:port. 这样后续获得的注册中心就会是基于zk的实现
    URL registryUrl = getRegistryUrl(originInvoker);
    return registryFactory.getRegistry(registryUrl);
}

在 RegistryProtocol 中存在一段这样的代码，很明显这是通过依赖注入来实现的扩展点。

private RegistryFactory registryFactory; 
public void setRegistryFactory(RegistryFactory registryFactory) {
    this.registryFactory = registryFactory;
}

按照扩展点的加载规则，我们可以先看看 /META-INF/dubbo/internal 路径下找到 RegistryFactory 的配置文件，这个 factory 有多个扩展点的实现。

dubbo=org.apache.dubbo.registry.dubbo.DubboRegistryFactory
multicast=org.apache.dubbo.registry.multicast.MulticastRegistryFactory
zookeeper=org.apache.dubbo.registry.zookeeper.ZookeeperRegistryFactory
redis=org.apache.dubbo.registry.redis.RedisRegistryFactory
consul=org.apache.dubbo.registry.consul.ConsulRegistryFactory

etcd3=org.apache.dubbo.registry.etcd.EtcdRegistryFactory

接着，找到 RegistryFactory 的实现, 发现它里面有一个自适应的方法，根据url中protocol传入的值进行适配

@SPI("dubbo")
public interface RegistryFactory {

    @Adaptive({"protocol"})
    Registry getRegistry(URL url);

RegistryFactory$Adaptive

由于在前面的代码中，url中的 protocol 已经改成了 zookeeper，那么这个时候根据 zookeeper 获得的 spi 扩展点应该是 ZookeeperRegistryFactory

import org.apache.dubbo.common.extension.ExtensionLoader;
public class RegistryFactory$Adaptive implements 
    org.apache.dubbo.registry.RegistryFactory {
    public org.apache.dubbo.registry.Registry 
        getRegistry(org.apache.dubbo.common.URL arg0)  {
        if (arg0 == null) throw new IllegalArgumentException("url == null");
        org.apache.dubbo.common.URL url = arg0;
        String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
        if(extName == null) throw new IllegalStateException("Failed to get extension (org.apache.dubbo.registry.RegistryFactory) name from url (" + url.toString() + ") use keys([protocol])");
        org.apache.dubbo.registry.RegistryFactory extension = (org.apache.dubbo.registry.RegistryFactory)ExtensionLoader.getExtensionLoader(
            org.apache.dubbo.registry.RegistryFactory.class).getExtension(extName);
        return extension.getRegistry(arg0);
    }
}

ZookeeperRegistryFactory

这个方法中并没有 getRegistry 方法，而是在父类 AbstractRegistryFactory

从缓存 REGISTRIES 中，根据 key 获得对应的 Registry；
如果不存在，则创建 Registry

public Registry getRegistry(URL url) {
    url = URLBuilder.from(url)
        .setPath(RegistryService.class.getName())
        .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
        .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY)
        .build();
    String key = url.toServiceStringWithoutResolving();
    // Lock the registry access process to ensure a single instance of the registry
    LOCK.lock();
    try {
        Registry registry = REGISTRIES.get(key);
        if (registry != null) {
            return registry;
        }
        //创建注册中心
        registry = createRegistry(url);
        if (registry == null) {
            throw new IllegalStateException("Can not create registry " + url);
        }
        REGISTRIES.put(key, registry);
        return registry;
    } finally {
        // Release the lock
        LOCK.unlock();
    }
}

createRegistry

创建一个 zookeeperRegistry，把 url 和 zookeepertransporter 作为参数传入。

zookeeperTransporter 这个属性也是基于依赖注入来赋值的，具体的流程就不再分析了，这个的值应该是
CuratorZookeeperTransporter 表示具体使用什么框架来和zk产生连接

public Registry createRegistry(URL url) {
    return new ZookeeperRegistry(url, zookeeperTransporter);
}

ZookeeperRegistry

这个方法中使用了 CuratorZookeeperTransport 来实现 zk 的连接

public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
    super(url);
    if (url.isAnyHost()) {
        throw new IllegalStateException("registry address == null");
    }
    //获得group名称
    String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
    if (!group.startsWith(Constants.PATH_SEPARATOR)) {
        group = Constants.PATH_SEPARATOR + group;
    }
    this.root = group;
    //产生一个zookeeper连接
    zkClient = zookeeperTransporter.connect(url);
    //添加zookeeper状态变化事件
    zkClient.addStateListener(state -> {
        if (state == StateListener.RECONNECTED) {
            try {
                recover();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    });
}

register

register(registryUrl, registeredProviderUrl);

registry.register(registedProviderUrl)

继续往下分析，会调用registry.register去讲dubbo://的协议地址注册到zookeeper上这个方法会调用FailbackRegistry类中的register. 为什么呢？

register(registryUrl, registeredProviderUrl);

public void register(URL registryUrl, URL registeredProviderUrl) {
    Registry registry = registryFactory.getRegistry(registryUrl);
    registry.register(registeredProviderUrl);
}

因为 ZookeeperRegistry 这个类中并没有 register 这个方法，但是他的父类FailbackRegistry中存在register方法，而这个类又重写了AbstractRegistry类中的register方法。所以我们可以直接定位到FailbackRegistry这个类中的register方法中

1566664444611

public class ZookeeperRegistry extends FailbackRegistry {
	// ...
}

FailbackRegistry.register

FailbackRegistry，从名字上来看，是一个失败重试机制：

调用父类的register方法，将当前url添加到缓存集合中
调用 doRegister 方法，这个方法很明显，是一个抽象方法，会由ZookeeperRegistry子类实现。

public void register(URL url) {
    super.register(url);
    removeFailedRegistered(url);
    removeFailedUnregistered(url);
    try {
        // 调用子类实现真正的服务注册，把url注册到zk上
        doRegister(url);
    } catch (Exception e) {
        Throwable t = e;

        // 如果开启了启动时检测，则直接抛出异常
        boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
            && url.getParameter(Constants.CHECK_KEY, true)
            && !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
        boolean skipFailback = t instanceof SkipFailbackWrapperException;
        if (check || skipFailback) {
            if (skipFailback) {
                t = t.getCause();
            }
            throw new IllegalStateException("Failed to register " + url + "to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
        } else {
            logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
        }

        // 将失败的注册请求记录到失败列表，定时重试
        addFailedRegistered(url);
    }
}

ZookeeperRegistry.doRegister

最终调用curator的客户端把服务地址注册到zk

public void doRegister(URL url) {
    try {
        zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
    } catch (Throwable e) {
        throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

思考服务消费应该要具备的逻辑

如果要实现服务的消费，大家可以结合之前手写rpc的课程来思考一下

生成远程服务的代理
获得目标服务的url地址
实现远程网络通信
实现负载均衡
实现集群容错

1566394934402

服务的消费

消费端的代码解析是从下面这段代码开始的

<dubbo:reference id="xxxService" interface="xxx.xxx.Service"/>

注解的方式的初始化入口是

ReferenceAnnotationBeanPostProcessor --> ReferenceBeanInvocationHandler.init --> ReferenceConfig.get() 获得一个远程代理类

ReferenceConﬁg.get

public synchronized T get() {
    checkAndUpdateSubConfigs(); //检查和修改配置

    if (destroyed) {
        throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
    }
    if (ref == null) { //如果当前接口的远程代理引用为空，则进行初始化
        init();
    }
    return ref;
}

init

初始化的过程，和服务发布的过程类似，会有特别多的判断以及参数的组装. 我们只需要关注 createProxy，创建代理类的方法。

private void init() {
    //省略...
    ref = createProxy(map);   
    //省略...
}

createProxy

代码比较长，但是逻辑相对比较清晰

判断是否为本地调用，如果是则使用 injvm 协议进行调用
判断是否为点对点调用，如果是则把url保存到urls集合中，如果url为1，进入步骤4，如果urls>1，则执行5
如果是配置了注册中心，遍历注册中心，把url添加到urls集合，url为1，进入步骤4，如果urls>1，则执行5
直连构建invoker
构建invokers集合，通过cluster合并多个invoker
最后调用 ProxyFactory 生成代理类

private T createProxy(Map<String, String> map) {
    if (shouldJvmRefer(map)) { //判断是否是在同一个jvm进程中调用
        URL url = new URL(Constants.LOCAL_PROTOCOL, Constants.LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
        invoker = refprotocol.refer(interfaceClass, url);
        if (logger.isInfoEnabled()) {
            logger.info("Using injvm service " + interfaceClass.getName());
        }
    } else {
        //url 如果不为空，说明是点对点通信
        if (url != null && url.length() > 0) { // user specified URL, could 
            be peer-to-peer address, or register center's address.
                String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
            if (us != null && us.length > 0) {
                for (String u : us) {
                    URL url = URL.valueOf(u);
                    if (StringUtils.isEmpty(url.getPath())) {
                        url = url.setPath(interfaceName);
                    }
                    // 检测 url 协议是否为 registry，若是，表明用户想使用指定的注册中心
                    if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                        // 将 map 转换为查询字符串，并作为 refer 参数的值添加到url 中
                        urls.add(url.addParameterAndEncoded(Constants.REFER_KEY,  StringUtils.toQueryString(map)));
                    } else {
                        // 合并 url，移除服务提供者的一些配置（这些配置来源于用户配置的url属性），
                        // 比如线程池相关配置。并保留服务提供者的部分配置，比如版本，group，时间戳等
                        // 最后将合并后的配置设置为 url 查询字符串中。
                        urls.add(ClusterUtils.mergeUrl(url, map));
                    }
                }
            }
        } else { // assemble URL from register center's configuration
            checkRegistry(); //校验注册中心的配置以及是否有必要从配置中心组装url
            //这里的代码实现和服务端类似，也是根据注册中心配置进行解析得到URL
            //这里的URL肯定也是：registry://ip:port/org.apache.dubbo.service.RegsitryService
            List<URL> us = loadRegistries(false);
            if (CollectionUtils.isNotEmpty(us)) {
                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.isEmpty()) {
                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.");
            }
        }
        // 如果值配置了一个注册中心或者一个服务提供者， 直接使用 refprotocol.refer
        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) {//遍历urls生成多个invoker
                invokers.add(refprotocol.refer(interfaceClass, url));
                if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                    registryURL = url; // use last registry url
                }
            }
            if (registryURL != null) { //如果registryUrl不为空，构建静态
                // 使用RegistryAwareCluster
                URL u = registryURL.addParameter(Constants.CLUSTER_KEY, 
                                                 RegistryAwareCluster.NAME);
                // 通过Cluster将多个invoker合并 
                // RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
                invoker = cluster.join(new StaticDirectory(u, invokers));
            } else { 
                invoker = cluster.join(new StaticDirectory(invokers));
            }
        }
    }
    //检查invoker的有效性
    if (shouldCheck() && !invoker.isAvailable()) {
        // make it possible for consumer to retry later if provider is 
        temporarily unavailable
            initialized = false;
        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());
    }
    /**
     * @since 2.7.0
     * ServiceData Store
     */
    MetadataReportService metadataReportService = null;
    if ((metadataReportService = getMetadataReportService()) != null) {
        URL consumerURL = new URL(Constants.CONSUMER_PROTOCOL, 
                                  map.remove(Constants.REGISTER_IP_KEY), 0, map.get(Constants.INTERFACE_KEY), map);
        metadataReportService.publishConsumer(consumerURL);
    }
    // create service proxy
    return (T) proxyFactory.getProxy(invoker);
}

protocol.refer

这里通过指定的协议来调用 refer 生成一个 invoker 对象，invoker 前面讲过，它是一个代理对象。那么在当前的消费端而言，invoker 主要用于执行远程调用。

这个protocol，又是一个自适应扩展点，它得到的是一个Protocol$Adaptive.

Protocol refprotocol = 
    ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension()

这段代码中，根据当前的协议url，得到一个指定的扩展点，传递进来的参数中，协议地址为 registry://，所以，我们可以直接定位到 RegistryProtocol.refer 代码

Protocol$Adaptive 中的 refer 方法

根据当前的协议扩展名 registry，获得一个被包装过的 RegistryProtocol

public org.apache.dubbo.rpc.Invoker refer(java.lang.Class arg0, 
                                          org.apache.dubbo.common.URL arg1) throws org.apache.dubbo.rpc.RpcException {
    if (arg1 == null) 
        throw new IllegalArgumentException("url == null");  org.apache.dubbo.common.URL url = arg1;
    String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
    if(extName == null) 
        throw new IllegalStateException("Failed to get extension (org.apache.dubbo.rpc.Protocol) name from url (" + url.toString() + ") use keys([protocol])");
    org.apache.dubbo.rpc.Protocol extension = (org.apache.dubbo.rpc.Protocol)ExtensionLoader.getExtensionLoader(
        org.apache.dubbo.rpc.Protocol.class).getExtension(extName);
    return extension.refer(arg0, arg1);
}

RegistryProtocol.refer

这里面的代码逻辑比较简单

组装注册中心协议的 url
判断是否配置了 group，如果有，则 cluster=getMergeableCluster()，构建 invoker
doRefer 构建 invoker

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    //这段代码也很熟悉，就是根据配置的协议，生成注册中心的url: zookeeper://
    url = URLBuilder.from(url)
        .setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
        .removeParameter(REGISTRY_KEY)
        .build();
    Registry registry = registryFactory.getRegistry(url);
    if (RegistryService.class.equals(type)) {
        return proxyFactory.getInvoker((T) registry, type, url);
    }

    // 解析group参数，根据group决定cluster的类型
    Map<String, String> qs = 
        StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
    String group = qs.get(Constants.GROUP_KEY);
    if (group != null && group.length() > 0) {
        if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || 
            "*".equals(group)) {
            return doRefer(getMergeableCluster(), registry, type, url);
        }
    }
    return doRefer(cluster, registry, type, url);
}

doRefer

doRefer 里面就稍微复杂一些，涉及到比较多的东西，我们先关注主线

构建一个 RegistryDirectory
构建一个 consumer：//协议的地址注册到注册中心
订阅 zookeeper 中节点的变化
调用 cluster.join 方法

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, 
                               Class<T> type, URL url) {
    //RegistryDirectory初始化
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    // all attributes of REFER_KEY
    Map<String, String> parameters = 
        new HashMap<String, String>(directory.getUrl().getParameters());
    //注册consumer://协议的url
    URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
    if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
        directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
        registry.register(directory.getRegisteredConsumerUrl());
    }
    directory.buildRouterChain(subscribeUrl);
    //订阅事件监听
    directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY, PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));
    //构建invoker
    Invoker invoker = cluster.join(directory);
    ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
    return invoker;
}

Cluster是什么

我们只关注一下Invoker这个代理类的创建过程,其他的暂且不关心

Invoker invoker=cluster.join(directory)

cluster其实是在RegistryProtocol中通过set方法完成依赖注入的，并且，它还是一个被包装的

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

Cluster扩展点的定义, 由于它是一个自适应扩展点，那么会动态生成一个Cluster$Adaptive的动态代理类

@SPI(FailoverCluster.NAME)
public interface Cluster {

    @Adaptive
    <T> Invoker<T> join(Directory<T> directory) throws RpcException;
}

Cluster$Adaptive

在动态适配的类中会基于 extName，选择一个合适的扩展点进行适配，由于默认情况下 cluster: failover，所以
getExtension("failover") 理论上应该返回 FailOverCluster。

但实际上，这里做了包装 MockClusterWrapper（FailOverCluster）

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

cluster.join

所以再回到doRefer方法，下面这段代码, 实际是调用 MockClusterWrapper(FailOverCluster.join)

 Invoker invoker = cluster.join(directory);

所以这里返回的 invoker，应该是 MockClusterWrapper(FailOverCluster((directory))

1566668274799

接着回到ReferenceConﬁg.createProxy方法中的最后一行

proxyFactory.getProxy

拿到invoker之后，会调用获得一个动态代理类

return (T) proxyFactory.getProxy(invoker);

而这里的 proxyFactory又是一个自适应扩展点，所以会进入下面的方法

JavassistProxyFactory.getProxy

通过这个方法生成了一个动态代理类，并且对 invoker 再做了一层处理，InvokerInvocationHandler。
意味着后续发起服务调用的时候，会由 InvokerInvocationHandler 来进行处理。

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

proxy.getProxy

在 proxy.getProxy 这个方法中会生成一个动态代理类，通过debug的形式可以看到动态代理类的原貌

在getProxy这个方法位置加一个断点

proxy = (Proxy) pc.newInstance();

然后在debug窗口，找到 ccp 这个变量 -> mMethods。

1566668840470

调整一下格式：

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

从这个sayHello方法可以看出，我们通过 @Reference注入的一个对象实例本质上就是一个动态代理类，通过调用这个类中的方法，会触发 handler.invoke()，而这个 handler 就是 InvokerInvocationHandler

网络连接的建立

前面分析的逻辑中，只讲到了动态代理类的生成，那么目标服务地址信息以及网络通信的建立在哪里实现的呢？我们继续回到 RegistryProtocol.refer 这个方法中

这里我们暂且关注 directory.subscribe 这个方法，它是实现服务目标服务订阅的

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);
    // all attributes of REFER_KEY
    Map<String, String> parameters = 
        new HashMap<String, String>(directory.getUrl().getParameters());
    URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
    if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
        directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
        registry.register(directory.getRegisteredConsumerUrl());
    }
    directory.buildRouterChain(subscribeUrl);
    directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));

    /*省略部分代码*/
    return invoker;
}

订阅注册中心指定节点的变化，如果发生变化，则通知到 RegistryDirectory。Directory 其实和服务的注册以及服务的发现有非常大的关联.

public void subscribe(URL url) {
    setConsumerUrl(url); //设置consumerUrl
    //把当前RegistryDirectory作为listener，去监听zk上节点的变化
    consumerConfigurationListener.addNotifyListener(this); 
    serviceConfigurationListener = new ReferenceConfigurationListener(this, url);
    registry.subscribe(url, this);//订阅 -> 这里的registry是zookeeperRegsitry
}

这里的registry 是ZookeeperRegistry ，会去监听并获取路径下面的节点。监听的路径是：

/dubbo/org.apache.dubbo.demo.DemoService/providers、
/dubbo/org.apache.dubbo.demo.DemoService/configurators、
/dubbo/org.apache.dubbo.demo.DemoService/routers

节点下面的子节点变动

listener为RegistryDirectory，后续要用到移除失效的listener，调用doSubscribe进行订阅

public void subscribe(URL url, NotifyListener listener) {
    super.subscribe(url, listener);
    removeFailedSubscribed(url, listener);
    try {
        // Sending a subscription request to the server side
        doSubscribe(url, listener);
    } catch (Exception e) {
        Throwable t = e;

        List<URL> urls = getCacheUrls(url);
        if (CollectionUtils.isNotEmpty(urls)) {
            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 {
            // If the startup detection is opened, the Exception is thrown directly.
            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);
            }
        }

        // Record a failed registration request to a failed list, retry regularly
        addFailedSubscribed(url, listener);
    }
}

ZookeeperRegistry.doSubscribe

这个方法是订阅，逻辑实现比较多，可以分两段来看，这里的实现把所有 Service 层发起的订阅以及指定的Service 层发起的订阅分开处理。所有 Service 层类似于监控中心发起的订阅。指定的 Service 层发起的订阅可以看作是服务消费者的订阅。我们只需要关心指定 service 层发起的订阅即可

public void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
            //省略部分代码
        } else {
            List<URL> urls = new ArrayList<>();
            for (String path : toCategoriesPath(url)) {
                ConcurrentMap<NotifyListener, ChildListener> listeners = 
                    zkListeners.get(url);
                // 如果之前该路径没有添加过listener，则创建一个map来放置listener
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
                    listeners = zkListeners.get(url);
                }
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) {
                    // 如果没有添加过对于子节点的listener，则创建,通知服务变化 回调NotifyListener
                    listeners.putIfAbsent(listener, (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)));

                    zkListener = listeners.get(listener);
                }
                zkClient.create(path, false);
                //添加path节点的当前节点及子节点监听，并且获取子节点信息
                //也就是dubbo://ip:port/...
                List<String> children = zkClient.addChildListener(path, zkListener);
                if (children != null) {
                    urls.addAll(toUrlsWithEmpty(url, path, children));
                }
            }
            //调用notify进行通知，对已经可用的列表进行通知
            notify(url, listener, urls);
        }
    } catch (Throwable e) {
        throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

FailbackRegistry.notify

调用 FailbackRegistry.notify，对参数进行判断。然后调用 AbstractRegistry.notify 方法

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) {
        // Record a failed registration request to a failed list, retry regularly
        addFailedNotified(url, listener, urls);
        logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
    }
}

AbstractRegistry.notify

这里面会针对每一个category，调用listener.notify进行通知，然后更新本地的缓存文件

protected void notify(URL url, NotifyListener listener, List<URL> urls) {
    //省略部分代码
    // keep every provider's category.
    Map<String, List<URL>> result = new HashMap<>();
    for (URL u : urls) {
        if (UrlUtils.isMatch(url, u)) {
            String category = u.getParameter(Constants.CATEGORY_KEY, 
                                             Constants.DEFAULT_CATEGORY);
            List<URL> categoryList = result.computeIfAbsent(category, 
                                                 k -> new ArrayList<>());
            categoryList.add(u);
        }
    }
    if (result.size() == 0) {
        return;
    }
    Map<String, List<URL>> categoryNotified = 
        notified.computeIfAbsent(url, u -> new ConcurrentHashMap<>());
    for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
        String category = entry.getKey();
        List<URL> categoryList = entry.getValue();
        categoryNotified.put(category, categoryList);
        listener.notify(categoryList);
        // We will update our cache file after each notification.
        // When our Registry has a subscribe failure due to network jitter, we can return at least the existing cache URL.
        saveProperties(url);
    }
}

消费端的 listener 是最开始传递过来的 RegistryDirectory，所以这里会触发 RegistryDirectory.notify

RegistryDirectory.notify

Invoker的网络连接以及后续的配置变更，都会调用这个notify方法

urls: zk的path数据，这里表示的是dubbo://

public synchronized void notify(List<URL> urls) {
    //对url列表进行校验、过滤，然后分成 config、router、provider 3个分组map
    Map<String, List<URL>> categoryUrls = urls.stream()
        .filter(Objects::nonNull)
        .filter(this::isValidCategory)
        .filter(this::isNotCompatibleFor26x)
        .collect(Collectors.groupingBy(url -> {
            if (UrlUtils.isConfigurator(url)) {
                return CONFIGURATORS_CATEGORY;
            } else if (UrlUtils.isRoute(url)) {
                return ROUTERS_CATEGORY;
            } else if (UrlUtils.isProvider(url)) {
                return PROVIDERS_CATEGORY;
            }
            return "";
        }));

    List<URL> configuratorURLs = 
        categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList());
    this.configurators = 
        Configurator.toConfigurators(configuratorURLs).orElse(this.configurators);
    // 如果router 路由节点有变化，则从新将router 下的数据生成router
    List<URL> routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, 
                                                     Collections.emptyList());
    toRouters(routerURLs).ifPresent(this::addRouters);

    // 获得provider URL，然后调用refreshOverrideAndInvoker进行刷新
    List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, 
                                                       Collections.emptyList());
    refreshOverrideAndInvoker(providerURLs);
}

refreshOverrideAndInvoker

逐个调用注册中心里面的配置，覆盖原来的url，组成最新的url 放入overrideDirectoryUrl 存储
根据 provider urls，重新刷新Invoker

private void refreshOverrideAndInvoker(List<URL> urls) {
    // mock zookeeper://xxx?mock=return null
    overrideDirectoryUrl();
    refreshInvoker(urls);
}

refreshInvoker

private void refreshInvoker(List<URL> invokerUrls) {
    Assert.notNull(invokerUrls, "invokerUrls should not be null");

    if (invokerUrls.size() == 1
        && invokerUrls.get(0) != null
        && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
        //省略部分代码，如果是空协议，则直接返回不允许访问
    } else {
        this.forbidden = false; // Allow to access
        //local reference
        Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; 

        if (invokerUrls.isEmpty()) {//如果url为空，则直接返回
            return; 
        }
        //根据provider url，生成新的invoker
        Map<String, Invoker<T>> newUrlInvokerMap = 
            toInvokers(invokerUrls);// 
        //转化为list
        List<Invoker<T>> newInvokers = 
            Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));

        routerChain.setInvokers(newInvokers);
        //如果服务配置了分组，则把分组下的provider包装成StaticDirectory,组成一个invoker
        //实际上就是按照group进行合并
        this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers;
        this.urlInvokerMap = newUrlInvokerMap;

        try {
            //旧的url 是否在新map里面存在，不存在，就是销毁url对应的Invoker
            //Close the unused Invoker
            destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap);  
        } catch (Exception e) {
            logger.warn("destroyUnusedInvokers error. ", e);
        }
    }
}

toInvokers

这个方法中有比较长的判断和处理逻辑，我们只需要关心invoker是什么时候初始化的就行。

这里用到了protocol.refer来构建了一个invoker

invoker = new InvokerDelegate<>(protocol.refer(serviceType, url), url, providerUrl);

构建完成之后，会保存在Map<String, Invoker<T>> urlInvokerMap 这个集合中

private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
    Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<>();
    // local reference
    Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; 

    Invoker<T> invoker = localUrlInvokerMap == 
        null ? null : localUrlInvokerMap.get(key);
    if (invoker == null) { // Not in the cache, refer again
        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) {
                invoker = new InvokerDelegate<>
                    (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) { // Put new invoker in cache
            newUrlInvokerMap.put(key, invoker);
        }
    } else {
        newUrlInvokerMap.put(key, invoker);
    }

    keys.clear();
    return newUrlInvokerMap;
}

protocol.refer

调用指定的协议来进行远程引用。protocol是一个Protocol$Adaptive类而真正的实现应该是：

ProtocolListenerWrapper(ProtocolFilterWrapper(QosProtocolWrapper(DubboProtocol.refer)

前面的包装过程，在服务发布的时候已经分析过了，我们直接进入 DubboProtocol.refer 方法

DubboProtocol.refer

优化序列化
构建DubboInvoker

在构建 DubboInvoker 时，会构建一个 ExchangeClient，通过 getClients(url) 方法，这里基本可以猜到到是服务的通信建立

@Override
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws 
    RpcException {
    optimizeSerialization(url);

    // create rpc invoker.
    DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
    invokers.add(invoker);

    return invoker;
}

getClients

这里面是获得客户端连接的方法

判断是否为共享连接，默认是共享同一个连接进行通信
是否配置了多个连接通道 connections，默认只有一个

private ExchangeClient[] getClients(URL url) {
    // whether to share connection

    boolean useShareConnect = false;

    int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
    List<ReferenceCountExchangeClient> shareClients = null;
    //如果没有配置连接数，则默认为共享连接
    if (connections == 0) {
        useShareConnect = true;

        /**
         * The xml configuration should have a higher priority than properties.
        */
        String shareConnectionsStr =  
            url.getParameter(Constants.SHARE_CONNECTIONS_KEY, (String) null);
        connections = Integer.parseInt(
            StringUtils.isBlank(shareConnectionsStr) ?  
            ConfigUtils.getProperty(Constants.SHARE_CONNECTIONS_KEY, 
                                    Constants.DEFAULT_SHARE_CONNECTIONS) : 
            shareConnectionsStr);
        shareClients = getSharedClient(url, connections);
    }

    ExchangeClient[] clients = new ExchangeClient[connections];
    for (int i = 0; i < clients.length; i++) {
        if (useShareConnect) {
            clients[i] = shareClients.get(i);

        } else {
            clients[i] = initClient(url);
        }
    }

    return clients;
}

getSharedClient

获得一个共享连接

private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
    String key = url.getAddress();
    List<ReferenceCountExchangeClient> clients = 
        referenceClientMap.get(key);
    //检查当前的key检查连接是否已经创建过并且可用，如果是，则直接返回并且增加连接的个数的统计
    if (checkClientCanUse(clients)) {
        batchClientRefIncr(clients);
        return clients;
    }
    //如果连接已经关闭或者连接没有创建过
    locks.putIfAbsent(key, new Object());
    synchronized (locks.get(key)) {
        clients = referenceClientMap.get(key);
        // 在创建连接之前，在做一次检查，防止连接并发创建
        if (checkClientCanUse(clients)) {
            batchClientRefIncr(clients);
            return clients;
        }

        // 连接数必须大于等于1
        connectNum = Math.max(connectNum, 1);

        //如果当前消费者还没有和服务端产生连接，则初始化
        if (CollectionUtils.isEmpty(clients)) {
            clients = buildReferenceCountExchangeClientList(url, connectNum);
            //创建clients之后，保存到map中
            referenceClientMap.put(key, clients);
        } else {//如果clients不为空，则从clients数组中进行遍历
            for (int i = 0; i < clients.size(); i++) {
                ReferenceCountExchangeClient referenceCountExchangeClient = 
                    clients.get(i);
                // 如果在集合中存在一个连接但是这个连接处于closed状态，则重新构建一个进行替换
                if (referenceCountExchangeClient == null || 
                    referenceCountExchangeClient.isClosed()) {
                    clients.set(i, buildReferenceCountExchangeClient(url));
                    continue;
                }
                //增加个数
                referenceCountExchangeClient.incrementAndGetCount();
            }
        }

        /**
          * I understand that the purpose of the remove operation here is to 
avoid the expired url key
          * always occupying this memory space.
          */
        locks.remove(key);

        return clients;
    }
}

buildReferenceCountExchangeClientList

根据连接数配置，来构建指定个数的链接。默认为1

private List<ReferenceCountExchangeClient> buildReferenceCountExchangeClientList(
    URL url, int connectNum) {
    List<ReferenceCountExchangeClient> clients = new CopyOnWriteArrayList<>();

    for (int i = 0; i < connectNum; i++) {
        clients.add(buildReferenceCountExchangeClient(url));
    }

    return clients;
}

private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url){
    ExchangeClient exchangeClient = initClient(url);
    return new ReferenceCountExchangeClient(exchangeClient);
}

initClient

终于进入到初始化客户端连接的方法了，猜测应该是根据url中配置的参数进行远程通信的构建

private ExchangeClient initClient(URL url) {

    // 获得连接类型
    String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
    //添加默认序列化方恨死
    url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
    // enable heartbeat by default
    //设置心跳时间
    url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, 
                                   String.valueOf(Constants.DEFAULT_HEARTBEAT));

    // 判断str是否存在于扩展点中，如果不存在则直接报错
    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).getSuppor tedExtensions(), " "));
    }

    ExchangeClient client;
    try {
        // 是否需要延迟创建连接，注意哦，这里的requestHandler是一个适配器
        if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
            client = new LazyConnectExchangeClient(url, requestHandler);
        } else {
            client = Exchangers.connect(url, requestHandler);
        }
    } catch (RemotingException e) {
        throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
    }

    return client;
}

Exchangers.connect

创建一个客户端连接

public static ExchangeClient connect(URL url, ExchangeHandler handler) throws 
    RemotingException {
    if (url == null) {
        throw new IllegalArgumentException("url == null");
    }
    if (handler == null) {
        throw new IllegalArgumentException("handler == null");
    }
    url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
    return getExchanger(url).connect(url, handler);
}

HeaderExchange.connect

主要关注transporters.connect

@Override
public ExchangeClient connect(URL url, ExchangeHandler handler) throws 
    RemotingException {
    return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
}

NettyTransport.connect

使用netty构建了一个客户端连接

@Override
public Client connect(URL url, ChannelHandler listener) throws 
    RemotingException {
    return new NettyClient(url, listener);
}

总结

我们讲到了RegistryProtocol.refer 过程中有一个关键步骤，即在监听到服务提供者url时触发RegistryDirectory.notify() 方法。

RegistryDirectory.notify() 方法调用refreshInvoker() 方法将服务提供者urls转换为对应的远程invoker ，最终调用到DubboProtocol.refer() 方法生成对应的DubboInvoker 。

DubboInvoker 的构造方法中有一项入参ExchangeClient[] clients ，即对应本文要讲的网络客户端Client 。

DubboInvoker就是通过调用client.request() 方法完成网络通信的请求发送和响应接收功能。

Client 的具体生成过程就是通过DubboProtocol 的initClient(URL url) 方法创建了一个 eaderExchangeClient。

dubbo进阶之服务注册消费

# Invoker是什么

# ProxyFacotory.getInvoker

# ProxyFactory

# ProxyFactory$Adaptive

# JavassistProxyFactory.getInvoker

# javassist生成的动态代理代码

# 服务注册流程

# 服务注册核心代码

# getRegistry

# RegistryFactory$Adaptive

# ZookeeperRegistryFactory

# createRegistry

# ZookeeperRegistry

# register

# registry.register(registedProviderUrl)

# FailbackRegistry.register

# ZookeeperRegistry.doRegister

# 思考服务消费应该要具备的逻辑

# 服务的消费

# ReferenceConﬁg.get

# init

# createProxy

# protocol.refer

# RegistryProtocol.refer

# doRefer

# Cluster是什么

# Cluster$Adaptive

# cluster.join

# proxyFactory.getProxy

# JavassistProxyFactory.getProxy

# proxy.getProxy

# 网络连接的建立

# RegistryDirectory.subscribe

# FailbackRegistry.subscribe

# ZookeeperRegistry.doSubscribe

# FailbackRegistry.notify

# AbstractRegistry.notify

# RegistryDirectory.notify

# refreshOverrideAndInvoker

# refreshInvoker

# toInvokers

# protocol.refer

# DubboProtocol.refer

# getClients

# getSharedClient

# buildReferenceCountExchangeClientList

# initClient

# Exchangers.connect

# HeaderExchange.connect

# NettyTransport.connect

# 总结

Invoker是什么

ProxyFacotory.getInvoker

ProxyFactory

ProxyFactory$Adaptive

JavassistProxyFactory.getInvoker

javassist生成的动态代理代码

服务注册流程

服务注册核心代码

getRegistry

RegistryFactory$Adaptive

ZookeeperRegistryFactory

createRegistry

ZookeeperRegistry

register

registry.register(registedProviderUrl)

FailbackRegistry.register

ZookeeperRegistry.doRegister

思考服务消费应该要具备的逻辑

服务的消费

ReferenceConﬁg.get

init

createProxy

protocol.refer

RegistryProtocol.refer

doRefer

Cluster是什么

Cluster$Adaptive

cluster.join

proxyFactory.getProxy

JavassistProxyFactory.getProxy

proxy.getProxy

网络连接的建立

RegistryDirectory.subscribe

FailbackRegistry.subscribe

ZookeeperRegistry.doSubscribe

FailbackRegistry.notify

AbstractRegistry.notify

RegistryDirectory.notify

refreshOverrideAndInvoker

refreshInvoker

toInvokers

protocol.refer

DubboProtocol.refer

getClients

getSharedClient

buildReferenceCountExchangeClientList

initClient

Exchangers.connect

HeaderExchange.connect

NettyTransport.connect

总结