Dubbo学习（三）：注册中心

前言

这篇文章主要记录学习Dubbo注册中心的相关内容，包括：注册中心的工作原理；注册中心的数据结构；订阅发布的实现；缓存机制；重试机制；服务注册与发现流程中Dubbo使用了哪些设计模式。

注册中心概述

在Dubbo微服务体系中，注册中心是其核心的组件之一。Dubbo通过注册中心实现了分布式环境中各服务之间的注册与发现，是各个分布式节点之间的纽带。主要作用有：
（1）动态加入：一个服务提供者通过注册中心可以动态的把自己暴露给其他消费者，无需消费者逐个去更新配置文件。
（2）动态发现：一个消费者可以动态感知新的配置、路由规则和新的服务提供者，无需重启服务使之生效。
（3）动态调整：注册中心支持参数的动态调整，新参数自动更新到所有相关服务节点。
（4）统一配置：避免了本地配置导致每个服务的配置不一致问题。

Dubbo注册中心源码在dubbo-registry中，里面包含了五个子模块：

image.png

从dubbo-registry模块可以看出，dubbo主要包含四种注册中心的实现，分别是：Zookeeper、Redis、Simple、Multicast.

其中Zookeeper是官方推荐的注册中心实现，在生产环境中已经有大量实际使用，具体的实现在Dubbo的源码 dubbo-registry-zookeeper模块中。而Redis注册中心在稳定性方面相比ZK就差了一些，其稳定性主要是依赖Redis本身，所以被使用的应该也不多吧。Simple是一个基于内存的简单的注册中心实现，它本身就是一个标准的RPC服务，不支持集群，可能出现单点故障。Multicast模式则不需要启动任何注册中心，只要通过广播地址，就可以互相发现。服务提供者启动时，会广播自己的地址，消费者启动时，会广播订阅请求，服务提供者收到订阅请求，会根据配置广播或单播给订阅者。

Dubbo拥有良好的扩展性，如果以上注册中心都不满足需求，那么我们还可以基于RegistryFactory和Registry自行扩展。

工作流程

注册中心的总体流程比较简单，Dubbo官方也有比较详细的说明，总体流程如下：

image.png

Provider注册：Provider启动时，会向注册中心写入自己的元数据信息，同时会订阅配置元数据信息。
Consumer订阅：这里的订阅指Consumer启动时也会向注册中心写入自己的元数据信息，并订阅服务提供者、路由和配置元数据信息。
服务治理中心启动：dubbo-admin启动时，会同时订阅所有消费者、服务提供者、路由和配置元数据信息。
动态注册、发现：当有新的Provider加入或者有离开时，注册中心服务提供者目录会发生变化，变化信息会动态通知给消费者、服务治理中心。
监控中心采集：当Consumer发起调用时，会异步将调用、统计信息等上报给监控中心。

数据结构

注册中心的总体流程相同，但是不同的注册中心有不同的实现方式，其数据结构也不相同。Zookeeper、Redis等注册中心都实现了这个流程。由于有些注册中心并不常用，因此我们下面将只重点关注下Zookeeper与Redis两种实现的数据结构。

Zookeeper注册中心实现原理概述

Zookeeper是树形结构的注册中心，每个节点的类型分为持久点、持久顺序节点、临时节点和临时顺序节点。Zookeeper节点的基本了解，可以查看我的这篇文章：http://www.relaxheart.cn/to/master/blog?uuid=140 。

Dubbo使用ZK作为注册中心时，只会创建临时节点和持久节点两种，对创建顺序并没有要求。

/dubbo/com.foo.BarService/providers是服务提供者在Zookeeper注册中心的路径示例，是一种属性结构，该结构分为四层：root（根节点，对应示例总的dubbo）、service（接口名称，对应示例中的com.foo.BarService）、四种服务目录（对应示例中的prividers，其他目录还有consumers、routers、configurators）。在服务分类节点下是具体的Dubbo服务URL。属性结构示例如下：

image.png

树形结构的关系：
（1）树的根节点是注册中心分组，下面有多个服务接口，分组值来自用户配置<dubbo:registry>中的group属性，默认是/dubbo.
（2）服务接口下包含四类子目录，分别是providers、consumers、routers、configurators，这个路径时持久节点。
（3）服务提供者目录（/dubbo/service/providers）下面包含的接口有多个服务提供者URL元数据信息。
（4）服务消费者目录（/dubbo/service/consumers）下面包含的解耦有多个消费者URL元数据信息。
（5）路由配置目录（/dubbo/service/routers）下面包含多个用于消费者路由策略URL元数据信息。
（6）动态配置目录（/dubbo/service/configurators）下面包含多个用于服务者动态配置URL元数据信息。

