如何基于Zookeeper实现注册中心模型？

在分布式系统中，通常会存在几十个甚至上百个服务，开发人员可能甚至都无法明确系统中到底有哪些服务正在运行。另一方面，我们很难同时确保所有服务都不出现问题，也很难保证当前的服务部署方式不做调整和优化。由于自动扩容、服务重启等因素，服务实例的运行时状态也会经常变化。通常，我们把这些服务实例的运行时状态信息统称为服务的元数据（Metadata）。

既然服务数量的增加以及服务实例的变化都不可避免，那么，有什么好的办法能够做到对这些服务实例进行有效的管理呢？这实际上就是一个服务治理的问题。我们需要管理系统中所有服务实例的运行时状态，并能够把这些状态的变化同步到各个服务中。就技术组件而言，我们可以通过引入注册中心轻松实现对大规模服务的高效治理。

注册中心模式和工具

在分布式系统中，我们引入注册中心的目的是为了实现服务的自动注册和发现机制。围绕这两个操作，我们可以先来探讨注册中心所应该具备的模型结构。

注册中心模型

注册中心保存着各个服务实例的元数据，涉及的角色包括如下三种。

注册中心

提供服务注册和发现能力。

服务提供者

将自身注册到注册中心，供服务消费者进行调用。

服务消费者

从注册中心获取服务提供者的元数据，并发起远程调用。

上述三个角色比较简单，但注册中心的具体组成结构还是有一些额外的特性。首先，注册中心本身可以认为是一种服务器，它也提供了对应的客户端组件。各个服务需要嵌入客户端组件才能完成与注册中心服务器之间的交互。然后，为了提高访问效率，服务的消费者一般都会构建一个本地缓存，用来保存那些已经访问过的服务实例元数据。下图展示了服务与注册中心的交互过程。

在上图中，基本的工作流程通过操作语义即可理解。但有一个问题需要解决，即一旦服务的运行时状态发生了变更，我们如何有效获取这些变更信息呢？这就需要在注册中心中进一步引入变更通知机制，如下图所示。

从设计理念上讲，我们希望这种来自注册中心的变更通知能够实时的同步到服务消费者，这时候就可以引入推送思想。那么，如何具体实现推送呢？我们可以采用监听机制。所谓监听机制，指的就是服务消费者对位于注册中心的元数据添加监听器，一旦元数据发生变化，就可以触发监听器中的回调函数。我们可以在回调函数中对已变更的元数据执行任何操作，如下所示。

可以看到，服务消费者可以对具体的服务实例节点添加监听器，当这些节点发生变化时，注册中心就能触发监听器中的回调函数确保更新通知到每一个服务消费者。显然，使用监听和通知机制具备实时的数据同步效果。

注册中心实现工具

以上关于注册中心的讨论为我们提供了理论基础。根据这些理论基础，业界也诞生了很多具体的实现工具，常见的包括Consul 、Zookeeper、Eureka和Nacos等。我们无意对这些工具做一一展开。在本文中，我们将基于Zookeeper来具体分析注册中心的实现模型。Zookeeper是基于监听和通知机制的典型框架。

从物理结构上讲，Zookeeper就是一个目录树，包含了一组被称为ZNode的节点，它的基本结构如下图所示。

在上图中，count节点位于/business/product/count路径，节点temp可以存储数据100，而节点/shop/order/1可能存储着类似{"id":"1","itemName":"Notebook","price":"4000",createTime="2022-06-16 22:39:15"}”等复杂数据结构和信息。Zookeeper中所有数据通过ZNode的路径被引用。

Zookeeper特性很多，我们可以从注册中心的基本实现需求出发，结合模型及其操作来把握用于构建注册中心的相关技术。

首先，Zookeeper专门设计并实现了一个监听器组件。我们可以在任何一个ZNode上添加监听器，并实现对应的回调函数，从而确保服务器端的变化能够通过回调机制通知到客户端。

另一方面，Zookeeper中也提供了临时节点的概念。所谓临时节点，指的是只要客户端与Zookeeper的连接发生中断，那么这个节点就会自动消失。显然，临时节点的这种特性可以用于控制该节点所包含的服务定义元数据的时效性。

