Dubbo 的注册中心虽然提供了多种实现，但生产上的事实标准基本上都是 基于 Zookeeper 实现的。这种注册中心的实现方法也是 Dubbo 最为推荐的。为了易于理解 Zookeeper 在 Dubbo 中的应用，我们先简单看一下 zookeeper。

由于 Dubbo 是一个分布式 RPC 开源框架，各服务之间单独部署，往往会出现资源之间数据不一致的问题，比如：某一个服务增加或减少了几台机器，某个服务提供者变更了服务地址，那么服务消费者是很难感知到这种变化的。而 Zookeeper 本身就有保证分布式数据一致性的特性。那么 Dubbo 服务是如何被 Zookeeper 的数据结构存储管理的呢，zookeeper 采用的是树形结构来组织数据节点，它类似于一个标准的文件系统，如下图所示。

avatar

该图展示了 dubbo 在 zookeeper 中存储的形式以及节点层级。dubbo 的 Root 层是根目录，通过的“group”来设置 zookeeper 的根节点，缺省值是“dubbo”。Service 层是服务接口的全名。Type 层是分类，一共有四种分类，分别是 providers 服务提供者列表、consumers 服务消费者列表、routes 路由规则列表、configurations 配置规则列表。URL 层 根据不同的 Type 目录：可以有服务提供者 URL 、服务消费者 URL 、路由规则 URL 、配置规则 URL 。不同的 Type 关注的 URL 不同。

zookeeper 以斜杠来分割每一层的 znode 节点，比如第一层根节点 dubbo 就是“/dubbo”，而第二层的 Service 层就是/dubbo/com.foo.Barservice，zookeeper 的每个节点通过路径来表示以及访问，例如服务提供者启动时，向/dubbo/com.foo.Barservice/providers 目录下写入自己的 URL 地址。

dubbo-registry-zookeeper 模块的工程结构如下图所示，里面就俩类，非常简单。

avatar

ZookeeperRegistry

