/**
 * RegistryDirectory的subscribe方法
 */
public void subscribe(URL url) {
    setConsumerUrl(url); //简单的设置RegistryDirectory成员变量consumerUrl
    //勾起成员变量Registry的subscribe方法，并将自身作为NotifyListener传入
    registry.subscribe(url, this); 
}

/**
 * 继承自FailbackRegistry的subscribe方法
 */ 
public void subscribe(URL url, NotifyListener listener) {
    //调用父类的订阅方法，
    super.subscribe(url, listener);
    //从failedSubscribed、failedUnsubscribed、failedNotified列表中移除该url对应的listener
    removeFailedSubscribed(url, listener);
    try {
        //调用由ZookeeperRegistry实现的doSubscribe方法
        doSubscribe(url, listener);
    } catch (Exception e) {
        //如果doSubscribe执行抛出异常
        Throwable t = e;
        //从加载的文件缓存Properties中获取url.getServiceKey对应的缓存
        List<URL> urls = getCacheUrls(url);
        if (urls != null && !urls.isEmpty()) {
            //如果读取到的缓存不为空，则对该url进行通知，通知的内容是文件缓存的内容
            notify(url, listener, urls);
            logger.error("Failed to subscribe ... Using cached list: ...");
        } else {
            //获取check标识，getUrl方法获取的是ZookeeperRegistry维护的registryUrl，
            //而url指的是服务消费者的URL，从它们的parameters字段获取check这个key的value；
            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 ... cause: ");
            } else {
                logger.error("Failed to subscribe url ... waiting for retry ...");
            }
        }
        //向成员变量 ConcurrentMap<URL, Set<NotifyListener>> failedSubscribed添加订阅失败的数据
        addFailedSubscribed(url, listener);
    }
}

/**
 * 由ZookeeperRegistry实现的doSubscribe方法
 */
protected void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
            //如果方法参数URL的paramters中key=interface为*，暂时先不讨论
            String root = toRootPath();
            ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
            if (listeners == null) {
                zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                listeners = zkListeners.get(url);
            }
            ChildListener zkListener = listeners.get(listener);
            if (zkListener == null) {
                listeners.putIfAbsent(listener, new ChildListener() {
                    @Override
                    public void childChanged(String parentPath, List<String> currentChilds) {
                        for (String child : currentChilds) {
                            child = URL.decode(child);
                            if (!anyServices.contains(child)) {
                                anyServices.add(child);
                                subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child,
                                                                           Constants.CHECK_KEY, String.valueOf(false)), listener);
                            }
                        }
                    }
                });
                zkListener = listeners.get(listener);
            }
            zkClient.create(root, false);
            List<String> services = zkClient.addChildListener(root, zkListener);
            if (services != null && !services.isEmpty()) {
                for (String service : services) {
                    service = URL.decode(service);
                    anyServices.add(service);
                    subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
                                                                 Constants.CHECK_KEY, String.valueOf(false)), listener);
                }
            }
        } else {
            //方法参数URL的paramters中key=interface不为*
            List<URL> urls = new ArrayList<URL>();
            //通过toCategoriesPath方法生成要订阅的ZK节点路径，以interface=AlarmTestService为例：
            //(1)/dubbo/com.bestpay.monitor.app.AlarmTestService/providers;
            //(2)/dubbo/com.bestpay.monitor.app.AlarmTestService/configurators;
            //(3)/dubbo/com.bestpay.monitor.app.AlarmTestService/routers;
            //然后遍历这3个路径
            for (String path : toCategoriesPath(url)) {
                //获取这个ZookeeperRegistry中的服务消费者URL对应的监听器Map
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                    listeners = zkListeners.get(url);
                }
                //获取NotifyListener，即RegistryDirectory对应的ChildListener。
                //一般来说RegistryDirectory与注册中心Registry和服务引用接口（如GenericService）绑定。
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) {
                    //如果没有ChildListener，则创建一个并设置到listeners这个map中
                    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);
                }
                //创建"XXXX/XXX/providers、configurators、providers"永久节点
                zkClient.create(path, false);
                //向Dubbo封装的ZkClient添加ChildListener
                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);
    }
}

