Dubbo之ZookeeperRegistry源码分析

ZookeeperRegistry的作用

ZookeeperRegistry是dubbo中常用的注册中心实现，它主要作用通过Zookeeper的目录监听机制，让消费者能够实时得到在线的提供者列表。并且一些服务治理的功能也是通过zookeeper这个监听特性巧妙的完成。

在具体讲解ZookeeperRegistry的相关源码之前，先来分析下dubbo在zookeeper的目录结构以及dubbo如何利用这个特性

Zookeeper目录结构

dubbo在zookeeper建立的目录是基于接口的，大致如下

image.png

针对每个接口节点会存在以下4个子节点

节点名	作用	子节点是否持久节点
consumers	存储消费者节点url	是
configuators	存储override或者absent url，用于服务治理	否
routers	用于设置路由url，用于服务治理	否
providers	存储在线提供者url	否

consumer节点存在的意义并不大，主要还是为了做监控
其他三个节点，都会设置被相应的监听器，发生改变时，会触发特定事件

Dubbo对Zookeeper监听机制的利用

Dubbo中通过ZookeeperClient的实现类来对zookeeper进行操作

image.png

ZookeeperClient提供设置两种监听器的方法，对应子节点监听器和状态监听器，这里我们关注子节点监听器ChildListener

public interface ChildListener {

    /**
     *
     * @param path 监听的节点
     * @param children 监听的节点的所有子节点
     */
    void childChanged(String path, List<String> children);

}

ZookeeperClient有两种实现，第一种通过官方提供的jar包，第二个通过Apache的Curator框架，默认使用第二种，我们讲解的也是Curator的对应实现
添加子节点监听器的方法为addChildListener

public List<String> addChildListener(String path, final ChildListener listener) {
        //对listener做缓存，因为ChildListener是dubbo提供的监听器接口，需要转换为cruator的监听器接口
        ConcurrentMap<ChildListener, TargetChildListener> listeners = childListeners.get(path);
        if (listeners == null) {
            childListeners.putIfAbsent(path, new ConcurrentHashMap<ChildListener, TargetChildListener>());
            listeners = childListeners.get(path);
        }
        TargetChildListener targetListener = listeners.get(listener);
        if (targetListener == null) {
            //createTargetChildListener会对监听器进行转换
            listeners.putIfAbsent(listener, createTargetChildListener(path, listener));
            targetListener = listeners.get(listener);
        }
        return addTargetChildListener(path, targetListener);
    }

Dubbo底层封装了2套Zookeeper API，所以通过ChildListener抽象了监听器，但是在实际调用时会通过createTargetChildListener转为对应框架的监听器实现
addTargetChildListener方法在添加监听器之后会返回监听path当前的所有的子节点

