Dubbo 服务调用源码学习（上）（六）

作者: jwfy | 来源:发表于2018-05-14 22:17 被阅读72次

笔记简述
本学习笔记就来学习dubbo中的消费方是如何和注册中心打交道的，以及如何实现和服务提供方的直连操作，这其中主要的是invoker生成操作
2018年05月23日22:10:03 添加了3.2.4 刷新invoker一小节，当订阅服务完成之后，客观情况下存在服务提供方发生变化的情况，这时候需要刷新RegistryDirectory中已存在的urlInvoker和methodInvoker信息，并且还需要保存对应的cache文件
更多内容可看[目录]Dubbo 源码学习

Dubbo 服务调用源码（上）学习（六）
1、Spring
2、ReferenceBean 介绍
3、源码学习
3.1、生成ReferenceBean的代理对象
3.2、生成Invoker
3.2.1、doRefer
3.2.2、注册到注册中心
3.2.3、订阅服务
3.2.4、刷新invoker
3.2.5、生成Invoker

1、Spring

dubbo的服务使用方是在xml配置了类似于<dubbo:reference interface="com.jwfy.dubbo.product.ProductService" id="productService" />的配置，意味着后续在spring中通过getBean('productService')就可以获取到远程代理对象。dubbo:reference 本身映射成为的bean是ReferenceBean，其会存储整个dubbo需要的各种信息，例如控制中心的注册地址，服务端的具体IO和端口等。

2、ReferenceBean 介绍

image

如上图就是ReferenceBean的类图，根据以往对spring的学习了解，有如下总结：

很清楚的认识到其是一个工厂Bean,后续需要getObject方法得到真正的对象（其实在这里不看源码，我们就应该能猜到常规做法是通过动态代理生成interface="com.jwfy.dubbo.product.ProductService"中接口对应的proxy对象），如果想获取ReferenceBean对象本身，则需要使用getBean("&productService")，如果这些结论还存在疑问，可以看看之前对spring的源码学习
通过InitializingBean的afterPropertiesSet方法去为当前的bean注入注册中心、均衡负责的方式、使用的协议等属性数据。

在这里生成代理对象是通过Java的动态代理方式，因为指明的是接口，（spring中默认的是如果通过接口生成代理对象，是使用动态代理，否则是使用cglib），那么根据对动态代理知识点的了解，InvocationHandler肯定是跑不掉的，通过invoke去调用执行函数的。

3、源码学习

3.1、生成ReferenceBean的代理对象

在上文说的getObject方法为入口打断点，最后可以追踪到ReferenceConfig类的createProxy方法为真正的生成代理对象的操作。

private T createProxy(Map<String, String> map) {
    URL tmpUrl = new URL("temp", "localhost", 0, map);
    // 这里的临时url可以是 temp://localhost?application=dubbo-consume&default.check=false
    //&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=15813&side=consumer&timestamp=1525921242794
    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 {
        if (url != null && url.length() > 0) { 
        // 用户指定URL，指定的URL可能是对点对直连地址，也可能是注册中心URL
        // 用户可以在dubbo:reference 中直接配置url参数，配置了就直接连接吧
            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);
            // 这个false的值含义是非服务提供方，获取到连接注册中心的配置
            // 切记！！！不是获取服务提供方的url属性，而是注册中心的配置
            // 生成的us是类似于registry://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?
            // application=dubbo-consume&client=zkclient&dubbo=2.5.3&group=dubbo-demo&owner=jwfy&pid=1527&registry=zookeeper&timestamp=1525943117155
            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.");
            }
        }
        
        // 在这里interfaceClass说的是 ==> com.jwfy.dubbo.product.ProductService
        // urls还是register协议
        
        // 总之下面这段代码是非常关键的，后面再补充

        if (urls.size() == 1) {
            invoker = refprotocol.refer(interfaceClass, urls.get(0));
            // refprotocol就是包装了两次的RegistryProtocol,直接生成对于的invoker
        } else {
            List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
            URL registryURL = null;
            for (URL url : urls) {
                // 在这里我们能够猜到其中一定有向注册中心订阅获取所有的服务提供方的信息
                // 并把获取到的信息返回生成一个invoker对象
                invokers.add(refprotocol.refer(interfaceClass, url));
                if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                    registryURL = url; // 用了最后一个registry url
                }
            }
            if (registryURL != null) { // 有 注册中心协议的URL
                // 对有注册中心的Cluster 只用 AvailableCluster
                URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME); 
                invoker = cluster.join(new StaticDirectory(u, invokers));
            }  else { // 不是 注册中心的URL
                invoker = cluster.join(new StaticDirectory(invokers));
            }
        }
    }

    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());
    }
    // 创建服务代理
    // 上面已经说了proxyFactory是StubProxyFactoryWrapper包装了JavassistProxyFactory类
    return (T) proxyFactory.getProxy(invoker);
}

