一、概念&模型

什么是Zookeeper？

分布式应用程序的分布式协调服务，以实现用于同步、配置维护以及组和命名的更高级别的服务

Zookeeper的数据模型？

ZooKeeper 提供的名称空间与标准文件系统的名称空间非常相似。名称是由斜杠 (/) 分隔的一系列路径元素。ZooKeeper 命名空间中的每个节点都由路径标识。

如上图，树中的节点称为znood，它既可以通过路径访问，又可以存储数据，znood有四个属性

data：znood存储的业务数据，注意父节点不能存储数据
children：存储当前节点的字节点的引用信息，因为内存限制，所以 znode 的子节点数不是无限的
stat: 包含 znode 节点的状态信息，比如: 事务 id、版本号、时间戳等，其中事务 id 和 ZK 的数据一直性、选主相关
acl: 记录客户端对 znode 节点的访问权限

znode 的数据操作具有原子性，读操作将获取与节点相关的所有数据，写操作也将替换掉节点的所有数据。znode 可存储的最大数据量是 1MB ，但实际上我们在 znode 的数据量应该尽可能小，因为数据过大会导致 zk 的性能明显下降。每个 ZNode 都对应一个唯一的路径。
znood的类型：节点根据生命周期的不同可以将划分为持久节点和临时节点。持久节点的存活时间不依赖于客户端会话，只有客户端在显式执行删除节点操作时，节点才消失；临时节点的存活时间依赖于客户端会话，当会话结束，临时节点将会被自动删除（当然也可以手动删除临时节点）。注意：临时节点不能拥有子节点。通过create /zk_test1 data1就创建了一个数据为data1的持久节点zk_test1，通过create -e /zk_test2 data2就创建了一个数据为data2的临时节点zk_test2。同时，create命令还可以加-s参数指定节点是否有序，创建顺序节点时，zk 会在路径后面自动追加一个 递增的序列号 ，这个序列号可以保证在同一个父节点下是唯一的，利用该特性我们可以实现分布式锁 等功能。

PERSISTENT：永久节点
EPHEMERAL：临时节点
PERSISTENT_SEQUENTIAL：永久顺序节点
EPHEMERAL_SEQUENTIAL：临时顺序节点

Zookeeper的特性？

顺序一致性——来自客户端的更新将按照它们发送的顺序应用。
原子性——更新要么成功要么失败。没有部分结果。
单一系统映像——无论连接到哪个服务器，客户端都将看到相同的服务视图。即，即使客户端故障转移到具有相同会话的不同服务器，客户端也永远不会看到系统的旧视图。
可靠性——应用更新后，它将一直持续到客户端覆盖更新为止。
及时性——系统的客户视图保证在特定时间范围内是最新的。

Watcher监听机制？
Watcher 监听机制由ZK 服务端、ZK 客户端、客户端的 WatchManager 对象实现，客户端首先将Watcher注册到服务端，同时将Watcher对象保存到客户端的WatchManager中。当ZK服务端监听的数据状态发生变化时，服务端会主动通知客户端，接着Watchmanager会触发相关Watcher来回调相应处理逻辑。

一、Zookeeper单机版安装部署

安装包下载地址https://zookeeper.apache.org/releases.html，这里选择最新的稳定版本3.7.1，需要注意，提供了两种安装包：apache-zookeeper-XYZtar.gz 是标准的纯源代码版本，apache-zookeeper-XYZ-bin.tar.gz 是包含二进制文件的便利压缩包，这里我下载的是apache-zookeeper-XYZ-bin.tar.gz。

1、安装

解压安装包tar -zxvf apache-zookeeper-3.7.1-bin.tar.gz，得到目录结构如下

进入conf目录，将zoo_simple.cfg文件重命名为zoo.cfg，mv zoo_sample.cfg zoo.cfg，高版本的zookeeper默认加载conf/zoo.cfg，修改配置文件（我这里主要修改了存储路径）

#ZooKeeper 使用的基本时间单位，以毫秒为单位。它用于执行心跳，最小会话超时将是 tickTime 的两倍
tickTime=2000
#initLimit和syncLimit是针对集群的参数，分别为通信时限和同步时限
initLimit=10
syncLimit=5
#存储内存数据库快照的位置，除非另有说明，否则存储更新数据库的事务日志
dataDir=/usr/local/zookeeper
#客户端连接的端口
clientPort=2181

至此，zookeeper已经安装配置完成，下面启动连接测试一下。

2、运行&测试

执行bin/zkServer.sh start命令启动zookeeper，启动成功如下

执行bin/zkCli.sh -server 127.0.0.1:2181连接到zookeeper服务端，如下

连接成功后，可以测试一些简单的cli命令，这里先附上官方文档https://zookeeper.apache.org/doc/r3.7.1/zookeeperCLI.html，然后我们这里简单测试几个

版本信息

创建一个新的znode，并将字符串“my_data”与该节点关联

查看刚才关联的数据，添加数据，再次查看发现数据被覆盖

三、Zookeeper集群搭建

1、集群搭建

这里采用的是伪集群方式搭建，和真实集群只是ip不同，搭建方式是一样的。将上面单机版的zk配置文件重命名为zoo-1.cfg，mv zoo.cfg zoo-1.cfg，修改配置文件，如下所示

#ZooKeeper 使用的基本时间单位，以毫秒为单位。它用于执行心跳，最小会话超时将是 tickTime 的两倍
tickTime=2000
#initLimit和syncLimit是针对集群的参数，分别为通信时限和同步时限
initLimit=10
syncLimit=5
#存储内存数据库快照的位置，除非另有说明，否则存储更新数据库的事务日志
dataDir=/usr/local/zookeeper1
#客户端连接的端口
clientPort=2181
#server.A=B:C:D为，A第几号服务器，B这个服务器的IP，C这个服务器与集群中的Leader服务器交换信息的端口，D如果Leader挂了，需要一个新的端口进行选举，选出的新Leader的端口，需要注意这里是伪集群IP一样，因此分配不同的实例端口，若真正集群，由于B不同，三台机器的C和D可以相同
#server.<节点ID>=<IP>:<数据同步端口>:<选举端口>
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890