ZNode是Zookeeper中可以用代码进行控制的主要实体。对ZNode的基本操作包括节点创建create、删除delete、获取子节点getChildren以及获取和设置节点数据的getData/setData方法。操作Zookeeper的客户端组件包括自带的ZooKeeper API和第三方zkClient、Curator等，这些客户端都对Zookeeper连接资源管理和对ZNode节点的各项操作做了不同程度的封装。Zookeeper中涉及的主要操作如下表所示，在源码解读过程中，我们会发现对Zookeeper的控制基本都是对这些操作的封装和应用。

操作	描述
create	在ZooKeeper命名空间的指定路径中创建一个znode
delete	从ZooKeeper命名空间的指定路径中删除一个znode
exists	检查路径中是否存在ZNode
getChildren	获取ZNode的子节点列表
getData	获取与ZNode相关的数据
setData	将数据设置/写入ZNode的数据字段
getACL	获取ZNode的访问控制列表（ACL）策略
setACL	在ZNode中设置访问控制列表（ACL）策略
sync	将客户端的ZNode视图与ZooKeeper同步

基于Zookeeper实现注册中心

介绍完注册中心模型以及Zookeeper框架，让我们回到Dubbo。作为一款主流的分布式服务框架，Dubbo也内置了一整完整的注册中心实现方案，默认采用的就是Zookeeper。

Dubbo注册中心模型

Dubbo中的注册中心代码位于dubbo-registry工程中，其中包含了一个dubbo-registry-api工程，该工程包含了Dubbo注册中心的抽象API，而剩下的dubbo-registry-default、dubbo-registry-zookeeper、dubbo-registry-nacos等工程则是这些API的具体实现，分别对应前面提到的各种注册中心实现方式。我们同样无意对所有这些注册中心实现方式做详细展开，而是重点关注抽象API以及基于Zookeeper的实现方式。

我们首先来看一下dubbo-registry-api工程，这里面最核心的就是在如下所示的RegistryService接口。

public interface RegistryService {

//注册

void register(URL url);

//取消注册

void unregister(URL url);

//订阅

void subscribe(URL url, NotifyListener listener);

//取消订阅

void unsubscribe(URL url, NotifyListener listener);

//根据URL查询对应的注册信息

List<URL> lookup(URL url);

}

请注意，RegistryService所有操作的对象都是URL，而订阅相关的操作中还附加了监听器NotifyListener，确保变更信息的推送。从命名上我们已经可以初步猜想Dubbo在注册信息变更时采用的就是监听和通知机制。通过确认NotifyListener接口的定义更加明确了我们的猜想，因为该接口中只有一个notify方法，用于将发生变更的注册信息以URL的形式进行通知，如下所示。

public interface NotifyListener {

void notify(List<URL> urls);

}

我们再来看RegistryFactory接口，如下所示。这里的@SPI("dubbo")注解我们会在第X讲介绍微内核模式时进行介绍，代表默认情况下使用Dubbo自身的注册中心。

@SPI("dubbo")

public interface RegistryFactory{

Registry getRegistry(URL url);

}

从接口的命名上可以看出RegistryFactory是Dubbo中创建注册中心的工厂类，通过对RegistryFactory的实现，Dubbo提供了Zookeeper、Redis等几种不同的注册中心实现方案。

可以说Dubbo中关于注册中心API层的抽象简单而清晰，比较适合先用来做对全局代码结构的把握。在这层API抽象之下，我们重点介绍ZookeeperRegistry和ZookeeperRegistryFactory。

Zookeeper注册中心实现过程

让我们来到Dubbo源码，来看一下ZookeeperRegistry的实现过程，而ZookeeperRegistry中最重要的就是它的构造函数，如下所示。

public ZookeeperRegistry(URL url, ZookeeperTransporter, zookeeperTransporter) {

...

//建立与Zookeeper的连接

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);

}

});

}

可以看到，这里执行了两个操作，一个是与Zookeeper建立连接，另一个就是添加了用于断线重连的状态监听器。根据对Zookeeper基本操作的了解和掌握，上述实现过程都是使用Zookeeper时的常规步骤。

为了理解这段代码，我们需要明确另外两个核心对象的创建过程，这两个核心对象分别是ZookeeperTransporter和ZookeeperClient。我们发现ZookeeperTransporter是在ZookeeperRegistryFactory工厂类创建ZookeeperRegistry时带进来的，如下所示。