/**
 * AbstractRegistry的subscribe方法
 */
public void subscribe(URL url, NotifyListener listener) {
    if (url == null) {
        throw new IllegalArgumentException("subscribe url == null");
    }
    if (listener == null) {
        throw new IllegalArgumentException("subscribe listener == null");
    }
    if (logger.isInfoEnabled()) {
        logger.info("Subscribe: " + url);
    }
    Set<NotifyListener> listeners = subscribed.get(url);
    if (listeners == null) {
        subscribed.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
        listeners = subscribed.get(url);
    }
    listeners.add(listener);
}

/**
 * 用于根据Curator客户端推送的连接状态，RECONNECTED进行恢复
 */
protected void recover() throws Exception {
    //获取恢复的注册中心的地址URL列表Set<URL>，getRegistered()获取成员变量Set<URL>
    Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());
    if (!recoverRegistered.isEmpty()) {
        if (logger.isInfoEnabled()) {
            logger.info("Recover register url " + recoverRegistered);
        }
        //将getRegistered()添加到成员变量Set<URL> failedRegistered中
        for (URL url : recoverRegistered) {
            failedRegistered.add(url);
        }
    }
    
    //获取恢复的订阅列表Map<URL, Set<NotifyListener>>，getSubscribed()获取成员变量subscribed
    Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>
                                                                            (getSubscribed());
    if (!recoverSubscribed.isEmpty()) {
        if (logger.isInfoEnabled()) {
            logger.info("Recover subscribe url " + recoverSubscribed.keySet());
        }
        for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {
            URL url = entry.getKey();
            //添加到成员变量failedSubscribed中
            for (NotifyListener listener : entry.getValue()) {
                addFailedSubscribed(url, listener);
            }
        }
    }
}

/**
 * 对于成员变量failedRegistered、failedUnregistered、failedSubscribed、
 * failedUnsubscribed、failedNotified进行重试，可以看到，如果重试成功则将
 * 其移出相应的重试列表，如果重试失败，则忽略异常等待下次重试
 */
