Dubbo服务引用

Dubbo RPC
笔记
发布日期:   2023-06-10
更新日期:   2023-06-10
文章字数:   1.8k
阅读时长:   8 分
阅读次数:  

前言

在SpringBoot与Dubbo整合源码分析那一篇博文中有说到在Spring启动过程中会对@DubboReference标注的对象进行赋值，赋值的内容就是ReferenceBean#get方法返回的代理对象。而这代理对象的生成便是服务引用的过程。现在我们直接来到ReferenceConfig#get方法

ReferenceConfig#get

public synchronized T get() {
    if (destroyed) {
    }
    if (ref == null) {
        // 核心方法
        init();
    }
    return ref;
}

ReferenceConfig#init

public synchronized void init() {
        // 防止重复初始化
        if (initialized) {
            return;
        }

        if (bootstrap == null) {
            bootstrap = DubboBootstrap.getInstance();
            bootstrap.init();
        }

        // 检查和更新配置
        checkAndUpdateSubConfigs();
        // 检验本地存根的合法性
        checkStubAndLocal(interfaceClass);
        ConfigValidationUtils.checkMock(interfaceClass, this);

        Map<String, String> map = new HashMap<String, String>();
        // ----------省略代码------------------
        // 省略部分主要是解析一些配置参数往上面的这个map里面放，不是核心

        // 核心方法 -------创建代理类-------
        ref = createProxy(map);

        serviceMetadata.setTarget(ref);
        serviceMetadata.addAttribute(PROXY_CLASS_REF, ref);
        ConsumerModel consumerModel = repository.lookupReferredService(serviceMetadata.getServiceKey());
        consumerModel.setProxyObject(ref);
        consumerModel.init(attributes);

        initialized = true;

        checkInvokerAvailable();

        // dispatch a ReferenceConfigInitializedEvent since 2.7.4
        dispatch(new ReferenceConfigInitializedEvent(this, invoker));
    }

ReferenceConfig#createProxy

private T createProxy(Map<String, String> map) {
    // injvm://    一般不使用
    if (shouldJvmRefer(map)) {
        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 {
        // urls 存储的是@DubboReference上的url或者注册中心地址
        urls.clear();
        // @DubboReference标明了url
        if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
            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.setPath(interfaceName);
                    }
                    // 如果是注册中心地址，则在url中添加一个refer参数
                    if (UrlUtils.isRegistry(url)) {
                        urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                    } else {
                        // 合并url中的参数到map
                        urls.add(ClusterUtils.mergeUrl(url, map));
                    }
                }
            }
        } else { // assemble URL from register center's configuration
            // if protocols not injvm checkRegistry
            if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
                // 检验注册中心
                checkRegistry();
                // 加载注册中心地址
                List<URL> us = ConfigValidationUtils.loadRegistries(this, false);
                if (CollectionUtils.isNotEmpty(us)) {
                    for (URL u : us) {
                        URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);
                        if (monitorUrl != null) {
                            map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                        }
                        // 对于注册中心地址都添加refer参数
                        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) {
            // 这里用了SPI自适应扩展，这里可能的结果是DubboProtocol或RegisterProtocol,当然他们会被一些Wrapper包裹
            // 调用refer生成invoker
            invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
        } else {
            // 注册中心多个
            List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
            URL registryURL = null;
            for (URL url : urls) {
                // 调用refer生成invoker
                invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
                // 记录注册中心
                if (UrlUtils.isRegistry(url)) {
                    registryURL = url; // use last registry url
                }
            }
            // 存在注册中心地址
            if (registryURL != null) { // registry url is available
                // for multi-subscription scenario, use 'zone-aware' policy by default
                String cluster = registryURL.getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME);
                // The invoker wrap sequence would be: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
                invoker = Cluster.getCluster(cluster, false).join(new StaticDirectory(registryURL, invokers));
            } else { // not a registry url, must be direct invoke.
                String cluster = CollectionUtils.isNotEmpty(invokers)
                        ? (invokers.get(0).getUrl() != null ? invokers.get(0).getUrl().getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME) : Cluster.DEFAULT)
                        : Cluster.DEFAULT;
                invoker = Cluster.getCluster(cluster).join(new StaticDirectory(invokers));
            }
        }
    }

    if (logger.isInfoEnabled()) {
        logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
    }
    /**
     * @since 2.7.0
     * ServiceData Store
     */
    String metadata = map.get(METADATA_KEY);
    WritableMetadataService metadataService = WritableMetadataService.getExtension(metadata == null ? DEFAULT_METADATA_STORAGE_TYPE : metadata);
    if (metadataService != null) {
        URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
        metadataService.publishServiceDefinition(consumerURL);
    }
    // create service proxy
    return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
}

这一段比较重要的就是把消费者配置的所有注册中心获取出来

  • 单注册中心：直接调用Protocol#refer(interfaceClass, urls.get(0))得到一个Invoker对象
  • 多注册中心，则遍历每个注册中心调用Protocol#refer(interfaceClass, urls.get(0))得到一个Invoker对象添加到invokers中，然后调用Cluster.getCluster(cluster, false).join(new StaticDirectory(registryURL, invokers))封装所有invokers得到一个invoker

现在问题就在于这个invoker是怎么得到的

Protocol#refer