JavassistProxyFactory 类

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

image

3.2、生成Invoker

Invoker是对可执行对象的引用，需要明确可引用的具体位置（服务端的明确IP和端口等信息）

现在我们已经拿到了当前服务的注册中心的配置，那么接下来就需要连接到注册中心，并获取到可以调用的机器情况（现实开发中，分布式系统基本上都存在多个机器信息），并组合成为需要的invoker。

下面这个代码段就是生产具体的Invoke操作，接下来好好分析一下

invoker = refprotocol.refer(interfaceClass, urls.get(0));
// refprotocol就是包装了两次的RegistryProtocol,直接生成对于的invoker
// url是registry开头的协议，在参数中包含了消息等协议和其参数
// interfaceClass是接口类

在分析源码之前，如果换成我们，我们会完成什么操作呢？

向注册中心订阅消费者，这样通过dubbo-admin就可以观察现有的生产者和消费者
从主存中心获取生产者的信息
类似均衡负责的操作，选择合适的生产者
直连生产者，获取结果

RegistryProtocol 类

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
   url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
   // 替换协议，一般情况下都是替换成为zookeeper
   Registry registry = registryFactory.getRegistry(url);
   // 这一步是获取到注册中心的配置信息，这里和服务暴露的操作一致
   // 其中的key是zookeeper://127.0.0.1:2182/dubbo-demo/com.alibaba.dubbo.registry.RegistryService
   // 那么同一个jvm内的key其实都是一致的，所以其连接到控制中心的数据也一致
   // 如果没有连接，则需要创建一个新的链接实体，其中会把当前的信息存储到文件中
   // 也会有对应的监听者等，然后真正的调用连接zk操作，确保zk是真实存在的
   if (RegistryService.class.equals(type)) {
    // 如果类型是RegistryService，则使用getInvoke操作直接拼接生成一个invoke对象
    // 这个操作和服务提供方生成invoker的方式一致
       return proxyFactory.getInvoker((T) registry, type, url);
   }

   // group="a,b" or group="*"
   Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
   // 这个qs包含了需要生成invoker的所有参数，例如接口名称，函数名，所述范畴（消费者）等信息
   String group = qs.get(Constants.GROUP_KEY);
   // 查看分组信息
   if (group != null && group.length() > 0 ) {
       if ( ( Constants.COMMA_SPLIT_PATTERN.split( group ) ).length > 1
               || "*".equals( group ) ) {
           return doRefer( getMergeableCluster(), registry, type, url );
       }
   }
   
   // 看有分组和没分组，其实就是集群cluster不一样
   return doRefer(cluster, registry, type, url);
}

private Cluster getMergeableCluster() {
   return ExtensionLoader.getExtensionLoader(Cluster.class).getExtension("mergeable");
   // 生成MergeableCluster 集群
   // 如下代码Cluster$Adpative类则就是cluster，默认的是为FailoverCluster集群
}

Cluster$Adpative 类

package com.alibaba.dubbo.rpc.cluster;

import com.alibaba.dubbo.common.extension.ExtensionLoader;

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

