dubbo源码阅读-服务订阅(八)之远程订阅(dubbo)

说明

从《服务订阅主流程》可以看到根据协议来注册 我们默认没有根据url直接配置所以url是registry SPI扩展就是走的RegistryProtocol

RegistryProtocol

<1>refer

/**
     * type为订阅接口
     * @param type Service class
     * @param url  URL address for the remote service
     * @param <T>
     * @return
     * @throws RpcException
     */
    @Override
    @SuppressWarnings("unchecked")
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        //删除registry协议 并获取url订阅协议默认dubbo 如果配置的注册中心是zookeeper://name协议就是zookeeper
        url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
        //@SPI实现  获得对应的registry实现
        Registry registry = registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) {
            return proxyFactory.getInvoker((T) registry, type, url);
        }

        // group="a,b" or group="*"
        //将url参数 转换为 map
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
        //获取group   https://blog.csdn.net/weixin_43868594/article/details/84956886
        String group = qs.get(Constants.GROUP_KEY);
        //如果配置了group
        if (group != null && group.length() > 0) {
            if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
                    || "*".equals(group)) {
                //<2>使用mergeable
                return doRefer(getMergeableCluster(), registry, type, url);
            }
        }
        //<2>cluster为SPI扩展 默认FailoverCluster 为集群策略
        return doRefer(cluster, registry, type, url);
    }      

<2>doRefer

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        //type为订阅接口类型 url为订阅url
        //zookeeper://IP:2181/com.alibaba.dubbo.registry.RegistryService?**
        RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // all attributes of REFER_KEY
        //获取参数
        Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());

        //订阅url
        URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
        if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
                && url.getParameter(Constants.REGISTER_KEY, true)) {
            URL registeredConsumerUrl = getRegisteredConsumerUrl(subscribeUrl, url);
            //将订阅url 添加到已注册列表 registryList
            registry.register(registeredConsumerUrl);
            //设置到directory
            directory.setRegisteredConsumerUrl(registeredConsumerUrl);
        }
        //<3>订阅服务 内部调用registry.subscribe
        directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
                Constants.PROVIDERS_CATEGORY
                        + "," + Constants.CONFIGURATORS_CATEGORY
                        + "," + Constants.ROUTERS_CATEGORY));

        //通过集群策略包装成一个Invoker返回
        Invoker invoker = cluster.join(directory);
        //注册到订阅列表
        ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
        return invoker;
    }      

RegistryDirectory

<3>subscribe

public void subscribe(URL url) {
        //此时的url为:consumer://192.168.2.1/com.alibaba.dubbo.demo.DemoService?*
        setConsumerUrl(url);
         //对应的注册中心实现类 调用订阅方法 通知添加监听器 就是当前对象 实现了 NotifyListener 具体可看:https://www.cnblogs.com/LQBlog/p/12522417.html
        registry.subscribe(url, this);
    }