由于Protocol是由SPI导入的，而这里的Protocol是有2种可能性的，一种是DubboProtocol还有一种是RegistryProtocol。这里我们一般是注册中心（register://）所以走RegistryProtocol

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    // 获取注册中心地址
    url = getRegistryUrl(url);
    Registry registry = registryFactory.getRegistry(url);
    if (RegistryService.class.equals(type)) {
        return proxyFactory.getInvoker((T) registry, type, url);
    }

    // group="a,b" or group="*"
    Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
    String group = qs.get(GROUP_KEY);
    if (group != null && group.length() > 0) {
        if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
            return doRefer(Cluster.getCluster(MergeableCluster.NAME), registry, type, url);
        }
    }
    // 这里qs.get(CLUSTER_KEY)是空，所以Cluster是default,又String DEFAULT = FailoverCluster.NAME;；所以Cluster是FailOverCluster又由于MockClusterWrapper存在，SPI可知cluster = MockClusterWrapper->FailOverCluster
    Cluster cluster = Cluster.getCluster(qs.get(CLUSTER_KEY));
    return doRefer(cluster, registry, type, url);
}

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    // new 一个服务目录
    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.getConsumerUrl().getParameters());
     // 消费者url
    URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
    if (directory.isShouldRegister()) {
        directory.setRegisteredConsumerUrl(subscribeUrl);
        // 注册简化后的消费者url
        registry.register(directory.getRegisteredConsumerUrl());
    }
    // 构造路由链,路由链会在引入服务时按路由条件进行过滤 ------挺重要（暂时不了解）--------
    directory.buildRouterChain(subscribeUrl);
    // 服务目录需要订阅的路径，用于监听 ------挺重要（暂时不了解）--------
    directory.subscribe(toSubscribeUrl(subscribeUrl));

    // 得到invoker, MockClusterWrapper
    Invoker<T> invoker = cluster.join(directory);
    List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);
    if (CollectionUtils.isEmpty(listeners)) {
        return invoker;
    }

    RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker);
    for (RegistryProtocolListener listener : listeners) {
        listener.onRefer(this, registryInvokerWrapper);
    }
    return registryInvokerWrapper;
}

这段代码主要做的事情有以下几点

  1. new 一个服务目录
  2. 注册消费者到注册中心
  3. 构造路由链（通过一些规则将服务提供者筛选一遍，一般可以用来蓝绿部署）
  4. 监听注册中心目录（包括服务提供者目录，可以实时的将服务提供者写到服务目录中去）
  5. 通过Cluster.join将服务目录放到FailoverClusterInvoker 构造出一个调用链invoker并返回

那么问题来了，服务提供者的url是什么时候获取并放入到服务目录里面的呢？

其实是在监听服务目录订阅路径这个操作下执行的。下面我们可以看下到底是怎么一回事。

监听订阅目录

RegisteryProtocol#doRefer
    --- directory#subscribe(toSubscribeUrl(subscribeUrl))
        --- registry#subscribe(url, this);
           --- FallbackRegistry#subscribe
           --- ZookeeperRegistry#doSubscribe

public void doSubscribe(final URL url, final NotifyListener listener) {
    try {
         // 订阅所有服务
        if (ANY_VALUE.equals(url.getServiceInterface())) {
            String root = toRootPath();
            ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
            ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (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)), k);
                    }
                }
            });
            zkClient.create(root, false);
            List<String> services = zkClient.addChildListener(root, zkListener);
            if (CollectionUtils.isNotEmpty(services)) {
                for (String service : services) {
                    service = URL.decode(service);
                    anyServices.add(service);
                    subscribe(url.setPath(service).addParameters(INTERFACE_KEY, service,
                            Constants.CHECK_KEY, String.valueOf(false)), listener);
                }
            }
        } else {
            // 单独订阅某一个服务
            List<URL> urls = new ArrayList<>();
            // 得到真正要监听的zk上的路径（见下图）
            for (String path : toCategoriesPath(url)) {
                 // 根据监听地址去拿listeners
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
                ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, k, toUrlsWithEmpty(url, parentPath, currentChilds)));
                zkClient.create(path, false);
                List<String> children = zkClient.addChildListener(path, zkListener);
                if (children != null) {
                    urls.addAll(toUrlsWithEmpty(url, path, children));
                }
            }
             // 这里的urls就是从现在所引入的服务的目录下查到的url
            notify(url, listener, urls);
        }
    } catch (Throwable e) {
        throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

文章作者: dm
文章链接: https://dmsupine.com/2023/06/10/dubbo-fu-wu-yin-yong/
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 dm !
Dubbo RPC
赏
评论
 上一篇
Dubbo服务导出源码 Dubbo服务导出源码
入口还记得springboot会注册DubboBootstrapApplicationListener监听事件，在这个事件中onContextRefreshedEvent调用了dubboBootstrap.start()方法。说明了在Spr
2023-07-01 笔记
Dubbo Java 源码
下一篇 
SpringBoot与Dubbo整合源码分析 SpringBoot与Dubbo整合源码分析
DemoSpringBoot整合Dubbo只需加个依赖，服务提供者加注解@Service (dubbo的不是spring的，后改为@DubboService)，服务消费者加注解@Reference注解 加依赖<!-- Dubbo集成
2023-06-01 笔记
Dubbo Java 源码 Spring SpringBoot
  目录