public List<String> addTargetChildListener(String path, CuratorWatcher listener) {
        try {
            //添加监听，并且返回这个目录当前所有子节点
            //这种监听方式是一次性的，在listener实现中会再次执行监听逻辑
            return client.getChildren().usingWatcher(listener).forPath(path);
        } catch (NoNodeException e) {
            return null;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

上述代码需要注意监听是一次性的，其实curator提供了TreeCache用作永久性的监听，这边不用到这个特性，应该是为了和官方API保持一致吧。
接下去看下Cruator监听器的封装

public CuratorWatcher createTargetChildListener(String path, ChildListener listener) {
        return new CuratorWatcherImpl(listener);
    }

private class CuratorWatcherImpl implements CuratorWatcher {

        private volatile ChildListener listener;

        public CuratorWatcherImpl(ChildListener listener) {
            this.listener = listener;
        }

        public void unwatch() {
            this.listener = null;
        }

        @Override
        public void process(WatchedEvent event) throws Exception {
            if (listener != null) {
                String path = event.getPath() == null ? "" : event.getPath();
                listener.childChanged(path,
                        // if path is null, curator using watcher will throw NullPointerException.
                        // if client connect or disconnect to server, zookeeper will queue
                        // watched event(Watcher.Event.EventType.None, .., path = null).
                        StringUtils.isNotEmpty(path)
                                //再次设置监听，并且把监听path的所有子节点传入childChanged方法
                                ? client.getChildren().usingWatcher(this).forPath(path)
                                : Collections.<String>emptyList());
            }
        }
    }

可以看到listener的触发逻辑以及入参来源

源码分析

image.png

通过ZookeeperRegistry的类继承图，逐上而下的分析源码

Registry接口

public interface Registry extends Node, RegistryService {
}

Registry继承Node和RegistryService两个接口，本身不提供接口方法

public interface Node {

    /**
     * get url.
     *
     * @return url.
     */
    URL getUrl();

    /**
     * is available.
     *
     * @return available.
     */
    boolean isAvailable();

    /**
     * destroy.
     */
    void destroy();

}

Node约束了三个生命周期相关的方法
getUrl用于获取当前组件的url配置
isAvailable检测组件是否可用
destroy用于销毁组件

public interface RegistryService {

   
    void register(URL url);

    void unregister(URL url);

    void subscribe(URL url, NotifyListener listener);

    void unsubscribe(URL url, NotifyListener listener);

    List<URL> lookup(URL url);

}

RegistryService规定了和注册中心相关的方法
register和unregister用于提供者向注册中心注册提供者url
subscribe和unsubscribe用于消费者向对应接口目录注册监听
lookup用于查找查找url，通过消费者url查找提供者url以及服务治理有关的url

AbstractRegistry

主要提供接口提供者本地缓存功能
以及基础register，unregister，subscribe，unsubscribe，notify，lookup，recover逻辑

register，unregister会(接触)注册提供者url，主要操作

private final Set<URL> registered = new ConcurrentHashSet<URL>();

subscribe，unsubscribe则会针对特定url提供监听，主要操作

private final ConcurrentMap<URL, Set<NotifyListener>> subscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();

notify方法会缓存最近通知的url到notified以及触发listener回调

/**
     * 这个方法不会直接触发，被FailbackRegistry重载
     * FailbackRegistry增加failback逻辑后，还是会调用这个方法
     * @param url
     * @param listener
     * @param urls
     */
    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>>();
        //根据url的category进行分类
        for (URL u : urls) {
            if (UrlUtils.isMatch(url, u)) {
                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;
        }
        //下面操作notified缓存
        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();
            //对notified内容进行覆盖，相当于会保存上一次的通知
            categoryNotified.put(category, categoryList);
            //每次通知后会刷新本地缓存
            saveProperties(url);
            //进行listener回调，每种category的url分别回调一次
            listener.notify(categoryList);
        }
    }

这个类的recover方法不分析，因为FailbackRegistry完全重写了这个方法

FailbackRegistry

FailbackRegistry重载了AbstractRegistry中的subscribe，unsubscribe，register，unregister，notify方法，在AbstractRegistry的基础上提供了失败重试机制，并且暴露模板方法doRegister，doUnregister，doSubscribe，doUnsubscribe让不同类型的注册中心实现。doNotify还是默认父类的逻辑。
同时也重载了recover方法，通过FailbackRegistry的重试机制实现recover

以registry方法作为样例看下添加的重试机制

/**
     * register行为，提供者使用
     * 在AbstractRegistry的基础上，增加失败重试机制
     * @param url
     */
    @Override
    public void register(URL url) {
        super.register(url);
        //这里成功，会删除failedRegistered，failedUnregistered中的url
        failedRegistered.remove(url);
        failedUnregistered.remove(url);
        try {
            // Sending a registration request to the server side
            //具体register逻辑交给子类实现
            doRegister(url);
        } catch (Exception e) {
            Throwable t = e;

            // If the startup detection is opened, the Exception is thrown directly.
            //如果注册中心或者提供者url的check为false的话，跳过抛出异常
            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);
            }

            // Record a failed registration request to a failed list, retry regularly
            //加入到失败重试集合
            failedRegistered.add(url);
        }
    }

注册失败后会把需要注册重试的url放入failedRegistered集合
然后在FailbackRegistry构造函数中起的定时任务会进行重试

public FailbackRegistry(URL url) {
        super(url);
        this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
        //重试的定时线程，使用future用于取消
        this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                // Check and connect to the registry
                try {
                    retry();
                } catch (Throwable t) { // Defensive fault tolerance
                    logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
                }
            }
        }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
    }

retry方法的具体逻辑，就是循环遍历这些失败集合，然后调用doXXX方法进行重试

recover方法会在和Zookeeper重连时触发，在断连状态下，dubbo进程内的注册，订阅行为是会被缓存下来的，然后对所有缓存的url进行重新注册，订阅。
这边有个细节点，可以看到failedRegistered这些集合使用的都是线程安全的集合

private final Set<URL> failedRegistered = new ConcurrentHashSet<URL>();

因为recover，retry这两个操作还是存在资源竞争的，但不仅限于这两个操作

ZookeeperRegistry

ZookeeperRegistry的工作就是通过Zookeeper API实现doRegister,doUnregister,doSubscribe,doUnsubscribe具体逻辑

首先来看下ZookeeperRegistry的构造函数，做的主要工作是初始化zk客户端

public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("registry address == null");
        }
        //如果不进行配置，默认dubbo根目录就是/dubbo
        String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
        if (!group.startsWith(Constants.PATH_SEPARATOR)) {
            group = Constants.PATH_SEPARATOR + group;
        }
        this.root = group;
        zkClient = zookeeperTransporter.connect(url);
        //zookeeper添加重连回调，会触发recover方法，进行失败任务重试
        //为什么FailbackRegistry都是用线程安全的集合，因为在这里存在线程竞争资源
        zkClient.addStateListener(new StateListener() {
            @Override
            public void stateChanged(int state) {
                if (state == RECONNECTED) {
                    try {
                        recover();
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }
            }
        });
    }