树形示意图，如下：

image.png

配置实现：

<beans>
  <!-- 适用于Zookeeper一个集群有多个节点，多个IP和端口逗号分割-->
  <dubbo:registry protocol="zookeeper" address="ip:port,ip:port,ip:port" />
  <!-- 适用于Zookeeper多个集群有多个节点，多个IP和端口用竖线分割-->
  <dubbo:registry protocol="zookeeper" address="ip:port|ip:port|ip:port" />
</beans>

Redis注册中心实现原理概述

Redis注册中心也沿用了Dubbo抽象的Root、Service、Type、URL四层结构，但是由于Redis属于NoSQL数据库，数据都是以键值对形式保存的，并不能像ZK一样直接实现树形目录结构。因此Redis使用了key/Map结构实现了这个需求，Root、Service、Type组合成Redis的key，Redis的value是一个Map结构，URL作为Map的key，超时时间作为Map的value，如下示意图：

image.png

数据结构组装逻辑在 org.apache.dubbo.registry.redis.RedisRegistry # doRegister(URL url)方法中，总要的几行代码以用注释标记如下：

@Override
    public void doRegister(URL url) {
         //  生成 Redis key
        String key = toCategoryPath(url);
         // 生成 URL
        String value = url.toFullString();
        // 计算过期时间
        String expire = String.valueOf(System.currentTimeMillis() + expirePeriod);
        boolean success = false;
        RpcException exception = null;
        for (Map.Entry<String, JedisPool> entry : jedisPools.entrySet()) {
            JedisPool jedisPool = entry.getValue();
            try {
                try (Jedis jedis = jedisPool.getResource()) {
                    // 注册到注册中心， expire是超时时间
                    jedis.hset(key, value, expire);
                    jedis.publish(key, REGISTER);
                    success = true;
                    if (!replicate) {
                        break; //  If the server side has synchronized data, just write a single machine
                    }
                }
            } catch (Throwable t) {
                exception = new RpcException("Failed to register service to redis registry. registry: " + entry.getKey() + ", service: " + url + ", cause: " + t.getMessage(), t);
            }
        }
        if (exception != null) {
            if (success) {
                logger.warn(exception.getMessage(), exception);
            } else {
                throw exception;
            }
        }
    }

发布/订阅

发布/订阅是整个注册中心的核心功能之一。在传统的应用系统中，我们通常会把配置信息写入一个配置文件，当配置需要变更时修改配置文件，在通过手动触发内存中的配置重新加载，比如重启服务等。在集群模式较小的情况下，这种方式到也可以方便运维，但是当服务节点数量不断上升时候，这种管理方式的的维护成本越来越高，同时手动触发内存更新（重载内置）有一定的风险，对服务的可用性有可能造成短时间的破坏。

但是如果使用了注册中心，上面的问题就可以很好的得到解决。通过监听-通知的机制实现节点变化是即使的通过到相应的服务省去了服务列表维护成本，整个过程是自动完成的。

Zookeeper的实现

1.发布的实现

provider和Consumer需要将自己注册到Zookeeper。服务提供者的注册是为了让消费者订阅（准确来说应该叫感知服务的存在），从而发起远程调用；也上服务治理中心感知有新的服务提供者上线。消费者的发布是为了让服务治理中心可以发现自己。Zookeeper发布订阅代码非常简单，只是调用Zookeeper 的 Client 库在注册中心创建一个目录而已，如下代码所示：

