zoukankan      html  css  js  c++  java
  • Dubbo消费端Reference过程(Version2.7.3)

    前言

    前面Dubbo服务暴露完成了两个任务:1. 启动本地服务器。2. 将服务注册到注册中心。

    服务暴露开始于ServiceBean,那么与之对应的,服务引用开始于ReferenceBean。

    入口

    入口有两个,都在ReferenceBean中

    // 懒汉式(在 ReferenceBean 对应的服务被注入到其他类中时)
    public Object getObject() {
        return get();
    }
    
    public void afterPropertiesSet() throws Exception {
        // 省略无关代码
    
        // 饿汉式(在 Spring 容器调用 ReferenceBean 的 afterPropertiesSet 方法时)
        if (shouldInit()) {
            getObject();
        }
    }

    默认不会走afterPropertiesSet方法里的getObject方法,在启动容器注入依赖的时候,会走ReferenceBean的父类ReferenceConfig#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方法(从这里开始,Apache的实现和alibaba的实现就完全不一样了,但是大体思路还是一样的

    // org.apache.dubbo.config.ReferenceConfig#init
    private void init() {
        // 已经初始化就返回
        if (initialized) {
            return;
        }
        // 检查local或者stub 本地存根 是否有配置一个当前接口类型参数的构造函数
        checkStubAndLocal(interfaceClass);
        // 检查mock配置
        checkMock(interfaceClass);
        Map<String, String> map = new HashMap<String, String>();
    
        map.put(SIDE_KEY, CONSUMER_SIDE);
    
        appendRuntimeParameters(map);
        // 不是泛型
        if (!isGeneric()) {
            // 获取接口版本:1.0
            String revision = Version.getVersion(interfaceClass, version);
            if (revision != null && revision.length() > 0) {
                map.put(REVISION_KEY, revision);
            }
            // 获取接口方法列表:{"hello"}
            String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
            if (methods.length == 0) {
                logger.warn("No method found in service interface " + interfaceClass.getName());
                map.put(METHODS_KEY, ANY_VALUE);
            } else {
                map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), COMMA_SEPARATOR));
            }
        }
        map.put(INTERFACE_KEY, interfaceName);
        appendParameters(map, metrics);
        appendParameters(map, application);
        appendParameters(map, module);
        // remove 'default.' prefix for configs from ConsumerConfig
        // appendParameters(map, consumer, Constants.DEFAULT_KEY);
        appendParameters(map, consumer);
        appendParameters(map, this);
        Map<String, Object> attributes = null;
        // 查看有没有方法配置(config)
        if (CollectionUtils.isNotEmpty(methods)) {
            attributes = new HashMap<String, Object>();
            for (MethodConfig methodConfig : methods) {
                appendParameters(map, methodConfig, methodConfig.getName());
                String retryKey = methodConfig.getName() + ".retry";
                if (map.containsKey(retryKey)) {
                    String retryValue = map.remove(retryKey);
                    if ("false".equals(retryValue)) {
                        map.put(methodConfig.getName() + ".retries", "0");
                    }
                }
                attributes.put(methodConfig.getName(), convertMethodConfig2AyncInfo(methodConfig));
            }
        }
        // 获取注册地址
        String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY);
        if (StringUtils.isEmpty(hostToRegistry)) {
            // 本地地址作为注册地址
            hostToRegistry = NetUtils.getLocalHost();
        } else if (isInvalidLocalHost(hostToRegistry)) {
            throw new IllegalArgumentException("Specified invalid registry ip from property:" + DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
        }
        map.put(REGISTER_IP_KEY, hostToRegistry);
    
        // *** 不仅仅是创建代理对象,还要创建Invoker,订阅服务,启动客户端等
        ref = createProxy(map);
    
        String serviceKey = URL.buildKey(interfaceName, group, version);
        ApplicationModel.initConsumerModel(serviceKey, buildConsumerModel(serviceKey, attributes));
        initialized = true;
    }

    上面代码做了什么事情呢

    1. 收集配置信息,比如注册地址、接口名称、接口方法、版本信息等

    2. 执行createProxy方法

    private T createProxy(Map<String, String> map) {
        // 是否从当前JVM中引用目标服务的实例(如果指定了injvm,则true;如果指定了url,则认为是远程调用,false;如果没有指定,但是目标服务是在相同的JVM中提供的,进行本地调用,true)
        if (shouldJvmRefer(map)) {
            // 生成本地引用 URL,协议为 injvm
            URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
            invoker = REF_PROTOCOL.refer(interfaceClass, url);
            if (logger.isInfoEnabled()) {
                logger.info("Using injvm service " + interfaceClass.getName());
            }
        } else {
            // 引用重试初始化会把url添加到url集合里,最终导致内存溢出。
            urls.clear(); 
            // url 不为空,表明用户可能想进行点对点调用
            if (url != null && url.length() > 0) { 
                // 当需要配置多个 url 时,可用分号进行分割,这里会进行切分
                String[] us = 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 = url.setPath(interfaceName);
                        }
                        // 检测 url 协议是否为 registry,若是,表明用户想使用指定的注册中心
                        if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                            urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                        } else {
                            urls.add(ClusterUtils.mergeUrl(url, map));
                        }
                    }
                }
            } else { // 从注册中心的配置中组装URL
                // 如果协议不是injvm
                if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){
                    // 检查注册中心
                    checkRegistry();
                    // 加载注册中心 url
                    List<URL> us = loadRegistries(false);
                    if (CollectionUtils.isNotEmpty(us)) {
                        for (URL u : us) {
                            URL monitorUrl = loadMonitor(u);
                            if (monitorUrl != null) {
                                map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                            }
                            urls.add(u.addParameterAndEncoded(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.");
                    }
                }
            }
    
            // 单个注册中心或服务提供者
            if (urls.size() == 1) {
                // ***调用 RegistryProtocol 的 refer 构建 Invoker 实例
                invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
            } else {
                // 多个注册中心或多个服务提供者,或者两者混合
                List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
                URL registryURL = null;
                for (URL url : urls) {
                    invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
                    if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                        registryURL = url; // use last registry url
                    }
                }
                if (registryURL != null) { // registry url is available
                    // use RegistryAwareCluster only when register's CLUSTER is available
                    URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME);
                    // The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
                    invoker = CLUSTER.join(new StaticDirectory(u, invokers));
                } else { // not a registry url, must be direct invoke.
                    invoker = CLUSTER.join(new StaticDirectory(invokers));
                }
            }
        }
    
        if (shouldCheck() && !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());
        }
        /**
         * @since 2.7.0
         * ServiceData Store
         */
        MetadataReportService metadataReportService = null;
        if ((metadataReportService = getMetadataReportService()) != null) {
            URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
            metadataReportService.publishConsumer(consumerURL);
        }
        // 最后创建服务代理对象
        return (T) PROXY_FACTORY.getProxy(invoker);
    }

    这么多,逻辑如下

    1. 收集服务引用的配置信息,比如注册地址、接口名称、接口方法、版本信息等
    2. 调用createProxy
      - 首先根据配置检查是否为本地调用,若是,则调用 InjvmProtocol 的 refer 方法生成 InjvmInvoker 实例。
      - 若不是,则读取直连配置项,或注册中心 url,并将读取到的 url 存储到 urls 中。
      - 若 urls 元素数量为1,则直接通过 Protocol 自适应拓展类构建 Invoker 实例接口。
      - 若 urls 元素数量大于1,即存在多个注册中心或服务直连 url,此时先根据 url 构建 Invoker。然后再通过 Cluster 合并多个 Invoker。
      [重点]创建 Invoker
        - Invoker 是由 Protocol 实现类 RegistryProtocol 构建而来。

      [重点]最后调用 ProxyFactory 生成代理类。

    [重点]创建 Invoker

    // org.apache.dubbo.registry.integration.RegistryProtocol#refer
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        url = URLBuilder.from(url)
                // 给注册中心URL设置协议,比如:zookeeper
                .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);
        }
    
        // 根据url解析查询参数,比如:"register.ip" -> "10.204.241.39"  "interface" -> "com.demo.common.HelloService"  键值对
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
        // 获取"group"对应的值,未设置为null
        String group = qs.get(GROUP_KEY);
        if (group != null && group.length() > 0) {
            if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
                return doRefer(getMergeableCluster(), registry, type, url);
            }
        }
        // 执行doRefer方法(这种方法风格和Spring一样,比如Spring里getBean调用doGetBean等)
        return doRefer(cluster, registry, type, url);
    }
    
    
    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        // 注册中心URL:zookeeper://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=dubbo-consumer&client=curator&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=108324&qos.enable=false&release=2.7.3&timeout=5000&timestamp=1584672980993
        // RegistryDirectory对象,基于注册中心动态感知服务提供者的变化
        RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // 获取URL查询参数键值对
        Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
        // 组装订阅URL
        // consumer://10.204.241.39/com.demo.common.HelloService?application=dubbo-consumer&dubbo=2.0.2&interface=com.demo.common.HelloService&lazy=false&methods=hello&pid=108324&qos.enable=false&release=2.7.3&revision=1.0&side=consumer&sticky=false&timestamp=1584672980764&version=1.0
        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;
    }

    上面完成了

    1. refer方法主要完成:根据注册中心URL获取注册中心实例。

    2. doRefer方法:

      - 获取RegistryDirectory对象,基于注册中心动态感知服务提供者的变化。

      * 执行subscribe方法。

      * 创建Invoker并返回

    执行subscribe方法

    // org.apache.dubbo.registry.integration.RegistryDirectory#subscribe
    public void subscribe(URL url) {
        // 设置消费者URL
        setConsumerUrl(url);
        CONSUMER_CONFIGURATION_LISTENER.addNotifyListener(this);
        serviceConfigurationListener = new ReferenceConfigurationListener(this, url);
        // 这里的registry是doRefer中设置的,所以这里执行的是注册中心的subscribe方法
        registry.subscribe(url, this);
    }
    // org.apache.dubbo.registry.support.FailbackRegistry#subscribe
    // Failback:出故障时自动恢复
    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 (CollectionUtils.isNotEmpty(urls)) {
                notify(url, listener, urls);
                logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + getUrl().getParameter(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);
        }
    }

    上面仅仅是做了一些属性设置,重点在于doSubscribe方法

    ZookeeperRegistry

    // org.apache.dubbo.registry.zookeeper.ZookeeperRegistry#doSubscribe
    public void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            // 处理所有Service层发起的订阅,例如监控中心的订阅
            if (ANY_VALUE.equals(url.getServiceInterface())) {
                // 获得根目录
                String root = toRootPath();
                // 获得url对应的监听器集合
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                // 不存在就创建监听器集合
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
                    listeners = zkListeners.get(url);
                }
                // 获得节点监听器
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) {
                    listeners.putIfAbsent(listener, (parentPath, currentChilds) -> {
                        // 遍历现有的节点,如果现有的服务集合中没有该节点,则加入该节点,然后订阅该节点
                        for (String child : currentChilds) {
                            child = URL.decode(child);
                            if (!anyServices.contains(child)) {
                                anyServices.add(child);
                                subscribe(url.setPath(child).addParameters(INTERFACE_KEY, child,
                                        Constants.CHECK_KEY, String.valueOf(false)), listener);
                            }
                        }
                    });
                    // 重新获取,为了保证一致性
                    zkListener = listeners.get(listener);
                }
                // 创建service节点
                zkClient.create(root, false);
                // 向zookeeper的service节点发起订阅,获得Service接口全名数组
                List<String> services = zkClient.addChildListener(root, zkListener);
                if (CollectionUtils.isNotEmpty(services)) {
                    // 遍历Service接口全名数组
                    for (String service : services) {
                        service = URL.decode(service);
                        anyServices.add(service);
                        // 发起该service层的订阅
                        subscribe(url.setPath(service).addParameters(INTERFACE_KEY, service,
                                Constants.CHECK_KEY, String.valueOf(false)), listener);
                    }
                }
            } else {
                // 处理指定 Service 层的发起订阅,例如服务消费者的订阅
                List<URL> urls = new ArrayList<>();
                for (String path : toCategoriesPath(url)) {
                    // path的可能值:
                    // 0 = "/dubbo/com.demo.common.HelloService/providers"
                    // 1 = "/dubbo/com.demo.common.HelloService/configurators"
                    // 2 = "/dubbo/com.demo.common.HelloService/routers"
                    // 获得监听器集合
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    if (listeners == null) {
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
                        listeners = zkListeners.get(url);
                    }
                    // 获得节点监听器
                    ChildListener zkListener = listeners.get(listener);
                    if (zkListener == null) {
                        listeners.putIfAbsent(listener, (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)));
                        zkListener = listeners.get(listener);
                    }
                    // 创建type节点
                    zkClient.create(path, false);
                    // 向zookeeper的type节点发起订阅
                    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

    // org.apache.dubbo.registry.support.FailbackRegistry#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
            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);
        }
    }
    
    protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
        // 交给父类
        super.notify(url, listener, urls);
    }

    AbstractRegistry

    // org.apache.dubbo.registry.support.AbstractRegistry#notify(URL, NotifyListener, List<URL>)
    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 ((CollectionUtils.isEmpty(urls))
                && !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);
        }
        // 按照类别存储URL,比如:"providers" "routers" "configurators" 三种类型作为key,value是对应的URL列表
        Map<String, List<URL>> result = new HashMap<>();
        for (URL u : urls) {
            if (UrlUtils.isMatch(url, u)) {
                String category = u.getParameter(CATEGORY_KEY, 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<>());
        // 遍历分好类别的URL
        for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
            String category = entry.getKey();
            List<URL> categoryList = entry.getValue();
            categoryNotified.put(category, categoryList);
            // 通知URL的变化(告诉观察者)
            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);
        }
    }

    这里注意,是通知逻辑的主要实现(设计模式为观察者模式),下面进入notify方法

    RegistryDirectory

    // org.apache.dubbo.registry.integration.RegistryDirectory#notify
    public synchronized void notify(List<URL> urls) {
        // 先把URL分好类
        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 "";
                }));
        // 处理类型为configurators的URL
        List<URL> configuratorURLs = categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList());
        this.configurators = Configurator.toConfigurators(configuratorURLs).orElse(this.configurators);
        // 处理类型的routers的URL
        List<URL> routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, Collections.emptyList());
        toRouters(routerURLs).ifPresent(this::addRouters);
    
        // 处理类型为providers的URL
        List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());
        refreshOverrideAndInvoker(providerURLs);
    }

    这里主要是收到URL的变化而响应的处理过程。我们知道Invoker 是 Dubbo 的核心模型,代表一个可执行体。那么我们检测到服务地址变化了,那么对应的Invoker肯定要一起改变。

    在Debug过程遇到三种格式的URL:

    empty://10.204.241.39/com.demo.common.HelloService?application=dubbo-consumer&category=routers&dubbo=2.0.2&interface=com.demo.common.HelloService&lazy=false&methods=hello&pid=10336&qos.enable=false&release=2.7.3&revision=1.0&side=consumer&sticky=false&timestamp=1584948513183&version=1.0

    empty://10.204.241.39/com.demo.common.HelloService?application=dubbo-consumer&category=configurators&dubbo=2.0.2&interface=com.demo.common.HelloService&lazy=false&methods=hello&pid=10336&qos.enable=false&release=2.7.3&revision=1.0&side=consumer&sticky=false&timestamp=1584948513183&version=1.0

    dubbo://10.204.241.14:20880/com.demo.common.HelloService?actives=5&anyhost=true&application=service-provider&bean.name=providers:dubbo:com.demo.common.HelloService:1.0&default.deprecated=false&default.dynamic=false&default.register=true&deprecated=false&dubbo=2.0.2&dynamic=false&generic=false&interface=com.demo.common.HelloService&methods=hello&pid=49328&register=true&release=2.7.1&retries=3&revision=1.0&side=provider&timeout=3000&timestamp=1563100343715&version=1.0

    来看刷新Invoker方法

    private void refreshOverrideAndInvoker(List<URL> urls) {
        // mock zookeeper://xxx?mock=return null
        overrideDirectoryUrl();
        refreshInvoker(urls);
    }
    
    
    private void refreshInvoker(List<URL> invokerUrls) {
        Assert.notNull(invokerUrls, "invokerUrls should not be null");
    
        // 如果地址只有一个并且协议参数值为empty,则禁止访问
        if (invokerUrls.size() == 1
                && invokerUrls.get(0) != null
                && EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
            this.forbidden = true; // Forbid to access
            this.invokers = Collections.emptyList();
            routerChain.setInvokers(this.invokers);
            destroyAllInvokers(); // Close all invokers
        } else {
            this.forbidden = false; // Allow to access
            // 临时存储旧的UrlInvokerMap
            Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
            if (invokerUrls == Collections.<URL>emptyList()) {
                invokerUrls = new ArrayList<>();
            }
            // 如果invokerUrls为空并且缓存不为空
            if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
                invokerUrls.addAll(this.cachedInvokerUrls);
            } else {
                // 否则,添加到缓存
                this.cachedInvokerUrls = new HashSet<>();
                this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
            }
            // 再次判断是否为空,为空则返回
            if (invokerUrls.isEmpty()) {
                return;
            }
            // 能走到这里的地址格式:
            // dubbo://10.204.241.14:20880/com.demo.common.HelloService?...
            // 生成Map,key为url,value为Invoker
            Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
    
            /**
             * If the calculation is wrong, it is not processed.
             *
             * 1. The protocol configured by the client is inconsistent with the protocol of the server.
             *    eg: consumer protocol = dubbo, provider only has other protocol services(rest).
             * 2. The registration center is not robust and pushes illegal specification data.
             *
             */
            if (CollectionUtils.isEmptyMap(newUrlInvokerMap)) {
                logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls
                        .toString()));
                return;
            }
    
            // 获取Invoker列表
            List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));
            // pre-route and build cache, notice that route cache should build on original Invoker list.
            // toMergeMethodInvokerMap() will wrap some invokers having different groups, those wrapped invokers not should be routed.
            routerChain.setInvokers(newInvokers);
            this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers;
            this.urlInvokerMap = newUrlInvokerMap;
    
            try {
                // 销毁未使用的Invoker
                destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
            } catch (Exception e) {
                logger.warn("destroyUnusedInvokers error. ", e);
            }
        }
    }
    
    private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
        Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<>();
        if (urls == null || urls.isEmpty()) {
            return newUrlInvokerMap;
        }
        Set<String> keys = new HashSet<>();
        String queryProtocols = this.queryMap.get(PROTOCOL_KEY);
        // 遍历URL
        for (URL providerUrl : urls) {
            // 如果在引用一端配置了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;
                }
            }
            // 如果URL的protocol参数为empty,则跳过此次循环
            if (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:
            // dubbo://10.204.241.14:20880/com.demo.common.HelloService?actives=5&anyhost=true&application=service-provider&bean.name=providers:dubbo:com.demo.common.HelloService:1.0&default.deprecated=false&default.dynamic=false&default.register=true&deprecated=false&dubbo=2.0.2&dynamic=false&generic=false&interface=com.demo.common.HelloService&methods=hello&pid=49328&register=true&release=2.7.1&retries=3&revision=1.0&side=provider&timeout=3000&timestamp=1563100343715&version=1.0
            // dubbo://10.204.241.14:20880/com.demo.common.HelloService?actives=5&anyhost=true&application=dubbo-consumer&bean.name=providers:dubbo:com.demo.common.HelloService:1.0&check=false&default.deprecated=false&default.dynamic=false&default.register=true&deprecated=false&dubbo=2.0.2&dynamic=false&generic=false&interface=com.demo.common.HelloService&lazy=false&methods=hello&pid=110788&qos.enable=false&register=true&register.ip=10.204.241.39&release=2.7.1&remote.application=service-provider&retries=3&revision=1.0&side=consumer&sticky=false&timeout=3000&timestamp=1563100343715&version=1.0
            URL url = mergeUrl(providerUrl);
    
            // 把URL当做key
            String key = url.toFullString(); // The parameter urls are sorted
            if (keys.contains(key)) { // Repeated url
                continue;
            }
            // 缓存key
            keys.add(key);
            // Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again
            Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
            // 查看这个key是否有对应的Invoker
            Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
            // 缓存中没有对应的Invoker,则会创建
            if (invoker == null) {
                try {
                    boolean enabled = true;
                    if (url.hasParameter(DISABLED_KEY)) {
                        enabled = !url.getParameter(DISABLED_KEY, false);
                    } else {
                        enabled = url.getParameter(ENABLED_KEY, true);
                    }
                    if (enabled) {
                        // 创建Invoker,来看refer方法
                        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;
    }

    这部分主要是关于刷新Invoker的:

    1. 如果地址只有一个并且协议参数值为empty,则禁止访问。

    2. 先查看缓存并保存副本,注意这个副本后面要用到。如果invokerUrls为空而缓存中不为空,则把缓存的值赋给invokerUrls;否则,把invokerUrls添加到缓存。

    3. 把invokerUrls转换成key为url,value为Invoker的新Map集合。

      - 遍历URL列表

      - 对URL的protocol参数做校验

      - 合并URL,并把URL的全称作为key

      - 查看本地缓存有没有相应的key,没有则创建(调用AbstractProtocol#refer),并放入缓存

    4. 把新生成的Map和之前保存的副本(旧Map)做对比,删除新Map中没有的元素(unused)。

    AbstractProtocol

    // org.apache.dubbo.rpc.protocol.AbstractProtocol#refer
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        // 调用由子类实现的protocolBindingRefer方法
        return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
    }

    DubboProtocol

    从RegistryProtocol到RegistryDirectory到DubboProtocol,绕了一大圈。
    好不容易找到这里,仔细想来也合情合理:
    1. 首先根据配置创建注册中心,然后 RegistryDirectory 执行服务订阅的逻辑;
    2. 数据发生变化,肯定要随之更新影响的组件,执行 notify 逻辑,更新服务URl和Invoker;
    3. Invoker又是由DubboProtocol创建的,就到了这里

    // org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#protocolBindingRefer
    public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
        optimizeSerialization(url);
        // 1. 创建 rpc invoker
        // 2. 启动Netty客户端
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        invokers.add(invoker);
        return invoker;
    }

    来看getClients

    private ExchangeClient[] getClients(URL url) {
        // whether to share connection
    
        boolean useShareConnect = false;
    
        // 如果没有配置,这里是0
        int connections = url.getParameter(CONNECTIONS_KEY, 0);
        List<ReferenceCountExchangeClient> shareClients = null;
        // if not configured, connection is shared, otherwise, one connection for one service
        if (connections == 0) {
            // 没有配置connection,就选择共享的;否则一个服务一个connection
            useShareConnect = true;
    
            /**
             * The xml configuration should have a higher priority than properties.
             */
            String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
            connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(SHARE_CONNECTIONS_KEY,
                    DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr);
            shareClients = getSharedClient(url, connections);
        }
    
        // 根据前面计算得出的connections,创建ExchangeClient数组
        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;
    }
    
    private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
        // 地址作为key:10.204.241.14:20880
        String key = url.getAddress();
        // 先从缓存获取
        List<ReferenceCountExchangeClient> clients = referenceClientMap.get(key);
    
        // 如果缓存中的客户端可用,则返回
        if (checkClientCanUse(clients)) {
            batchClientRefIncr(clients);
            return clients;
        }
    
        locks.putIfAbsent(key, new Object());
        synchronized (locks.get(key)) {
            // 再获取一遍,排除已创建的情况
            clients = referenceClientMap.get(key);
            // dubbo check
            // 这里我觉得作者应该想写成 double check(双重检查)
            if (checkClientCanUse(clients)) {
                batchClientRefIncr(clients);
                return clients;
            }
    
            // connectNum 必须大于等于1
            connectNum = Math.max(connectNum, 1);
    
            // clients为空,先初始化
            if (CollectionUtils.isEmpty(clients)) {
                // *** 创建客户端
                clients = buildReferenceCountExchangeClientList(url, connectNum);
                // 放入缓存
                referenceClientMap.put(key, clients);
    
            } else {
                for (int i = 0; i < clients.size(); i++) {
                    ReferenceCountExchangeClient referenceCountExchangeClient = clients.get(i);
                    // If there is a client in the list that is no longer available, create a new one to replace him.
                    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;
        }
    }
    
    private List<ReferenceCountExchangeClient> buildReferenceCountExchangeClientList(URL url, int connectNum) {
        List<ReferenceCountExchangeClient> clients = new ArrayList<>();
        // 根据connectNum决定初始化多少个客户端
        for (int i = 0; i < connectNum; i++) {
            clients.add(buildReferenceCountExchangeClient(url));
        }
        return clients;
    }
    
    // 包装一个ReferenceCountExchangeClient
    private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url) {
        // 初始化
        ExchangeClient exchangeClient = initClient(url);
        return new ReferenceCountExchangeClient(exchangeClient);
    }
    
    private ExchangeClient initClient(URL url) {
    
        // 获取客户端类型,比如:netty
        String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT));
    
        url = url.addParameter(CODEC_KEY, DubboCodec.NAME);
        // 默认开启心跳检测
        url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT));
    
        // 如果没有str表示类型的Transporter具体实现类,抛出异常
        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 {
            // 判断是否懒加载
            if (url.getParameter(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;
    }

    这么多,做了什么事呢

    1. 判断是否共享连接。没有配置connections属性则获取共享客户端,否则初始化新的客户端。【实际上如果共享客户端没有初始化,也会执行初始化的逻辑。】

    2. 从URl中获取 IP:端口,作为key。先从缓存中获取引用客户端(ReferenceCountExchangeClient),缓存中没有则新建。

    3. 根据connectNum决定初始化多少个客户端,执行初始化代码。

    4. 先获取客户端类型,并校验是否存在此类型的处理类;然后判断是否配置为懒加载,否则把初始化任务交给Exchange层的connect方法。

    Exchangers

    // org.apache.dubbo.remoting.exchange.Exchangers#connect(URL, ExchangeHandler)
    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");
        // 上面做了参数校验,然后通过Dubbo SPI获取具体的处理类,执行对应的connect方法
        return getExchanger(url).connect(url, handler);
    }
    
    // org.apache.dubbo.remoting.exchange.support.header.HeaderExchanger#connect
    public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
        // 把处理交给Transport层,并将结果封装为HeaderExchangeClient
        return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
    }

    Dubbo把这一次定义为【exchange 信息交换层】,作用是封装请求响应模式,同步转异步。

    Transporters

    // org.apache.dubbo.remoting.Transporters#connect(URL, ChannelHandler...)
    public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
        // 初始化ChannelHandler
        ChannelHandler handler;
        if (handlers == null || handlers.length == 0) {
            handler = new ChannelHandlerAdapter();
        } else if (handlers.length == 1) {
            handler = handlers[0];
        } else {
            handler = new ChannelHandlerDispatcher(handlers);
        }
        // 根据Dubbo SPI获取具体的处理类,并执行connect方法
        return getTransporter().connect(url, handler);
    }
    
    // org.apache.dubbo.remoting.transport.netty4.NettyTransporter#connect
    public Client connect(URL url, ChannelHandler listener) throws RemotingException {
        // new一个Netty客户端
        return new NettyClient(url, listener);
    }

    Dubbo把这一层定义为【transport 网络传输层】,作用是抽象 mina 和 netty 为统一接口。

    NettyClient

    // org.apache.dubbo.remoting.transport.netty4.NettyClient
    public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
        // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
        // the handler will be warped: MultiMessageHandler->HeartbeatHandler->handler
        super(url, wrapChannelHandler(url, handler));
    }
    
    // 这里面的 doOpen 和 connect 体现的是模板方法模式,调用过程由抽象类指定,方法由子类实现
    public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
        super(url, handler);
    
        needReconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false);
    
        try {
            // 开启客户端
            doOpen();
        } catch (Throwable t) {
            close();
            throw new RemotingException(url.toInetSocketAddress(), null,
                    "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
                            + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
        }
        try {
            // 连接
            connect();
            if (logger.isInfoEnabled()) {
                logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress());
            }
        } catch (RemotingException t) {
            if (url.getParameter(Constants.CHECK_KEY, true)) {
                close();
                throw t;
            } else {
                logger.warn("Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
                        + " connect to the server " + getRemoteAddress() + " (check == false, ignore and retry later!), cause: " + t.getMessage(), t);
            }
        } catch (Throwable t) {
            close();
            throw new RemotingException(url.toInetSocketAddress(), null,
                    "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
                            + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
        }
    
        executor = (ExecutorService) ExtensionLoader.getExtensionLoader(DataStore.class)
                .getDefaultExtension().get(CONSUMER_SIDE, Integer.toString(url.getPort()));
        ExtensionLoader.getExtensionLoader(DataStore.class)
                .getDefaultExtension().remove(CONSUMER_SIDE, Integer.toString(url.getPort()));
    }
    
    // org.apache.dubbo.remoting.transport.netty4.NettyClient#doOpen
    protected void doOpen() throws Throwable {
        final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
        bootstrap = new Bootstrap();
        bootstrap.group(nioEventLoopGroup)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
                .channel(NioSocketChannel.class);
    
        if (getConnectTimeout() < 3000) {
            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
        } else {
            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
        }
    
        bootstrap.handler(new ChannelInitializer() {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
                ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                        .addLast("decoder", adapter.getDecoder())
                        .addLast("encoder", adapter.getEncoder())
                        .addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))
                        .addLast("handler", nettyClientHandler);
                String socksProxyHost = ConfigUtils.getProperty(SOCKS_PROXY_HOST);
                if(socksProxyHost != null) {
                    int socksProxyPort = Integer.parseInt(ConfigUtils.getProperty(SOCKS_PROXY_PORT, DEFAULT_SOCKS_PROXY_PORT));
                    Socks5ProxyHandler socks5ProxyHandler = new Socks5ProxyHandler(new InetSocketAddress(socksProxyHost, socksProxyPort));
                    ch.pipeline().addFirst(socks5ProxyHandler);
                }
            }
        });
    }

    这部分就是和Netty客户端最接近的代码了,自行去熟悉Netty即可。

    除了Netty构建客户端的过程之外,这部分值得学习的就是模板方法的实际应用了。

    最后看下整体的调用过程

    createProxy:396, ReferenceConfig
    refer:392, RegistryProtocol
    doRefer:411, RegistryProtocol
    
    subscribe:172, RegistryDirectory
    subscribe:295, FailbackRegistry
    doSubscribe:183, ZookeeperRegistry
    notify:360, FailbackRegistry
    doNotify:369, FailbackRegistry
    notify:418, AbstractRegistry
    
    notify:233, RegistryDirectory
    refreshOverrideAndInvoker:239, RegistryDirectory
    refreshInvoker:280, RegistryDirectory
    toInvokers:423, RegistryDirectory
    
    refer:91, AbstractProtocol
    protocolBindingRefer:407, DubboProtocol
    getClients:430, DubboProtocol
    getSharedClient:475, DubboProtocol
    buildReferenceCountExchangeClientList:550, DubboProtocol
    buildReferenceCountExchangeClient:563, DubboProtocol
    initClient:595, DubboProtocol
    
    connect:109, Exchangers
    connect:39, HeaderExchanger
    connect:75, Transporters
    connect:40, NettyTransporter
    <init>:80, NettyClient
    <init>:60, AbstractClient
    doOpen:91, NettyClient

    这个大体上满足官方所阐述的《整体设计》层次

    另附各层说明

    • config 配置层:对外配置接口,以 ServiceConfigReferenceConfig 为中心,可以直接初始化配置类,也可以通过 spring 解析配置生成配置类
    • proxy 服务代理层:服务接口透明代理,生成服务的客户端 Stub 和服务器端 Skeleton, 以 ServiceProxy为中心,扩展接口为 ProxyFactory
    • registry 注册中心层:封装服务地址的注册与发现,以服务 URL 为中心,扩展接口为 RegistryFactoryRegistryRegistryService
    • cluster 路由层:封装多个提供者的路由及负载均衡,并桥接注册中心,以 Invoker 为中心,扩展接口为 ClusterDirectoryRouterLoadBalance
    • monitor 监控层:RPC 调用次数和调用时间监控,以 Statistics 为中心,扩展接口为 MonitorFactoryMonitorMonitorService
    • protocol 远程调用层:封装 RPC 调用,以 InvocationResult 为中心,扩展接口为 ProtocolInvokerExporter
    • exchange 信息交换层:封装请求响应模式,同步转异步,以 RequestResponse 为中心,扩展接口为 ExchangerExchangeChannelExchangeClientExchangeServer
    • transport 网络传输层:抽象 mina 和 netty 为统一接口,以 Message 为中心,扩展接口为 ChannelTransporterClientServerCodec
    • serialize 数据序列化层:可复用的一些工具,扩展接口为 SerializationObjectInputObjectOutputThreadPool

    可以结合这个层次结构去熟悉源码

    [重点]最后调用 ProxyFactory 生成代理类

    进入的是抽象代理工厂

    AbstractProxyFactory

    // org.apache.dubbo.rpc.proxy.AbstractProxyFactory
    public <T> T getProxy(Invoker<T> invoker) throws RpcException {
        // 交给重载方法
        return getProxy(invoker, false);
    }
    
    
    public <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException {
        // interface com.demo.common.HelloService -> zookeeper://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?actives=5&anyhost=true&application=dubbo-consumer&bean.name=providers:dubbo:com.demo.common.HelloService:1.0&check=false&default.deprecated=false&default.dynamic=false&default.register=true&deprecated=false&dubbo=2.0.2&dynamic=false&generic=false&interface=com.demo.common.HelloService&lazy=false&methods=hello&pid=11484&qos.enable=false&register=true&register.ip=10.204.241.39&release=2.7.3&remote.application=service-provider&retries=3&revision=1.0&side=consumer&sticky=false&timeout=3000&timestamp=1584952094835&version=1.0
        Class<?>[] interfaces = null;
        // 获取interfaces参数值
        String config = invoker.getUrl().getParameter(INTERFACES);
        // 如果有interfaces参数值
        if (config != null && config.length() > 0) {
            // 拆分接口列表
            String[] types = COMMA_SPLIT_PATTERN.split(config);
            if (types != null && types.length > 0) {
                interfaces = new Class<?>[types.length + 2];
                interfaces[0] = invoker.getInterface();
                interfaces[1] = EchoService.class;
                for (int i = 0; i < types.length; i++) {
                    // 加载接口类
                    // 官网文档是不是有错误,是interfaces[i + 1]。如果这样的话,这里就会覆盖interfaces[1]的值了。
                    interfaces[i + 2] = ReflectUtils.forName(types[i]);
                }
            }
        }
        // 初始化interfaces
        if (interfaces == null) {
            // invoker.getInterface(): com.demo.common.HelloService
            interfaces = new Class<?>[]{invoker.getInterface(), EchoService.class};
        }
    
        // 为 http 和 hessian 协议提供泛化调用支持
        if (!GenericService.class.isAssignableFrom(invoker.getInterface()) && generic) {
            int len = interfaces.length;
            Class<?>[] temp = interfaces;
            interfaces = new Class<?>[len + 1];
            System.arraycopy(temp, 0, interfaces, 0, len);
            interfaces[len] = com.alibaba.dubbo.rpc.service.GenericService.class;
        }
    
        // 进入JavassistProxyFactory类
        return getProxy(invoker, interfaces);
    }

    这一步主要利用Invoker创建interface的Class列表。

    JavassistProxyFactory

    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        // 1. Proxy.getProxy(interfaces),利用Javassist动态生成Proxy子类,其中实现了newInstance的细节。
        // 2. 调用Proxy子类的newInstance方法生成服务接口代理类的一个实例
        return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
    }

    进入getProxy方法:

    吐槽一下:这两个类的名字真骚啊,注意p的大小写

    看一下生成的类大概是什么样子的。

    利用阿里 Java 应用诊断工具 Arthas 

    1. 下载解压

     2. 输入  java -jar arthas-boot.jar

     3. 上图红框的是我消费者应用,选择2

     4. 输入 sc *.proxy0

    5. 输入 jad org.apache.dubbo.common.bytecode.proxy0

    完整代码

    ClassLoader:
      +-sun.misc.Launcher$AppClassLoader@18b4aac2
      +-sun.misc.Launcher$ExtClassLoader@4b53f538
    
    Location:
    /E:/apache-maven-repo/org/apache/dubbo/dubbo/2.7.3/dubbo-2.7.3.jar
    
    /*
     * Decompiled with CFR.
     *
     * Could not load the following classes:
     *  com.demo.common.HelloService
     */
    package org.apache.dubbo.common.bytecode;
    
    import com.alibaba.dubbo.rpc.service.EchoService;
    import com.demo.common.HelloService;
    import java.lang.reflect.InvocationHandler;
    import java.lang.reflect.Method;
    import org.apache.dubbo.common.bytecode.ClassGenerator;
    
    public class proxy0
    implements ClassGenerator.DC,
    HelloService,
    EchoService {
        public static Method[] methods;
        private InvocationHandler handler;
    
        public String hello() {
            Object[] arrobject = new Object[]{};
            Object object = this.handler.invoke(this, methods[0], arrobject);
            return (String)object;
        }
    
        @Override
        public Object $echo(Object object) {
            Object[] arrobject = new Object[]{object};
            Object object2 = this.handler.invoke(this, methods[1], arrobject);
            return object2;
        }
    
        public proxy0() {
        }
    
        public proxy0(InvocationHandler invocationHandler) {
            this.handler = invocationHandler;
        }
    }

    再看一下Proxy子类的,输入 sc *.Proxy0

    输入 jad org.apache.dubbo.common.bytecode.Proxy0

    这个类里实现了Proxy抽象类定义的newInstance方法。

    所以Proxy.getProxy方法一共生成了两个类的字节码,不是一个。

    总结

    注册中心细节:

    服务提供者启动时: 向 /dubbo/com.demo.common.HelloService/providers 目录下写入自己的 URL 地址
    服务消费者启动时: 订阅 /dubbo/com.demo.common.HelloService/providers 目录下的提供者 URL 地址。并向 /dubbo/com.demo.common.HelloService/consumers 目录下写入自己的 URL 地址。

    读源码体会:

    1. 要先去了解框架的层次,比如自上而下:config 配置层、proxy 服务代理层、registry 注册中心层、cluster 路由层、monitor 监控层、protocol 远程调用层、exchange 信息交换层、transport 网络传输层、serialize 数据序列化层。这些层次在源码中都有体现。

    2. 要先大致清楚这个框架或者框架的某个功能解决了什么问题,你才能大致“猜到”某些源代码的目的。

    3. 在代码细节方面学习的是代码习惯,在全局范围方面学习的是设计模式以及解决问题的思路,进而学习别人如何把系统变得健壮易扩展。

    ========================================2020-05-22更新:服务引用流程图

  • 相关阅读:
    Springmvc数据验证
    Springmvc文件上传
    BaseController
    说说NSProxy
    Objective-C的动态设计
    UITableView卡片式分组
    RunLoop应用之性能优化
    OC与JS交互之JavaScriptCore
    Core Data 迁移
    一个广告轮播视图的实现
  • 原文地址:https://www.cnblogs.com/LUA123/p/12516366.html
Copyright © 2011-2022 走看看