Dubbo源码学习--Zookeeper注册中心（二）

╰半橙微兮° 2022-05-29 11:22 180阅读 0赞

[Zookeeper][] 是 Apacahe Hadoop 的子项目，是一个树型的目录服务，支持变更推送，适合作为 Dubbo 服务的注册中心，工业强度较高，可用于生产环境，并推荐使用 [1][]。

![/user-guide/images/zookeeper.jpg][user-guide_images_zookeeper.jpg]

流程说明：

*  服务提供者启动时: 向 `/dubbo/com.foo.BarService/providers` 目录下写入自己的 URL 地址
 *  服务消费者启动时: 订阅 `/dubbo/com.foo.BarService/providers` 目录下的提供者 URL 地址。并向 `/dubbo/com.foo.BarService/consumers` 目录下写入自己的 URL 地址
 *  监控中心启动时: 订阅 `/dubbo/com.foo.BarService` 目录下的所有提供者和消费者 URL 地址。

支持以下功能：

*  当提供者出现断电等异常停机时，注册中心能自动删除提供者信息
 *  当注册中心重启时，能自动恢复注册数据，以及订阅请求
 *  当会话过期时，能自动恢复注册数据，以及订阅请求
 *  当设置 `<dubbo:registry check="false" />` 时，记录失败注册和订阅请求，后台定时重试
 *  可通过 `<dubbo:registry username="admin" password="1234" />` 设置 zookeeper 登录信息
 *  可通过 `<dubbo:registry group="dubbo" />` 设置 zookeeper 的根节点，不设置将使用无根树
 *  支持 `*` 号通配符 `<dubbo:reference group="*" version="*" />`，可订阅服务的所有分组和所有版本的提供者

## 使用 ##

在 provider 和 consumer 中增加 zookeeper 客户端 jar 包依赖：

<dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.3.3</version>
    </dependency>

或直接[下载][Link 1]。

Dubbo 支持 zkclient 和 curator 两种 Zookeeper 客户端实现：

### 使用 zkclient 客户端 ###

从 `2.2.0` 版本开始缺省为 zkclient 实现，以提升 zookeeper 客户端的健状性。[zkclient][] 是 Datameer 开源的一个 Zookeeper 客户端实现。

缺省配置：

<dubbo:registry ... client="zkclient" />

或：

dubbo.registry.client=zkclient

或：

zookeeper://10.20.153.10:2181?client=zkclient

需依赖或直接[下载][Link 2]：

<dependency>
        <groupId>com.github.sgroschupf</groupId>
        <artifactId>zkclient</artifactId>
        <version>0.1</version>
    </dependency>

### 使用 curator 客户端 ###

从 `2.3.0` 版本开始支持可选 curator 实现。[Curator][] 是 Netflix 开源的一个 Zookeeper 客户端实现。

如果需要改为 curator 实现，请配置：

<dubbo:registry ... client="curator" />

或：

dubbo.registry.client=curator

或：

zookeeper://10.20.153.10:2181?client=curator

需依赖或直接[下载][Link 3]：

<dependency>
        <groupId>com.netflix.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>1.1.10</version>
    </dependency>

Zookeeper 单机配置:

<dubbo:registry address="zookeeper://10.20.153.10:2181" />

或：

<dubbo:registry protocol="zookeeper" address="10.20.153.10:2181" />

Zookeeper 集群配置：

<dubbo:registry address="zookeeper://10.20.153.10:2181?backup=10.20.153.11:2181,10.20.153.12:2181" />

或：

<dubbo:registry protocol="zookeeper" address="10.20.153.10:2181,10.20.153.11:2181,10.20.153.12:2181" />

同一 Zookeeper，分成多组注册中心:

<dubbo:registry id="chinaRegistry" protocol="zookeeper" address="10.20.153.10:2181" group="china" />
    <dubbo:registry id="intlRegistry" protocol="zookeeper" address="10.20.153.10:2181" group="intl" />

Zookeeper注册中心的提供RegistryFactory的实现类为ZookeeperRegistryFactory，其是一个Registry的工厂类，产生ZookeeperRegistry实例。

public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
    
        private ZookeeperTransporter zookeeperTransporter;
    
        public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
            this.zookeeperTransporter = zookeeperTransporter;
        }
    
    	//创建ZookeeperRegistry实例
        public Registry createRegistry(URL url) {
            return new ZookeeperRegistry(url, zookeeperTransporter);
        }
    }

