dubbo源码阅读-注册中心(十三)之Zookeeper

类图

 服务注册

RegistryProtocol

<1>register

参见《dubbo源码阅读-服务暴露(七)之远程暴露(dubbo)》

 public void register(URL registryUrl, URL registedProviderUrl) {
        //<2>SPI扩展点 此时url已经变成了配置的协议而不是registry 如:zookeeper:/
        Registry registry = registryFactory.getRegistry(registryUrl);
        //<3>注册到注册中心
        registry.register(registedProviderUrl);
    }

服务订阅

RegistryDirectory

<6>subscribe

参见《dubbo源码阅读-服务订阅(八)之远程订阅(dubbo)》

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

ZookeeperRegistry

<3>register

com.alibaba.dubbo.registry.support.FailbackRegistry#register

   @Override
    public void register(URL url) {
        //<4>在成员列表增加registered 已注册url
        super.register(url);
        //成员变变量失败集合url
        failedRegistered.remove(url);
        failedUnregistered.remove(url);
        try {
            //<5>模板方法模式 具体的注册逻辑由子类实现
            doRegister(url);
        } catch (Exception e) {
            Throwable t = e;

            // 是否配置了跳过检查 默认不跳过 check=true 如果注册失败会抛出异常  否则只打印错误日志
            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);
        }
    }

<4>supper.register

com.alibaba.dubbo.registry.support.FailbackRegistry#register

->

com.alibaba.dubbo.registry.support.AbstractRegistry#register

 @Override
    public void register(URL url) {
        if (url == null) {
            throw new IllegalArgumentException("register url == null");
        }
        if (logger.isInfoEnabled()) {
            logger.info("Register: " + url);
        }
        registered.add(url);
    }

<5>doRegister

com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistry#doRegister

 @Override
    protected void doRegister(URL url) {
        try {
            //添加节点/dubbo/com.alibaba.dubbo.demo.DemoService/providers 同时添加服务信息
            zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
        } catch (Throwable e) {
            throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

注:zkClikent也是SPI注解 可选有CuratorZookeeperClient，ZkclientZookeeperClient

<7>subscribe

com.alibaba.dubbo.registry.support.FailbackRegistry#subscribe

   @Override
    public void subscribe(URL url, NotifyListener listener) {
        //将监听器添加到 成员变量 key为url value为lists
        super.subscribe(url, listener);
        //从异常监听里面移除
        removeFailedSubscribed(url, listener);
        try {
            // <8>具体的订阅逻辑 子类实现 模板方法模式
            doSubscribe(url, listener);
        } catch (Exception e) {
            Throwable t = e;

            List<URL> urls = getCacheUrls(url);
            if (urls != null && !urls.isEmpty()) {
                notify(url, listener, urls);
                logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + getUrl().getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/dubbo-registry-" + url.getHost() + ".cache") + ", cause: " + t.getMessage(), t);
            } else {
                // If the startup detection is opened, the Exception is thrown directly.
                boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                        && url.getParameter(Constants.CHECK_KEY, true);
                boolean skipFailback = t instanceof SkipFailbackWrapperException;
                //是否配了check如果配了如果订阅失败就报错
                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);
        }
    }

<8>doSubscribe

    @Override
    protected void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            // 处理所有 Service 层的发起订阅，例如监控中心的订阅
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
          .....
            } else {
                List<URL> urls = new ArrayList<URL>();
                for (String path : toCategoriesPath(url)) {
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    //根据Url获取zk的监听器 没获取到则初始化
                    if (listeners == null) {
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                        listeners = zkListeners.get(url);
                    }
                    // 获得 ChildListener 对象 zk使用
                    ChildListener zkListener = listeners.get(listener);
                    if (zkListener == null) {
                        //获取不到则添加一个 这里可以理解成 将dubbo的监听器 适配 当zk节点变动触发
                        listeners.putIfAbsent(listener,  new ChildListener() {
                            @Override
                            public void childChanged(String parentPath, List<String> currentChilds) {
                                //发起通知
                                ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                            }
                        });
                        zkListener = listeners.get(listener);
                    }
                    //ZK创建一个节点 如:/dubbo/com.alibaba.dubbo.demo.DemoService/providers
                    zkClient.create(path, false);
                    //像path节点发起订阅
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) {
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                //<9>订阅成功调用notify 通知监听器 如:RegistryDirectory
                notify(url, listener, urls);
            }
        } catch (Throwable e) {
            throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

 <9>notify

前面还有多个方法处理忽略了 只放出了 最终的notify

com.alibaba.dubbo.registry.support.AbstractRegistry#notify(com.alibaba.dubbo.common.URL, com.alibaba.dubbo.registry.NotifyListener, java.util.List<com.alibaba.dubbo.common.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 ((urls == null || urls.isEmpty())
                && !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)) {
                //将订阅url根据category进行分组 默认为providers
                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);
            }
        }
        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);
            saveProperties(url);
            //<10>将分组后的通知lintener
            listener.notify(categoryList);
        }
    }

 RegistryDirectory