使用zookeeperTransporter扩展点加载zk客户端实现，默认为Curator框架

@SPI("curator")
public interface ZookeeperTransporter {

    @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
    ZookeeperClient connect(URL url);

}

注册/取消注册实现

然后再来看doRegister和doUnregister方法，对于zk来说，就是创建目录呗

/**
     * 注册的逻辑，就是在zookeeper创建节点，节点路径为toUrlPath(url)
     * 具体格式为 /{group}/{interfaceName}/{category}/{url.toFullString}
     * DYNAMIC_KEY表示是否创建永久节点，true表示不是，断开连接后会消失，所以需要进行recover
     * @param url
     */
    @Override
    protected void doRegister(URL url) {
        try {
            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);
        }
    }

/**
     * 取消注册，就是删除那个节点
     * @param url
     */
    @Override
    protected void doUnregister(URL url) {
        try {
            zkClient.delete(toUrlPath(url));
        } catch (Throwable e) {
            throw new RpcException("Failed to unregister " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

需要注意下节点的路径生成格式，也就是toUrlPath(url)方法，格式为 /{group}/{interfaceName}/{category}/{url.toFullString}，
group一般不配置的话为dubbo，
interfaceName对应具体接口，
category开始就讲过，分为consumers,configuators,routers,providers
url.toFullString就是我们的url配置

对于registry来讲category=providers

取消注册就是对应删除那个节点

订阅/取消订阅实现

订阅的行为对于消费者来讲，用于获取providers和routers，用于得到路由后的提供者
对于提供者来讲，订阅configuators，通过新的配置重新暴露
在ZookeeperRegistry，我们只关注如何进行订阅，具体监听器的作用，在用到的模块再讲
doSubscribe方法支持订阅全局和订阅特定接口
如果interface=*，即订阅全局，对于新增和已存在的所有接口的改动都会触发回调
如果interface=特定接口，那么只有这个接口的子节点改变时，才触发回调

@Override
    protected void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
                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);
                                    //如果consumer的interface为*，会订阅每一个url，会触发另一个分支的逻辑
                                    //这里是用来对/dubbo下面提供者新增时的回调，相当于增量
                                    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);
                        //如果consumer的interface为*，会订阅每一个url，会触发另一个分支的逻辑
                        //这里的逻辑只执行一次，一次全量
                        subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
                                Constants.CHECK_KEY, String.valueOf(false)), listener);
                    }
                }
            } else {
                //这边是针对明确interface的订阅逻辑
                List<URL> urls = new ArrayList<URL>();
                //针对每种category路径进行监听
                for (String path : toCategoriesPath(url)) {
                    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) {
                                ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                            }
                        });
                        zkListener = listeners.get(listener);
                    }
                    //创建节点
                    zkClient.create(path, false);
                    //增加回调
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) {
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                //如果有子节点，直接进行触发一次，对应AbstractRegsitry的lookup方法
                //意思就是第一次订阅，如果订阅目录存在子节点，直接会触发一次
                notify(url, listener, urls);
            }
        } catch (Throwable e) {
            throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

这边需要注意一点的是，每次进行订阅，最重要的第一次，会使用当前订阅节点的子节点数据触发一次notify，执行对应监听器逻辑，这个在后面RegistryDirectory中会用到这个特性

取消订阅没什么好讲的，删除订阅数据即可

讲了这么多，对于lookup方法，使用消费者查找提供者的逻辑其实也很简单。使用消费者url构造出zk中provider的目录，然后返回所有子节点即可

/**
     * 查找消费者url 对应 提供者url实现
     * 这边的url为消费者url
     * @param url
     * @return
     */
    @Override
    public List<URL> lookup(URL url) {
        if (url == null) {
            throw new IllegalArgumentException("lookup url == null");
        }
        try {
            List<String> providers = new ArrayList<String>();
            //返回inteface下面所有category的url
            for (String path : toCategoriesPath(url)) {
                List<String> children = zkClient.getChildren(path);
                if (children != null) {
                    providers.addAll(children);
                }
            }
            //返回匹配的url
            return toUrlsWithoutEmpty(url, providers);
        } catch (Throwable e) {
            throw new RpcException("Failed to lookup " + url + " from zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

总结

1.Zookeeper监听器的妙用，在Elasticjob也是使用到了这个特性，进行任务触发
2.通过zookeeperTransporter以及ZookeeperClient对Zookeeper操作进行抽象，进而支持两种zookeeper客户端框架。包括在remoting模块也是采用这种设计模式，和底层框架解耦。
3.Zookeeper默认的监听是一次性的，Curator框架实现了永久监听，但是dubbo没用到Curator这个特性。
4.写完这部分，Dirctory模块就比较容易写下去了，东西太多，有些地方的理解肯定存在偏差，希望读者能多多交流

最后

希望大家关注下我的公众号

image