3.2.1、doRefer

经过上面的操作，现在来到了doRefer函数操作，其结果返回的就是invoker对象

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    // cluster 集群，目前可能为FailoverCluster，如果在有分组的情况下则是MergeableCluster
    // registry 注册中心信息
    // type 就是上面说的
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
    if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
            && url.getParameter(Constants.REGISTER_KEY, true)) {
            // 接口名字有意义而且是注册协议
            // 把当前url信息当做消费者节点注册到注册中心中区
        registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                Constants.CHECK_KEY, String.valueOf(false)));
    }
    directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
            Constants.PROVIDERS_CATEGORY 
            + "," + Constants.CONFIGURATORS_CATEGORY 
            + "," + Constants.ROUTERS_CATEGORY));
        
    // 订阅服务，并生成invoker对象
    return cluster.join(directory);
}

在上面一笔带过了如何注册、订阅、生成invoke的，接下来依次拆分各个细节

3.2.2、注册到注册中心

registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY, Constants.CHECK_KEY, String.valueOf(false)));

registry是ZookeeperRegistry对象
subscribeUrl是consumer://192.168.10.123/com.jwfy.dubbo.product.ProductService?application=dubbo-consume&default.check=false&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=1196&side=consumer&timestamp=1526204222984

表示是一个消费者

进入到FailbackRegistry类

public void register(URL url) {
    super.register(url);
    // 会把当前的url添加到registered集合中，表示该url注册了
    failedRegistered.remove(url);
    failedUnregistered.remove(url);
    // 既然都要注册了，肯定从失败的集合和取消注册的集合中移除掉
    try {
        // 向服务器端发送注册请求，也是真正的开始注册操作
        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);
        }

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

// 关于failedRegistered的重试操作，在构造函数中有
// 开启定时任务运行的操作，默认时间是5秒执行一次
    this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
        public void run() {
            // 检测并连接注册中心
            try {
                retry();
                // 里面肯定有遍历failedRegistered集合的doRegister操作（事实上确实有）
            } catch (Throwable t) { // 防御性容错
                // 个人觉得这个代码写的很好，因为突出一个点，什么时候抛出异常，什么时候处理异常
                // 这点问题上，自己曾经踩了很多坑，也发现很多人也有这个毛病
                logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
            }
        }
    }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);

在doRegister操作中，是利用zk的API存储如下的path
/dubbo-jwfy/com.jwfy.dubbo.product.ProductService/consumers/consumer%3A%2F%2F192.168.10.123%2Fcom.jwfy.dubbo.product.ProductService%3Fapplication%3Ddubbo-consume%26category%3Dconsumers%26check%3Dfalse%26default.check%3Dfalse%26dubbo%3D2.5.3%26interface%3Dcom.jwfy.dubbo.product.ProductService%26methods%3Dprint%2CgetStr%26owner%3Djwfy%26pid%3D1196%26side%3Dconsumer%26timestamp%3D1526204222984

如下图，在调用doRegister前后zk注册中心节点的情况，很明显已经注册成功

image

3.2.3、订阅服务

服务订阅说通俗些就是获取zk中需要的节点信息，本例中是获取生产者的连接信息

directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
            Constants.PROVIDERS_CATEGORY 
            + "," + Constants.CONFIGURATORS_CATEGORY 
            + "," + Constants.ROUTERS_CATEGORY));
// 这个url 添加了生产者、配置、路由三个节点的配置参数

url是consumer://192.168.10.123/com.jwfy.dubbo.product.ProductService?application=dubbo-consume&category=providers,configurators,routers&default.check=false&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=1196&side=consumer&timestamp=1526204222984
directory 是RegistryDirectory，其中参数registry就是上面的注册中心ZookeeperRegistry的配置

FailbackRegistry 类

public void subscribe(URL url, NotifyListener listener) {
    super.subscribe(url, listener);
    // 会设置registry的subscribed信息，其中subscribed是一个Map<Url,Set<NotifyListener>>的容器
    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);
    }
}