在ZookeeperRegistry中实现了如下接口：  
（1）doRegister(URL url);//向注册中心注册服务  
（2）doUnregister(URL url);//取消服务注册  
（3）doSubscribe(URL url, NotifyListener listener);//向注册中心监听服务

（4）doUnsubscribe(URL url, NotifyListener listener);//取消监听注册中心服务

public class ZookeeperRegistry extends FailbackRegistry {
    
        private final static Logger logger = LoggerFactory.getLogger(ZookeeperRegistry.class);
    
        private final static int DEFAULT_ZOOKEEPER_PORT = 2181;
    
        private final static String DEFAULT_ROOT = "dubbo";
    
        private final String root;
    
        private final Set<String> anyServices = new ConcurrentHashSet<String>();
    
        private final ConcurrentMap<URL, ConcurrentMap<NotifyListener, ChildListener>> zkListeners = new ConcurrentHashMap<URL, ConcurrentMap<NotifyListener, ChildListener>>();
    
        private final ZookeeperClient zkClient;
    	
    	//构造函数中创建与Zookeeper服务的连接客户端
        public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
            super(url);
            if (url.isAnyHost()) {
                throw new IllegalStateException("registry address == null");
            }
            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);
            zkClient.addStateListener(new StateListener() {
                public void stateChanged(int state) {
                    if (state == RECONNECTED) {
                        try {
                            recover();
                        } catch (Exception e) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            });
        }
    
        static String appendDefaultPort(String address) {
            if (address != null && address.length() > 0) {
                int i = address.indexOf(':');
                if (i < 0) {
                    return address + ":" + DEFAULT_ZOOKEEPER_PORT;
                } else if (Integer.parseInt(address.substring(i + 1)) == 0) {
                    return address.substring(0, i + 1) + DEFAULT_ZOOKEEPER_PORT;
                }
            }
            return address;
        }
    
        public boolean isAvailable() {
            return zkClient.isConnected();
        }
    
        public void destroy() {
            super.destroy();
            try {
                zkClient.close();
            } catch (Exception e) {
                logger.warn("Failed to close zookeeper client " + getUrl() + ", cause: " + e.getMessage(), e);
            }
        }
    	
    	//将服务地址在Zookeeper中创建相关路径
        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);
            }
        }
    	
    	//将相关服务路径从Zookeeper中删除
        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);
            }
        }
    
    	//利用Zookeeper的监听功能，监听某个路径的变化
        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() {
                            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 {
                    List<URL> urls = new ArrayList<URL>();
                    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() {
                                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));
                        }
                    }
    				//监听服务变化，发出通知修改客户端
                    notify(url, listener, urls);
                }
            } catch (Throwable e) {
                throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
            }
        }
    
    	//取消监听器listener
        protected void doUnsubscribe(URL url, NotifyListener listener) {
            ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
            if (listeners != null) {
                ChildListener zkListener = listeners.get(listener);
                if (zkListener != null) {
                    zkClient.removeChildListener(toUrlPath(url), zkListener);
                }
            }
        }
    
        public List<URL> lookup(URL url) {
            if (url == null) {
                throw new IllegalArgumentException("lookup url == null");
            }
            try {
                List<String> providers = new ArrayList<String>();
                for (String path : toCategoriesPath(url)) {
                    List<String> children = zkClient.getChildren(path);
                    if (children != null) {
                        providers.addAll(children);
                    }
                }
                return toUrlsWithoutEmpty(url, providers);
            } catch (Throwable e) {
                throw new RpcException("Failed to lookup " + url + " from zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
            }
        }
    
    }

[Zookeeper]: http://zookeeper.apache.org/
[1]: http://dubbo.io/books/dubbo-user-book/references/registry/zookeeper.html#fn_1
[user-guide_images_zookeeper.jpg]: http://dubbo.io/books/dubbo-user-book/sources/images/zookeeper.jpg
[Link 1]: http://repo1.maven.org/maven2/org/apache/zookeeper/zookeeper
[zkclient]: https://github.com/sgroschupf/zkclient
[Link 2]: http://repo1.maven.org/maven2/com/github/sgroschupf/zkclient
[Curator]: https://github.com/Netflix/curator
[Link 3]: http://repo1.maven.org/maven2/com/netflix/curator/curator-framework