protected void retry() {
    if (!failedRegistered.isEmpty()) {
        Set<URL> failed = new HashSet<URL>(failedRegistered);
        if (failed.size() > 0) {
            if (logger.isInfoEnabled()) {
                logger.info("Retry register " + failed);
            }
            try {
                for (URL url : failed) {
                    try {
                        doRegister(url);
                        failedRegistered.remove(url);
                    } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
                        logger.warn("Failed to retry register ... waiting for again, cause: ...");
                    }
                }
            } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
                logger.warn("Failed to retry register ... waiting for again, cause: ...");
            }
        }
    }
    if (!failedUnregistered.isEmpty()) {
        Set<URL> failed = new HashSet<URL>(failedUnregistered);
        if (!failed.isEmpty()) {
            if (logger.isInfoEnabled()) {
                logger.info("Retry unregister " + failed);
            }
            try {
                for (URL url : failed) {
                    try {
                        doUnregister(url);
                        failedUnregistered.remove(url);
                    } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
                        logger.warn("Failed to retry unregister ... waiting for again, cause: ...");
                    }
                }
            } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
                logger.warn("Failed to retry unregister ... waiting for again, cause: ");
            }
        }
    }
    if (!failedSubscribed.isEmpty()) {
        Map<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>(failedSubscribed);
        for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>
                                                         (failed).entrySet()) {
            if (entry.getValue() == null || entry.getValue().size() == 0) {
                failed.remove(entry.getKey());
            }
        }
        if (failed.size() > 0) {
            if (logger.isInfoEnabled()) {
                logger.info("Retry subscribe " + failed);
            }
            try {
                for (Map.Entry<URL, Set<NotifyListener>> entry : failed.entrySet()) {
                    URL url = entry.getKey();
                    Set<NotifyListener> listeners = entry.getValue();
                    for (NotifyListener listener : listeners) {
                        try {
                            doSubscribe(url, listener);
                            listeners.remove(listener);
                        } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
                            logger.warn("Failed to retry subscribe ... waiting for again, cause: ...");
                        }
                    }
                }
            } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
                logger.warn("Failed to retry subscribe ... waiting for again, cause: ");
            }
        }
    }
    if (!failedUnsubscribed.isEmpty()) {
        Map<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>
                                                                (failedUnsubscribed);
        for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>
                                                         (failed).entrySet()) {
            if (entry.getValue() == null || entry.getValue().isEmpty()) {
                failed.remove(entry.getKey());
            }
        }
        if (failed.size() > 0) {
            if (logger.isInfoEnabled()) {
                logger.info("Retry unsubscribe " + failed);
            }
            try {
                for (Map.Entry<URL, Set<NotifyListener>> entry : failed.entrySet()) {
                    URL url = entry.getKey();
                    Set<NotifyListener> listeners = entry.getValue();
                    for (NotifyListener listener : listeners) {
                        try {
                            doUnsubscribe(url, listener);
                            listeners.remove(listener);
                        } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
                            logger.warn("Failed to retry unsubscribe ... waiting for again, cause: ..");
                        }
                    }
                }
            } catch (Throwable t) { 
                logger.warn("Failed to retry unsubscribe ... waiting for again, cause: ...");
            }
        }
    }
    if (!failedNotified.isEmpty()) {
        Map<URL, Map<NotifyListener, List<URL>>> failed = new HashMap<URL, Map<NotifyListener, 
                                                    List<URL>>>(failedNotified);
        for (Map.Entry<URL, Map<NotifyListener, List<URL>>> entry : new HashMap<URL, Map<NotifyListener, 
                                                     List<URL>>>(failed).entrySet()) {
            if (entry.getValue() == null || entry.getValue().size() == 0) {
                failed.remove(entry.getKey());
            }
        }
        if (failed.size() > 0) {
            if (logger.isInfoEnabled()) {
                logger.info("Retry notify " + failed);
            }
            try {
                for (Map<NotifyListener, List<URL>> values : failed.values()) {
                    for (Map.Entry<NotifyListener, List<URL>> entry : values.entrySet()) {
                        try {
                            NotifyListener listener = entry.getKey();
                            List<URL> urls = entry.getValue();
                            listener.notify(urls);
                            values.remove(listener);
                        } catch (Throwable t) { 
                            logger.warn("Failed to retry notify ... waiting for again, cause: ...");
                        }
                    }
                }
            } catch (Throwable t) { 
                logger.warn("Failed to retry notify ... waiting for again, cause: ...");
            }
        }
    }
}

public ZookeeperClient connect(URL url) {
    return new CuratorZookeeperClient(url);
}

 public CuratorZookeeperClient(URL url) {
     super(url);
     try {
         CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
             .connectString(url.getBackupAddress())
             .retryPolicy(new RetryNTimes(1, 1000))
             .connectionTimeoutMs(5000);
         String authority = url.getAuthority();
         if (authority != null && authority.length() > 0) {
             builder = builder.authorization("digest", authority.getBytes());
         }
         client = builder.build();
         client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
             @Override
             public void stateChanged(CuratorFramework client, ConnectionState state) {
                 //回调自身维护的监听器List<StateListener>
                 if (state == ConnectionState.LOST) {
                     CuratorZookeeperClient.this.stateChanged(StateListener.DISCONNECTED);
                 } else if (state == ConnectionState.CONNECTED) {
                     CuratorZookeeperClient.this.stateChanged(StateListener.CONNECTED);
                 } else if (state == ConnectionState.RECONNECTED) {
                     CuratorZookeeperClient.this.stateChanged(StateListener.RECONNECTED);
                 }
             }
         });
         client.start();
     } catch (Exception e) {
         throw new IllegalStateException(e.getMessage(), e);
     }
 }