来到了doSubscribe方法,在这里我们将会了解到如何从注册中心获取到生产者的连接信息的

protected void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
            // Constants.ANY_VALUE是*
            // url.getServiceInterface 是 com.jwfy.dubbo.product.ProductService
            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) {
                            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) {
                anyServices.addAll(services);
                for (String service : services) {
                    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);
                // zkListeners 是 Map<URL, Map<NotifyListener, ChildListener>>
                // 管理url与其对应的监听者和zk监听者的映射关系
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                    listeners = zkListeners.get(url);
                }
                ChildListener zkListener = listeners.get(listener);
                // 从集合中获取到zk的监听者
                if (zkListener == null) {
                    // 如果没有，则需要手动创建一个新的，其中包含了自定义实现的更新子节点的函数操作
                    // 里面会调用notify订阅方法（这个方法很重要，由zkClient主动调用childChanged方法）
                    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);
                // 创建永久节点，此时的path可能为/dubbo-jwfy/com.jwfy.dubbo.product.ProductService/providers
                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);
    }
}

AbstractZookeeperClient 类

public List<String> addChildListener(String path, final ChildListener listener) {
    ConcurrentMap<ChildListener, TargetChildListener> listeners = childListeners.get(path);
    // 从zkClient本身的childListeners获取path的监听者信息
    if (listeners == null) {
        childListeners.putIfAbsent(path, new ConcurrentHashMap<ChildListener, TargetChildListener>());
        listeners = childListeners.get(path);
    }
    TargetChildListener targetListener = listeners.get(listener);
    // 再获取目标节点信息（这个就包含了生产者的信息了）
    if (targetListener == null) {
        listeners.putIfAbsent(listener, createTargetChildListener(path, listener));
        targetListener = listeners.get(listener);
    }
    return addTargetChildListener(path, targetListener);
}


public List<String> addTargetChildListener(String path, final IZkChildListener listener) {
    return client.subscribeChildChanges(path, listener);
    // 这一步就会深入到zkClientjar包内进行最后的_zk.getChildren操作
    // 返回的List<String> 就是对应的path路径的内容
    
    // 此时path 是 /dubbo-jwfy/com.jwfy.dubbo.product.ProductService/providers
    // 返回的内容 是 dubbo%3A%2F%2F192.168.10.123%3A20880%2Fcom.jwfy.dubbo.product.ProductService
    // %3Fanyhost%3Dtrue%26application%3Ddubbo-demo%26default.loadbalance%3Drandom%26
    // dubbo%3D2.5.3%26interface%3Dcom.jwfy.dubbo.product.ProductService%26
    // methods%3Dprint%2CgetStr%26owner%3Djwfy%26pid%3D1081%26side%3Dprovider%26
    // timestamp%3D1526198287386%26token%3Dfdfdf
}

现在完成了和注册中心的操作了，通过path顺利拿到生产者的信息，如果仔细观察上述的参数信息，会发现pid是1081，再看看jps显示的进程号,如下图，恰好说明获取到的生产者信息是对的

image

接着来到toUrlsWithEmpty函数，如果有仔细观察这个方法会发现，参数列表都改成了(URL consumer, String path, List<String> providers)这已经很明确的告诉我们，第一个参数是消费者的url，第二个是当前zk的path信息，第三个是获取到的生产者列表信息（为啥是列表呢？因为生产者可以是多个，而且存在多个的情况下，后续均衡负责还需要选择一个可用的生产者进行网络信息交互操作）

当然toUrlsWithEmpty函数主要是进行生产者和消费者的url信息对比操作，如果没有合适的url则添加一个empty协议的url信息（后期就是通过这个empty判断是否存在有用的生产者，日常开发中的无效黑白名单的错误就产生在这里）

3.2.4、刷新invoker

紧接着来到了notify方法，如下图的具体各个参数具体值

image

紧接着来到了AbstractRegistry类