<10>notify

com.alibaba.dubbo.registry.integration.RegistryDirectory#notify

    @Override
    public synchronized void notify(List<URL> urls) {
        //提供者url
        List<URL> invokerUrls = new ArrayList<URL>();
        //路由规则https://cloud.tencent.com/developer/article/1443518
        List<URL> routerUrls = new ArrayList<URL>();
        //处理配置规则 URL 集合 http://dubbo.apache.org/zh-cn/docs/user/demos/config-rule-deprecated.html
        List<URL> configuratorUrls = new ArrayList<URL>();
        /**
         *通过订阅到数据进行数据分组
         *
         */
        for (URL url : urls) {
            String protocol = url.getProtocol();
            String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
            //当url是route或者是routes表示是路由规则
            if (Constants.ROUTERS_CATEGORY.equals(category)
                    || Constants.ROUTE_PROTOCOL.equals(protocol)) {
                routerUrls.add(url);
                //当url是configurators 或者 override 表示是路由规则
            } else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
                    || Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
                configuratorUrls.add(url);
                //当url是providers 表示是提供者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.isEmpty()) {
            //<11>
            this.configurators = toConfigurators(configuratorUrls);
        }
        // routers
        if (routerUrls != null && !routerUrls.isEmpty()) {
            //获得routers
            List<Router> routers = toRouters(routerUrls);
            if (routers != null) { // null - do nothing
                //<12>
                setRouters(routers);
            }
        }
        List<Configurator> localConfigurators = this.configurators; // local reference
        // merge override parameters
        this.overrideDirectoryUrl = directoryUrl;
        if (localConfigurators != null && !localConfigurators.isEmpty()) {
            for (Configurator configurator : localConfigurators) {
                this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
            }
        }
        // <13>处理服务提供者 URL 集合
        refreshInvoker(invokerUrls);
    }

<11>toConfigurators

com.alibaba.dubbo.registry.integration.RegistryDirectory#toConfigurators

 public static List<Configurator> toConfigurators(List<URL> urls) {
        if (urls == null || urls.isEmpty()) {
            return Collections.emptyList();
        }
        List<Configurator> configurators = new ArrayList<Configurator>(urls.size());
        for (URL url : urls) {
            //如果协议为空 则忽略 表示未配置 配置规则
            /**
             * empty://10.3.17.72/com.alibaba.dubbo.demo.DemoService?accesslog=true&application=soa-promotion-consumer&category=configurators&dubbo=2.0.2&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello,save&pid=23772&side=consumer&timestamp=1584603775927&validation=jvalidation
             */
            if (Constants.EMPTY_PROTOCOL.equals(url.getProtocol())) {
                configurators.clear();
                break;
            }
            Map<String, String> override = new HashMap<String, String>(url.getParameters());
            //The anyhost parameter of override may be added automatically, it can't change the judgement of changing url
            override.remove(Constants.ANYHOST_KEY);
            if (override.size() == 0) {
                configurators.clear();
                continue;
            }
            //SPI扩展点 获得对应的configurator
            configurators.add(configuratorFactory.getConfigurator(url));
        }
        Collections.sort(configurators);
        return configurators;
    }