该类继承了 FailbackRegistry 抽象类，针对注册中心核心的 服务注册、服务订阅、取消注册、取消订阅，查询注册列表进行展开，这里用到了 模板方法设计模式，FailbackRegistry 中定义了 register()、subscribe()等模板方法和 doRegister()、doSubscribe()抽象方法，ZookeeperRegistry 基于 zookeeper 对这些抽象方法进行了实现。其实你会发现 zookeeper 虽然是最被推荐的，反而它的实现逻辑相对简单，因为调用了 zookeeper 服务组件，很多的逻辑不需要在 dubbo 中自己去实现。

  1. /*
  2.  * Licensed to the Apache Software Foundation (ASF) under one or more
  3.  * contributor license agreements.  See the NOTICE file distributed with
  4.  * this work for additional information regarding copyright ownership.
  5.  * The ASF licenses this file to You under the Apache License, Version 2.0
  6.  * (the "License"); you may not use this file except in compliance with
  7.  * the License.  You may obtain a copy of the License at
  8.  *
  9.  *     http://www.apache.org/licenses/LICENSE-2.0
  10.  *
  11.  * Unless required by applicable law or agreed to in writing, software
  12.  * distributed under the License is distributed on an "AS IS" BASIS,
  13.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14.  * See the License for the specific language governing permissions and
  15.  * limitations under the License.
  16.  */
  17. package com.alibaba.dubbo.registry.zookeeper;
  19. import com.alibaba.dubbo.common.Constants;
  20. import com.alibaba.dubbo.common.URL;
  21. import com.alibaba.dubbo.common.logger.Logger;
  22. import com.alibaba.dubbo.common.logger.LoggerFactory;
  23. import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
  24. import com.alibaba.dubbo.common.utils.UrlUtils;
  25. import com.alibaba.dubbo.registry.NotifyListener;
  26. import com.alibaba.dubbo.registry.support.FailbackRegistry;
  27. import com.alibaba.dubbo.remoting.zookeeper.ChildListener;
  28. import com.alibaba.dubbo.remoting.zookeeper.StateListener;
  29. import com.alibaba.dubbo.remoting.zookeeper.ZookeeperClient;
  30. import com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter;
  31. import com.alibaba.dubbo.rpc.RpcException;
  33. import java.util.ArrayList;
  34. import java.util.List;
  35. import java.util.Set;
  36. import java.util.concurrent.ConcurrentHashMap;
  37. import java.util.concurrent.ConcurrentMap;
  39. /**
  40.  * ZookeeperRegistry
  41.  *
  42.  * Zookeeper Registry 实现类
  43.  */
  44. public class ZookeeperRegistry extends FailbackRegistry {
  46.     private final static Logger logger = LoggerFactory.getLogger(ZookeeperRegistry.class);
  48.     /**
  49.      * 默认端口
  50.      */
  51.     private final static int DEFAULT_ZOOKEEPER_PORT = 2181;
  52.     /**
  53.      * 默认 Zookeeper 根节点
  54.      */
  55.     private final static String DEFAULT_ROOT = "dubbo";
  57.     /**
  58.      * Zookeeper 根节点
  59.      */
  60.     private final String root;
  61.     /**
  62.      * Service 接口全名集合
  63.      */
  64.     private final Set<String> anyServices = new ConcurrentHashSet<String>();
  65.     /**
  66.      * 监听器集合
  67.      */
  68.     private final ConcurrentMap<URL, ConcurrentMap<NotifyListener, ChildListener>> zkListeners = new ConcurrentHashMap<URL, ConcurrentMap<NotifyListener, ChildListener>>();
  69.     /**
  70.      * Zookeeper 客户端
  71.      */
  72.     private final ZookeeperClient zkClient;
  74.     public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
  75.         super(url);
  76.         if (url.isAnyHost()) {
  77.             throw new IllegalStateException("registry address == null");
  78.         }
  79.         // 获得 Zookeeper 根节点
  80.         String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT); // `url.parameters.group` 参数值
  81.         if (!group.startsWith(Constants.PATH_SEPARATOR)) {
  82.             group = Constants.PATH_SEPARATOR + group;
  83.         }
  84.         this.root = group;
  85.         // 创建 Zookeeper Client
  86.         zkClient = zookeeperTransporter.connect(url);
  87.         // 添加 StateListener 对象。该监听器，在重连时，调用恢复方法。
  88.         zkClient.addStateListener(new StateListener() {
  89.             public void stateChanged(int state) {
  90.                 if (state == RECONNECTED) {
  91.                     try {
  92.                         recover();
  93.                     } catch (Exception e) {
  94.                         logger.error(e.getMessage(), e);
  95.                     }
  96.                 }
  97.             }
  98.         });
  99.     }
  101.     // 目前只有测试方法使用
  102.     static String appendDefaultPort(String address) {
  103.         if (address != null && address.length() > 0) {
  104.             int i = address.indexOf(':');
  105.             if (i < 0) {
  106.                 return address + ":" + DEFAULT_ZOOKEEPER_PORT;
  107.             } else if (Integer.parseInt(address.substring(i + 1)) == 0) {
  108.                 return address.substring(0, i + 1) + DEFAULT_ZOOKEEPER_PORT;
  109.             }
  110.         }
  111.         return address;
  112.     }
  114.     @Override
  115.     public boolean isAvailable() {
  116.         return zkClient.isConnected();
  117.     }
  119.     @Override
  120.     public void destroy() {
  121.         // 调用父方法，取消注册和订阅
  122.         super.destroy();
  123.         try {
  124.             // 关闭 Zookeeper 客户端连接
  125.             zkClient.close();
  126.         } catch (Exception e) {
  127.             logger.warn("Failed to close zookeeper client " + getUrl() + ", cause: " + e.getMessage(), e);
  128.         }
  129.     }
  131.     @Override
  132.     protected void doRegister(URL url) {
  133.         try {
  134.             zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
  135.         } catch (Throwable e) {
  136.             throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
  137.         }
  138.     }
  140.     @Override
  141.     protected void doUnregister(URL url) {
  142.         try {
  143.             zkClient.delete(toUrlPath(url));
  144.         } catch (Throwable e) {
  145.             throw new RpcException("Failed to unregister " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
  146.         }
  147.     }
  149.     @Override
  150.     protected void doSubscribe(final URL url, final NotifyListener listener) {
  151.         try {
  152.             // 处理所有 Service 层的发起订阅，例如监控中心的订阅
  153.             if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
  154.                 String root = toRootPath();
  155.                 // 获得 url 对应的监听器集合
  156.                 ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
  157.                 if (listeners == null) { // 不存在，进行创建
  158.                     zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
  159.                     listeners = zkListeners.get(url);
  160.                 }
  161.                 // 获得 ChildListener 对象
  162.                 ChildListener zkListener = listeners.get(listener);
  163.                 if (zkListener == null) { // 不存在 ChildListener 对象，进行创建 ChildListener 对象
  164.                     listeners.putIfAbsent(listener, new ChildListener() {
  165.                         public void childChanged(String parentPath, List<String> currentChilds) {
  166.                             for (String child : currentChilds) {
  167.                                 child = URL.decode(child);
  168.                                 // 新增 Service 接口全名时（即新增服务），发起该 Service 层的订阅
  169.                                 if (!anyServices.contains(child)) {
  170.                                     anyServices.add(child);
  171.                                     subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child,
  172.                                             Constants.CHECK_KEY, String.valueOf(false)), listener);
  173.                                 }
  174.                             }
  175.                         }
  176.                     });
  177.                     zkListener = listeners.get(listener);
  178.                 }
  179.                 // 创建 Service 节点。该节点为持久节点。
  180.                 zkClient.create(root, false);
  181.                 // 向 Zookeeper ，Service 节点，发起订阅
  182.                 List<String> services = zkClient.addChildListener(root, zkListener);
  183.                 // 首次全量数据获取完成时，循环 Service 接口全名数组，发起该 Service 层的订阅
  184.                 if (services != null && !services.isEmpty()) {
  185.                     for (String service : services) {
  186.                         service = URL.decode(service);
  187.                         anyServices.add(service);
  188.                         subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
  189.                                 Constants.CHECK_KEY, String.valueOf(false)), listener);
  190.                     }
  191.                 }
  192.             // 处理指定 Service 层的发起订阅，例如服务消费者的订阅
  193.             } else {
  194.                 // 子节点数据数组
  195.                 List<URL> urls = new ArrayList<URL>();
  196.                 // 循环分类数组
  197.                 for (String path : toCategoriesPath(url)) {
  198.                     // 获得 url 对应的监听器集合
  199.                     ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
  200.                     if (listeners == null) { // 不存在，进行创建
  201.                         zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
  202.                         listeners = zkListeners.get(url);
  203.                     }
  204.                     // 获得 ChildListener 对象
  205.                     ChildListener zkListener = listeners.get(listener);
  206.                     if (zkListener == null) { // 不存在 ChildListener 对象，进行创建 ChildListener 对象
  207.                         listeners.putIfAbsent(listener, new ChildListener() {
  208.                             public void childChanged(String parentPath, List<String> currentChilds) {
  209.                                 // 变更时，调用 `#notify(...)` 方法，回调 NotifyListener
  210.                                 ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
  211.                             }
  212.                         });
  213.                         zkListener = listeners.get(listener);
  214.                     }
  215.                     // 创建 Type 节点。该节点为持久节点。
  216.                     zkClient.create(path, false);
  217.                     // 向 Zookeeper ，PATH 节点，发起订阅
  218.                     List<String> children = zkClient.addChildListener(path, zkListener);
  219.                     // 添加到 `urls` 中
  220.                     if (children != null) {
  221.                         urls.addAll(toUrlsWithEmpty(url, path, children));
  222.                     }
  223.                 }
  224.                 // 首次全量数据获取完成时，调用 `#notify(...)` 方法，回调 NotifyListener
  225.                 notify(url, listener, urls);
  226.             }
  227.         } catch (Throwable e) {
  228.             throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
  229.         }
  230.     }
  232.     @Override
  233.     protected void doUnsubscribe(URL url, NotifyListener listener) {
  234.         ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
  235.         if (listeners != null) {
  236.             ChildListener zkListener = listeners.get(listener);
  237.             if (zkListener != null) {
  238.                 // 向 Zookeeper ，移除订阅
  239.                 zkClient.removeChildListener(toUrlPath(url), zkListener);
  240.             }
  241.         }
  242.     }
  244.     @Override
  245.     public List<URL> lookup(URL url) {
  246.         if (url == null) {
  247.             throw new IllegalArgumentException("lookup url == null");
  248.         }
  249.         try {
  250.             // 循环分类数组，获得所有的 URL 数组
  251.             List<String> providers = new ArrayList<String>();
  252.             for (String path : toCategoriesPath(url)) {
  253.                 List<String> children = zkClient.getChildren(path);
  254.                 if (children != null) {
  255.                     providers.addAll(children);
  256.                 }
  257.             }
  258.             // 匹配
  259.             return toUrlsWithoutEmpty(url, providers);
  260.         } catch (Throwable e) {
  261.             throw new RpcException("Failed to lookup " + url + " from zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
  262.         }
  263.     }
  265.     /**
  266.      * 获得根目录
  267.      * @return 路径
  268.      */
  269.     private String toRootDir() {
  270.         if (root.equals(Constants.PATH_SEPARATOR)) {
  271.             return root;
  272.         }
  273.         return root + Constants.PATH_SEPARATOR;
  274.     }
  276.     /**
  277.      * @return 根路径
  278.      */
  279.     private String toRootPath() {
  280.         return root;
  281.     }
  283.     /**
  284.      * 获得服务路径
  285.      *
  286.      * Root + Type
  287.      *
  288.      * @param url URL
  289.      * @return 服务路径
  290.      */
  291.     private String toServicePath(URL url) {
  292.         String name = url.getServiceInterface();
  293.         if (Constants.ANY_VALUE.equals(name)) {
  294.             return toRootPath();
  295.         }
  296.         return toRootDir() + URL.encode(name);
  297.     }
  299.     /**
  300.      * 获得分类路径数组
  301.      *
  302.      * Root + Service + Type
  303.      *
  304.      * @param url URL
  305.      * @return 分类路径数组
  306.      */
  307.     private String[] toCategoriesPath(URL url) {
  308.         // 获得分类数组
  309.         String[] categories;
  310.         if (Constants.ANY_VALUE.equals(url.getParameter(Constants.CATEGORY_KEY))) { // * 时，
  311.             categories = new String[]{Constants.PROVIDERS_CATEGORY, Constants.CONSUMERS_CATEGORY,
  312.                     Constants.ROUTERS_CATEGORY, Constants.CONFIGURATORS_CATEGORY};
  313.         } else {
  314.             categories = url.getParameter(Constants.CATEGORY_KEY, new String[]{Constants.DEFAULT_CATEGORY});
  315.         }
  316.         // 获得分类路径数组
  317.         String[] paths = new String[categories.length];
  318.         for (int i = 0; i < categories.length; i++) {
  319.             paths[i] = toServicePath(url) + Constants.PATH_SEPARATOR + categories[i];
  320.         }
  321.         return paths;
  322.     }
  324.     /**
  325.      * 获得分类路径
  326.      *
  327.      * Root + Service + Type
  328.      *
  329.      * @param url URL
  330.      * @return 分类路径
  331.      */
  332.     private String toCategoryPath(URL url) {
  333.         return toServicePath(url) + Constants.PATH_SEPARATOR + url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
  334.     }
  336.     /**
  337.      * 获得 URL 的路径
  338.      *
  339.      * Root + Service + Type + URL
  340.      *
  341.      * 被 {@link #doRegister(URL)} 和 {@link #doUnregister(URL)} 调用
  342.      *
  343.      * @param url URL
  344.      * @return 路径
  345.      */
  346.     private String toUrlPath(URL url) {
  347.         return toCategoryPath(url) + Constants.PATH_SEPARATOR + URL.encode(url.toFullString());
  348.     }
  350.     /**
  351.      * 获得 providers 中，和 consumer 匹配的 URL 数组
  352.      *
  353.      * @param consumer 用于匹配 URL
  354.      * @param providers 被匹配的 URL 的字符串
  355.      * @return 匹配的 URL 数组
  356.      */
  357.     private List<URL> toUrlsWithoutEmpty(URL consumer, List<String> providers) {
  358.         List<URL> urls = new ArrayList<URL>();
  359.         if (providers != null && !providers.isEmpty()) {
  360.             for (String provider : providers) {
  361.                 provider = URL.decode(provider);
  362.                 if (provider.contains("://")) { // 是 url
  363.                     URL url = URL.valueOf(provider); // 将字符串转化成 URL
  364.                     if (UrlUtils.isMatch(consumer, url)) { // 匹配
  365.                         urls.add(url);
  366.                     }
  367.                 }
  368.             }
  369.         }
  370.         return urls;
  371.     }
  373.     /**
  374.      * 获得 providers 中，和 consumer 匹配的 URL 数组
  375.      *
  376.      * 若不存在匹配，则创建 `empty://` 的 URL返回。通过这样的方式，可以处理类似服务提供者为空的情况。
  377.      *
  378.      * @param consumer 用于匹配 URL
  379.      * @param path 被匹配的 URL 的字符串
  380.      * @param providers 匹配的 URL 数组
  381.      * @return 匹配的 URL 数组
  382.      */
  383.     private List<URL> toUrlsWithEmpty(URL consumer, String path, List<String> providers) {
  384.         // 获得 providers 中，和 consumer 匹配的 URL 数组
  385.         List<URL> urls = toUrlsWithoutEmpty(consumer, providers);
  386.         // 若不存在匹配，则创建 `empty://` 的 URL返回
  387.         if (urls == null || urls.isEmpty()) {
  388.             int i = path.lastIndexOf('/');
  389.             String category = i < 0 ? path : path.substring(i + 1);
  390.             URL empty = consumer.setProtocol(Constants.EMPTY_PROTOCOL).addParameter(Constants.CATEGORY_KEY, category);
  391.             urls.add(empty);
  392.         }
  393.         return urls;
  394.     }
  395. }

ZookeeperRegistryFactory

ZookeeperRegistryFactory 继承了 AbstractRegistryFactory 抽象类，实现了其中的抽象方法 如 createRegistry()，源码如下。

  1. /**
  2.  * Zookeeper Registry 工厂
  3.  */
  4. public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
  6.     /**
  7.      * Zookeeper 工厂
  8.      */
  9.     private ZookeeperTransporter zookeeperTransporter;
  11.     /**
  12.      * 设置 Zookeeper 工厂，该方法，通过 Dubbo SPI 注入
  13.      *
  14.      * @param zookeeperTransporter Zookeeper 工厂对象
  15.      */
  16.     public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
  17.         this.zookeeperTransporter = zookeeperTransporter;
  18.     }
  20.     @Override
  21.     public Registry createRegistry(URL url) {
  22.         return new ZookeeperRegistry(url, zookeeperTransporter);
  23.     }
  24. }