protected void notify(URL url, NotifyListener listener, List<URL> urls) {
    // url 是注册到注册到注册中心的url
    // listener是监听器，其实就是RegistryDirectory
    // urls 是 从注册中心获取到到的 服务提供方的url集合
    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)) {
           // isMatch(URL consumerUrl, URL providerUrl) 函数的参数
           // 已经非常清楚的说明了url是服务调用方的url，u是服务提供方的url
           // 也和我们上面的描述是一致的
           // 对两个url的接口、类目、enable、分组、版本号、classifier等内容进行匹配
           // 如果匹配合适，就认为为true
           // 没有针对协议进行匹配操作
            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);
        }
    }
    // 按照类目进行分组成为一个map
    
    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);
    }
    for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
        String category = entry.getKey();
        List<URL> categoryList = entry.getValue();
        categoryNotified.put(category, categoryList);
        // 替换notified.get(url)的map信息
        saveProperties(url);
        // 初次看，这个url就是上面的服务调用方的url信息，没必要每次都保存吧
        // 进入到这个函数可以发现，他会每次又调用notified.get(url)去获取最新的map数据，默认为异步保存数据
        listener.notify(categoryList);
        // 接着来到了RegistryDirectory的notify方法
    }
}

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);
        }
    }
    // invokerUrls只能是服务提供方目录去刷新已存的invoke
    refreshInvoker(invokerUrls);
}

private void refreshInvoker(List<URL> invokerUrls){
    if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
            && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
            // empty协议，而且只有1个
            // 需要注意到这个方法只有服务提供方的url才可以被调用
            // 服务提供方只包含了一个empty协议的无效url，设置forbidden=true
            // 这个true就是后面dubbo中经常出现的黑白名单错误情况
        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;
        }
        // 后面的操作就是去刷新现存的invoker列表
        
        Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表
        // 这个invoker是InvokerDelegete类，是包装了一层的InvokerWrapper
        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);
        }
    }
}

这个函数具体的操作包含了
1、更新ZookeeperRegistry中的notify参数信息（把生产者、配置、路由等url信息存入其中）
2、保存节点信息文档到dubbo的cache文件中
3、刷新生成invoke的数据信息

到此整个的订阅服务的操作就完成了

3.2.5、生成Invoker

现在就剩下最关键的一句话cluster.join(directory)，会层层包装，最后形成的invoker如图所示

image

在directory中包含了所有的注册信息，在后面的真正的函数调用其实也是通过invoker.invoker去调用执行

InvokerInvocationHandler 类

被包装的类通过动态代理反射时，内嵌入的InvocationHandler类，具体方法调用则会进入到invoke方法中

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(invoker, args);
    }
    if ("toString".equals(methodName) && parameterTypes.length == 0) {
        return invoker.toString();
    }
    if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
        return invoker.hashCode();
    }
    if ("equals".equals(methodName) && parameterTypes.length == 1) {
        return invoker.equals(args[0]);
    }
    return invoker.invoke(new RpcInvocation(method, args)).recreate();
}

image

由于后续的流程太多，接下来的内容分为一小节学习，到此invoker就生成了。

Dubbo 服务调用源码学习（上）（六）

1、Spring

2、ReferenceBean 介绍

3、源码学习

3.1、生成ReferenceBean的代理对象

3.2、生成Invoker

3.2.1、doRefer

3.2.2、注册到注册中心

3.2.3、订阅服务

3.2.4、刷新invoker

3.2.5、生成Invoker

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Dubbo

Dubbo 源码学习

dubbo

Dubbo 服务调用 源码学习（上）（六）

1、Spring

2、ReferenceBean 介绍

3、源码学习

3.1、生成ReferenceBean的代理对象

3.2、生成Invoker

3.2.1、doRefer

3.2.2、注册到注册中心

3.2.3、订阅服务

3.2.4、刷新invoker

3.2.5、生成Invoker

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Dubbo

Dubbo 源码学习

dubbo

Dubbo 服务调用源码学习（上）（六）