public class ZookeeperRegistryFactory extends AbstractRegistryFactory {

private ZookeeperTransporter zookeeperTransporter;

public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {

this.zookeeperTransporter = zookeeperTransporter;

}

public Registry createRegistry(URL url) {

return new ZookeeperRegistry(url, zookeeperTransporter);

}

ZookeeperTransporter本身是一个接口，定义也比较简单，就是根据传入的URL创建与Zookeeper服务器的连接并获取一个ZookeeperClient对象，如下所示。

@SPI("zkclient")

public interface ZookeeperTransporter {

@Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})

ZookeeperClient connect(URL url);

}

另一方面，在ZookeeperClient接口的定义中包含了注册中心运行过程中所有的数据操作，如创建和删除路径、获取子节点、添加和删除Listener、获取URL等实现发布-订阅模式的入口。这些方法名与Zookeeper原生操作基本一致，如下所示。

public interface ZookeeperClient {

void create(String path, boolean ephemeral);

void delete(String path);

List<String> getChildren(String path);

List<String> addChildListener(String path, ChildListener

listener);

void removeChildListener(String path, ChildListener listener);

void addStateListener(StateListener listener);

void removeStateListener(StateListener listener);

boolean isConnected();

void close();

URL getUrl();

}

目前可以与Zookeeper服务器进行交互的客户端有很多，Dubbo中提供了对Zkclient和Curator这两个客户端工具的集成，对应的Transporter和ZookeeperClient实现类见下图。Dubbo使用Zkclient作为其默认实现。

接下来终于到了分析注册中心具体操作的时候了，ZookeeperRegistry提供了doRegister、doUnregister、doSubscribe和doUnsubscribe方法分别对应注册/取消注册、订阅/取消订阅这四个具体操作。我们首先来看一下注册方法doRegister，如下所示。

protected void doRegister(URL url) {

try {

zkClient.create(toUrlPath(url),

url.getParameter(Constants.DYNAMIC_KEY, true));

} catch (Throwable e) {

...

}

不难看出，注册操作的实现方式就是在Zookeeper中创建一个节点。请注意，默认创建的节点都是临时节点，当连接断开之后会自动删除。对应的，我们也不难想象取消注册的实现方式就是删除这个临时节点，如下所示。

protected void doUnregister(URL url) {

try {

zkClient.delete(toUrlPath(url));

} catch (Throwable e) {

...

}

我们再来看订阅过程。在订阅URL过程中，Dubbo将传入的回调接口NotifyListener转换成Zookeeper中的ChildListener，并主动根据服务提供者URL调用NotifyListener。doSubscribe方法比较长，我们提取其中的核心代码，如下所示。

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);

}

可以看到，Dubbo会订阅父级目录, 而当有子节点发生变化时就会触发ChildListener中的回调函数，该回调函数会对该路径下的所有子节点执行subscribe操作。

而取消订阅URL的过程实际上只是去掉URL上已经注册的监听器，doUnsubscribe方法如下所示。

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);

}

到此为止，ZookeeperRegistry类中的构造函数和核心方法已经分析完毕。大家看到这里可能会好奇，doRegister、doUnregister、doSubscribe和doUnsubscribe这四个方法是在哪里被调用的呢？毕竟ZookeeperRegistry本来应该实现的是RegistryService接口中的register、unregister、subscribe和unsubscribe方法才对。通过阅读代码，我们发现 ZookeeperRegistry并不是RegistryService的直接实现类，从类层结构上，ZookeeperRegistry扩展了FailbackRegistry，而FailbackRegistry又扩展了AbstractRegistry，注意FailbackRegistry和AbstractRegistry都是抽象类。而前面提到的这些方法在RegistryService不同层级的实现类中被调用，这里面涉及到的类层结构如下图所示。

我们继续往下看，发现真正调用doRegister、doUnregister、doSubscribe和doUnsubscribe这四个方法的地方分别是在FailbackRegistry对应的register、unregister、subscribe和unsubscribe方法中，这点自然比较好理解。但我们发现这四个方法还同时出现在FailbackRegistry的retry方法中。事实上，在FailbackRegistry构造函数中会创建一个定时任务，每隔一段时间执行该retry方法。在这个retry方法，以注册场景为例（其他场景也类似），我们从注册失败的集合中获取URL，然后对每个URL执行doRegister操作从而实现重新注册，如下所示。

if (!failedRegistered.isEmpty()) {

Set<URL> failed = new HashSet<URL>(failedRegistered);

if (failed.size() > 0) {

try {

for (URL url : failed) {

try {

//重新注册

doRegister(url);

failedRegistered.remove(url);

} catch (Throwable t) {

…

}