@Override
    public void doRegister(URL url) {
        try {

             // 调用zkClient的create方法创建一个目录即可
            zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true));
        } catch (Throwable e) {
            throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

取消发布对应也很简单，只是把ZK注册中心上对应的路径删除，如下代码所示：

@Override
    public void doUnregister(URL url) {
        try {

            // 删除ZK相应的目录
            zkClient.delete(toUrlPath(url));
        } catch (Throwable e) {
            throw new RpcException("Failed to unregister " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

2.订阅的实现

订阅通常有pull和push两种方式，一种是客户端定时轮训注册中心拉取配置，另一种是注册中心主动推送数据给客户端。这两种方式各有利弊，目前Dubbo采用的是第一次启动拉取方式，后续接收事件重新拉取数据。

在服务暴露时，服务端会订阅configurators用于监听动态配置，在消费端启动时，消费端会订阅providers、routers和configurators这三个目录，分别对应服务提供者、路由和动态配置变更通知。

** Dubbo 中有哪些Zookeeper客户端实现？ **
提供了两种不同ZK开源客户端库的封装，分别对应接口：

• Apache Curator
• zkClient

我们可以在<dubbo:registry> 的client属性中设置curator、zkClient来使用不同的客户端实现库，如果不设置默认使用Curator作为实现。

Zookeeper客户端采用的是“事件通知” + “客户端拉取”的方式，客户端在第一次连接上注册中心时，会获取对应目录西安全量的数据。并在订阅的节点上注册一个watcher，客户端与注册中心之间保持TCP长连接，后续每个节点有任何数据变化的时候，注册中心会根据watcher的回调主动通知客户端（事件通知），客户端接到通知后，会把对应节点下的全量数据都拉取过来（客户端拉取），这一点在NotifyListener#notify(List<URL> urls)接口上就有说明。全量拉取有一个局限，党委服务节点较多时会对网络造成很大的压力。

Zookeeper每个节点都有一个版本号，当某个节点的数据发生变化时，对应的版本号就会变化，并 触发watcher事件，推送数据给订阅方。版本号强调的是变化次数，即使该节点的值没有变化，只要有更新操作，依然会是版本号变化。

Zookeeper实现服务订阅的核心代码在ZookeeperRegistry中，包含COnsumer与Dubbo服务治理中心订阅的逻辑：

@Override
    public void doSubscribe(final URL url, final NotifyListener listener) {
        try {
          
            if (ANY_VALUE.equals(url.getServiceInterface())) {
          
                /***              服务治理中心订阅全部服务               ***/

                // 订阅所有数据
                String root = toRootPath();
                // 获取Listeners
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                if (listeners == null) {
                    // 为获取到监听器，这里创建一个监听器并放入缓存。
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
                    listeners = zkListeners.get(url);
                }
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) {
                    // zkListener为空，说明是第一次，新建一个listener
                    listeners.putIfAbsent(listener, (parentPath, currentChilds) -> {
                        // 这是一个内部类实现，不会立即执行，只会在触发变更通知时执行
                        // 如果子节点有变化则会接收到通知，遍历所有子节点
                        for (String child : currentChilds) {
                            child = URL.decode(child);
                            // 如果存在子节点还未被订阅，说明是新增节点吗，则进行订阅
                            if (!anyServices.contains(child)) {
                                anyServices.add(child);
                                 // 订阅新节点
                                subscribe(url.setPath(child).addParameters(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 (CollectionUtils.isNotEmpty(services)) {
                   // 遍历所有子节点进行订阅
                    for (String service : services) {
                        service = URL.decode(service);
                        anyServices.add(service);
                        subscribe(url.setPath(service).addParameters(INTERFACE_KEY, service,
                                Constants.CHECK_KEY, String.valueOf(false)), listener);
                    }
                }
            } else {

                /***             普通消费者服务订阅               ***/
             

                 List<URL> urls = new ArrayList<>();
                //toCategoriesPath(url): 根据url类别，获取一组要订阅的路径
                for (String path : toCategoriesPath(url)) {
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    if (listeners == null) {
                        // 如果listeners缓存为空则创建缓存
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
                        listeners = zkListeners.get(url);
                    }
                    ChildListener zkListener = listeners.get(listener);
                    // 如果zkListener缓存为空则创建缓存
                    if (zkListener == null) {
                        listeners.putIfAbsent(listener, (parentPath, 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));
                    }
                }
                // 回调NotifyListener, 更新本地缓存信息
                notify(url, listener, urls);
            }
        } catch (Throwable e) {
            throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

Redis的发布订阅

redis实现注册中心的数据结构与Zookeeper不同，其在发布订阅的方式上也是跟ZK不同的，关于Redis如何实现的我也只是粗略的看了下，所以这里就不展开了。

缓存机制

缓存的存在就是用空间换取时间的一种机制。想想，如果Consumer每次远程调用都要先去注册中心拉取一次可调用的服务列表，则会让注册中心承受巨大的流量压力。另外，每个额外的网络请求也会让整个系统的性能下降，同时服务列表变化的频率本身并不是很高，除非服务提供商对接口做了升级、或是服务节点新增或下线（从某各角度来看这并不是一个很高频的操作），所以每次都拉取也就显得并不那么必要。

缓存的实现

因此针对这个问题，dubbo的注册中心实现了通用的缓存机制，在抽象类AbstractRegistry中实现。AbstractRegistry类结构关系图如下所示：

image.png

消费者或者服务治理中心获取注册信息后会做本地缓存。内存中会有一份，保存在Properties对象里，磁盘里也会有一份文件，通过file对象引用。在AbstractRegistry抽象类中有如下定义：

/**  本地缓存对象 **/
private final Properties properties = new Properties();
/** 磁盘文件服务缓存对象 **/
private File file;
/** 内存中的服务缓存对象 **/
private final ConcurrentMap<URL, Map<String, List<URL>>> notified = new ConcurrentHashMap<URL, Map<String, List<URL>>>();

其中内存中的缓存notified是ConcurrentHashMap里面又封装了一个Map，外层Map的key是消费者的URL，内层Map的key是分类，包含了providers、consumers、routers、configutators四种，value则对应的服务列表，对于没有服务提供者提供服务的URL，它会以特殊的empty://前缀了开头。

缓存的加载

在服务初始化时候，AbstractRegistry构造器函数里会从本地磁盘文件中把持久化的注册数据到Properties对象里，并加载到内存缓存中，核心代码如下：

 private void loadProperties() {
        if (file != null && file.exists()) {
            InputStream in = null;
            try {
                // 读取磁盘文件
                in = new FileInputStream(file);
                // 把数据写入到内存缓存中
                properties.load(in);
                ……
            } catch (Throwable e) {
                ……
            } finally {
                ……
            }
        }
    }

Properties保存了所有服务提供者的URL，使用URL#serviceKey（）作为key，提供者列表、路由规则列表、配置规则列表等作为value。如果应用在启动过程中注册中心无法连接或岩机，则Dubbo框架会自动通过本地缓存加载Invokers。

缓存的保存与更新

缓存的保存有同步和异步两种方式。异步会使用线程池异步保存，如果线程在执行过程中出现异常，则会再次调用线程池不断重试，代码如下：

if(syncSaveFile){
    // 同步保存
    doSaveProperties(version);
} else {
    // 异步保存,放入线程池。会传入一个AtomicLong的版本号保证数据是最新的
    registryCacheExecutor.execute(new SaveProperties(version));
}

AbstractRegistry#notify 方法中封装了更新内存和本地文件缓存的逻辑。当客户端第一次订阅获取全量数据的时候，或者后续由于订阅的数据发生变更时，都会调用该方法进行保存。

重试机制

上面缓存实现部分给了一张类关系图，可以看出FailbackRegistry继承了AbstarctRegistry，并在此基础上增加了失败重试机制作为抽象能力。ZookeeperRegistry与RegisRegistry继承该抽象法方法后，直接使用即可。

FailbackRegistry中定义了一个ScheduledExecutorService，每经过固定间隔（默认为5秒）调用FailbackRegistry#retry（）方法。关于ScheduledExecutorService的用法可以查看我的这篇文章：初识ScheduledExecutorService。

Dubbo注册中心使用了那些设计模式

模板模式与工厂模式，Dubbo注册中心拥有良好的扩展性，我们可以在其基础上快速开发出符合我们自己业务需求的注册中心。这种扩展性和Dubbo中使用的设计模式莫不可分，学习这里提到的两个主要的设计模式有助于我们对注册中心源码的阅读。

模板模式

image.png

AbstractRegistry实现了Registry接口中的注册、订阅、查询、通知等方法，还实现了磁盘文件持久化注册信息这一通用方法。但是注册、订阅、查询、通知等方法只是简单地把URL加入对应的集合，没有具体的注册或订阅逻辑。

FailbackRegistry 又集成了AbstractRegistry，重写了父类的注册、订阅、查询、通知等方法，并且添加了重试机制。此外，还添加了四个尉氏县的抽象模板方法，如下：

protected abstract void doRegister(URL url);
protected abstract void doUnregister(URL url);
protected abstract void doSubscribe(URL url, NotifyListener listener);
protected abstract void doUnsubscribe(URL url, NotifyListener listener);

工厂模式

所有的注册中心实现都是通过工厂创建的。类图：

image.png

AbstractRegistryFactory 实现了 RegistryFactory 接口的 getRegistry（URL url）方法，是一个通用实现，主要完成了加锁，以及调用抽象模板方法 createRegistry（URL url）创建具体实现等操作，并缓存在内存中。抽象模板方法会有具体子类继承实现。

虽然每种注册中心都有自己的工厂实现，但是在什么地方判断，应该调用哪个工厂实现呢？？ 代码中并没有显式的判断。答案就在RegistryFactory接口中，该接口里有一个Registry getRegistry(URL url)方法，该方法上有@Adaptive({"protocol"})注解，如下：

@SPI("dubbo")
public interface RegistryFactory {
 
         @Adaptive({"protocol"})
          Registry getRegistry(URL url);
}

了解AOP的话，应该知道这个注解会自动织入一些逻辑，它的value参数会从URL中获取protocol键的值，并根据获取的值来调用不同的工厂类。

总结

本文从注册中心的工作原理；注册中心的数据结构；订阅发布的实现；缓存机制；重试机制；服务注册与发现流程中Dubbo使用了哪些设计模式几个方面对Dubbo注册中心组件的核心内容做了说明。

更多个人博客，欢迎访问我的个人博客网：Tec博客