<12>setRouters

com.alibaba.dubbo.rpc.cluster.directory.AbstractDirectory#setRouters

    protected void setRouters(List<Router> routers) {
        // copy list
        routers = routers == null ? new ArrayList<Router>() : new ArrayList<Router>(routers);
        // 根据url配置的route通过SPI获取并添加到routers
        String routerkey = url.getParameter(Constants.ROUTER_KEY);
        if (routerkey != null && routerkey.length() > 0) {
            RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getExtension(routerkey);
            routers.add(routerFactory.getRouter(url));
        }
        // append mock invoker selector
        //添加默认router
        routers.add(new MockInvokersSelector());
        routers.add(new TagRouter());
        Collections.sort(routers);
        this.routers = routers;
    }

<13>refreshInvoker

private void refreshInvoker(List<URL> invokerUrls) {
        //如果是empty
        if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
                && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
            // 设置禁止访问
            this.forbidden = true;
            // 销毁所有 Invoker 集合
            this.methodInvokerMap = null; // Set the method invoker map to null
            destroyAllInvokers(); // Close all invokers
        } else {
            // // 设置允许访问
            this.forbidden = false; // Allow to access
            // 引用老的 urlInvokerMap
            Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
            //// 传入的 invokerUrls 为空，说明是路由规则或配置规则发生改变，此时 invokerUrls 是空的，直接使用 cachedInvokerUrls 。
            if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
                invokerUrls.addAll(this.cachedInvokerUrls);
            } else {
                this.cachedInvokerUrls = new HashSet<URL>();
                this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
            }
            if (invokerUrls.isEmpty()) {
                return;
            }
            //<14>将传入的 invokerUrls ，转成新的 urlInvokerMap
            Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
            //    // 转换出新的 methodInvokerMap
            Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // Change method name to map Invoker Map
            // state change
            // If the calculation is wrong, it is not processed.
            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;
            }
            //    // 若服务引用多 group ，则按照 method + group 聚合 Invoker 集合
            this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
            //   // 销毁不再使用的 Invoker 集合
            this.urlInvokerMap = newUrlInvokerMap;
            try {
                destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
            } catch (Exception e) {
                logger.warn("destroyUnusedInvokers error. ", e);
            }
        }
    }

 <14>toInvokers

    /**
     * Turn urls into invokers, and if url has been refer, will not re-reference.
     *
     * @param urls
     * @return invokers
     */
    private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
        Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
        // 若为空，直接返回
        if (urls == null || urls.isEmpty()) {
            return newUrlInvokerMap;
        }
        // 已初始化的服务器提供 URL 集合 避免重复初始化
        Set<String> keys = new HashSet<String>();
        // 获得引用服务的协议
        String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
        // 循环服务提供者 URL 集合，转成 Invoker 集合
        for (URL providerUrl : urls) {
            // If protocol is configured at the reference side, only the matching protocol is selected
            // 如果 reference 端配置了 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;
                }
            }
            // 忽略，若为 `empty://` 协议
            if (Constants.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 url = mergeUrl(providerUrl);
            // 添加到 `keys` 中
            String key = url.toFullString(); // The parameter urls are sorted
            //如果初始化过 则忽略
            if (keys.contains(key)) { // Repeated url
                continue;
            }
            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
            // 如果服务端 URL 发生变化，则重新 refer 引用
            Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
            Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
            //未在缓存中，重新引用 防止重复生成
            if (invoker == null) { // Not in the cache, refer again
                try {
                    //判断是否开始 disabled=true
                    boolean enabled = true;
                    if (url.hasParameter(Constants.DISABLED_KEY)) {
                        enabled = !url.getParameter(Constants.DISABLED_KEY, false);
                    } else {
                        enabled = url.getParameter(Constants.ENABLED_KEY, true);
                    }
                    if (enabled) {
                        //<15>引用服务protocol.refer(serviceType, url)
                        invoker = new InvokerDelegate<T>(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;
    }

DubboProtocol

<15>refer

@Override
    public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
        optimizeSerialization(url);
        //创建DubboInvoker <16>getClients
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        //添加到已订阅列表
        invokers.add(invoker);
        return invoker;
    }

<16>getClients

 private ExchangeClient[] getClients(URL url) {
        // whether to share connection
        //是否共享连接 同一个远程服务公用同一个连接 文档:http://dubbo.apache.org/zh-cn/docs/user/demos/config-connections.html
        //http://dubbo.apache.org/zh-cn/docs/user/references/xml/dubbo-reference.html
        boolean service_share_connect = false;
        //从url获取是否共享连接 默认为0 共享
        int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
        // if not configured, connection is shared, otherwise, one connection for one service
        if (connections == 0) {
            service_share_connect = true;
            connections = 1;
        }

        ExchangeClient[] clients = new ExchangeClient[connections];
        for (int i = 0; i < clients.length; i++) {
            if (service_share_connect) {
                // // <17>共享
                clients[i] = getSharedClient(url);
            } else { //<18>不共享
                clients[i] = initClient(url);
            }
        }
        return clients;
    }

<17>getSharedClient

 /**
     * Get shared connection
     */
    private ExchangeClient getSharedClient(URL url) {
        //获得url地址 ip:port
        String key = url.getAddress();
        //在缓存中获取
        ReferenceCountExchangeClient client = referenceClientMap.get(key);
        //不等于null表示初始化过了
        if (client != null) {
            //是否关闭
            if (!client.isClosed()) {
                //记录获取次数
                client.incrementAndGetCount();
                return client;
            } else {
                //如果关闭移除 重新建立连接
                referenceClientMap.remove(key);
            }
        }
        //存入锁对象 如果不存在 如果存在直接返回
        locks.putIfAbsent(key, new Object());
        //加锁防止重复初始化
        synchronized (locks.get(key)) {
            //防止锁穿透
            if (referenceClientMap.containsKey(key)) {
                return referenceClientMap.get(key);
            }
            //<18>初始化clieeent
            ExchangeClient exchangeClient = initClient(url);
           // 将 `exchangeClient` 包装，创建 ReferenceCountExchangeClient 对象
            client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
            //添加到集合
            referenceClientMap.put(key, client);
            ghostClientMap.remove(key);
            locks.remove(key);
            return client;
        }
    }

<18>initClient

 /**
     * Create new connection
     */
    private ExchangeClient initClient(URL url) {

        // client type setting.
        String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));

        url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
        // enable heartbeat by default
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));

        // BIO is not allowed since it has severe performance issue.
        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 {
            // 懒连接，创建 LazyConnectExchangeClient 对象
            // connection should be lazy
            if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
                client = new LazyConnectExchangeClient(url, requestHandler);
            } else {
                // 直接连接，创建 HeaderExchangeClient 对象
                client = Exchangers.connect(url, requestHandler);
            }
        } catch (RemotingException e) {
            throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
        }
        return client;
    }

<2>RegistryFactory

接口定义

@SPI("dubbo")//缺省值dubbo
public interface RegistryFactory {

    /**
     * Connect to the registry
     * <p>
     * Connecting the registry needs to support the contract: <br>
     * 1. When the check=false is set, the connection is not checked, otherwise the exception is thrown when disconnection <br>
     * 2. Support username:password authority authentication on URL.<br>
     * 3. Support the backup=10.20.153.10 candidate registry cluster address.<br>
     * 4. Support file=registry.cache local disk file cache.<br>
     * 5. Support the timeout=1000 request timeout setting.<br>
     * 6. Support session=60000 session timeout or expiration settings.<br>
     *
     * @param url Registry address, is not allowed to be empty
     * @return Registry reference, never return empty value
     */
    @Adaptive({"protocol"}) //url带有protocol参数
    Registry getRegistry(URL url);

}

类图