接下来我们复制zoo-1.cfg，复制出2和3，修改clientPort和dataDir即可。然后，在前面创建的三个存储目录编辑myid文件，分别输入1，2，3，对应server.A这个A，接下来通过使用不同配置文件启动

bin/zkServer.sh start conf/zoo-1.cfg
bin/zkServer.sh start conf/zoo-2.cfg
bin/zkServer.sh start conf/zoo-3.cfg

bin/zkServer.sh stop conf/zoo-1.cfg
bin/zkServer.sh stop conf/zoo-2.cfg
bin/zkServer.sh stop conf/zoo-3.cfg

启动之后，可以通过bin/zkServer.sh status conf/zoo-1.cfg命令查看状态



至此，集群搭建完成。

2、Zookeeper的选举机制

第一次启动

假设目前有1、2、3三台机器，依次启动，1号启动给自己投了一票，这时没有超过半数以上，因此，没有master，1号机器处于looking状态，2号机器启动，也给自己一票，此时1号机器发现2号机器的myid大于自己，因此将自己的票投给2号，此时超过一半以上，因此2号机器为master，1号为follower，3号机器启动后，此时1号机器已经没了选票，2号有两张，3号自己一张，少数服从多数，3号也成为follower

非第一次启动选举

leader存活，只是某个follower挂掉，当它重新启动之后，集群中还是存在leader的，只需要重新建立连接

leader不存活，每个节点在运行时都有如下的三个id（epoch，zxid，sid），epoch是leader任期时id（选举一次为一期），zxid是事务id，sid就是其myid，当某时刻leader宕机后，首先看存活节点的epoch id，最大的选举为leader，如果epoch都相同，则看其事务id  zxid ，选举zxid最大的为leader，如果zxid都相同，则看其sid，就是节点的myid，myid大的为leader

四、SpringBoot集成

1、引入依赖

在Zookeeper官网提供了Java Api的示例，但是原生的Api需要开发人员创建对象、注册watcher等，因此这里使用一个封装好的curator，它提供了各种应用场景，实现了Fluent风格的Api接口，是操作zk的最好的框架。引入依赖，注意对应你的zk版本。

<!-- zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.1</version>
        </dependency>

2、配置类&配置文件

server.port=8002
server.servlet.context-path=/zk
#zk地址
curator.connectString=192.168.1.1:2181,192.168.1.1:2182,192.168.1.1:2183
#重试次数
curator.retryCount=1
#重试时间间隔
curator.elapsedTimeMs=2000
#session超时时间
curator.sessionTimeoutMs=6000
#连接超时时间
curator.connectionTimeoutMs=10000

package com.example.zookeeper.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author zy
 * @version 1.0.0
 * @ClassName zkConfig.java
 * @Description TODO
 * @createTime 2022/12/5
 */
@Configuration
public class ZKConfig {
    @Value("${curator.connectString}")
    private String connectString;
    @Value("${curator.retryCount}")
    private String retryCount;
    @Value("${curator.elapsedTimeMs}")
    private String elapsedTimeMs;
    @Value("${curator.sessionTimeoutMs}")
    private String sessionTimeoutMs;
    @Value("${curator.connectionTimeoutMs}")
    private String connectionTimeoutMs;

    @Bean("CuratorFramework")
    public CuratorFramework curatorFramework() throws InterruptedException {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(
                connectString,
                Integer.valueOf(sessionTimeoutMs),
                Integer.valueOf(connectionTimeoutMs),
                new RetryNTimes(Integer.valueOf(retryCount),Integer.valueOf(elapsedTimeMs))
        );
        curatorFramework.start();
        //这里是阻塞直到连接成功
        curatorFramework.blockUntilConnected();
        return  curatorFramework;
    }
}

需要注意，如果curatorFramework没有启动连接成功，直接使用会报错java.lang.IllegalStateException: Expected state [STARTED] was [LATENT]

3、简单测试

package com.example.zookeeper.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;

/**
 * @author zy
 * @version 1.0.0
 * @ClassName TestController.java
 * @Description TODO
 * @createTime 2022/12/5
 */
@RestController
public class TestController {
    @Autowired
    private CuratorFramework client;

    @RequestMapping("/test")
    public void test(){
        System.out.println(client.toString());
    }

    @RequestMapping("/query")
    public String query(@RequestParam String url) throws Exception{
        byte[] data = client.getData().forPath(url);
        return new String(data);
    }

    @RequestMapping("/put")
    public String put(@RequestParam String url,String value) throws Exception{
        String path = client.create().creatingParentsIfNeeded().forPath(url, value.getBytes(StandardCharsets.UTF_8));
        return path;
    }
}

连接控制台，添加一个测试数据，然后查询如下

创建一个新的文件路径，并赋值，通过控制台查询


可以看到，集群中的三个节点都能查询到新增的数据。

项目启动过程中控制台有一个报错，这个不影响项目使用，将curator升级到5.2.0就可以解决了，是curator的BUG。

总结

zookeeper作为分布式存储系统，为许多中间件提供提供协调服务，例如我们熟悉的Dubbo、Codis、ClickHouse等，都直接或间接的注册、配置到zookeeper，在本文中，针对zk的一些基本概念、数据结构、搭建方式和SpringBoot集成做了一些叙述，基于本文的内容与官方文档、其他博客相结合，可以基本掌握zk的使用，如需精通，还要再进一步的深入学习。