ZooKeeper

1.ZooKeeper 简介

Zookeeper(动物园管理者)简称ZK，一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby的一个开源的实现，是Hadoop和Hbase的重要组件。Zookeeper使用java编写，但是支持java和C两种编程语言。

1.1 ZooKeeper的应用场景

• dubbo框架 SpringCloud框架，使用Zk作为注册中心。
• Hadoop HBASE组件，集群架构，zk是集群管理者
• zk实现分布式锁

2 ZK内存数据模型

2.1 模型结构

2.2 模型的特点

• 每个子目录如[/node1]都被称作一个znode(节点)。这个znode是被它所在的路径唯一标识。
• znode可以有子节点目录，并且每个znode可以储存数据
• znode是有版本的，每一个znode中存储的数据可以有多个版本，也就是一个访问路径中可以存储多份数据。
• znode可以被监控，包括这个目录节点存储的数据的修改，子节点目录的变化等，一旦变化，可以通知设置监控的客户端。

3 节点分类

3.1 持久节点(PERSISTENT)
是指节点在创建后，就一直存在，直到有删除操作来主动删除这个节点–不会因为创建该节点的客户端会话失效而消失

---

3.2 持久顺序节点(PERSISTENT_SEQUENTIAL)
这类节点的基本特性和上面的节点类型是一致的。额外的特性是在ZK中每个父节点会为它的第一级子节点维护一份时序，会记录每个子节点创建的先后顺序。基于这个特性，在创建子节点的时候，可以设置这个属性，那么在创建节点过程中，ZK会自动为节点加上一个数字后缀，作为新的节点名。这个数字后缀的范围是整形的最大值。

---

3.3 临时节点(EPHEMERAL)
和持久节点不同的是，临时节点的生命周期和客户端会话绑定，也就是说，如果客户端会话失效，这个节点就会 自动被清除掉。注意，这里提到的是会话失效，而非连接断开。另外，在临时节点下面不能创建子节点。

---

3.4 临时顺序节点(EPHEMERAL_SEQUENTIAL)
具有临时节点特点，每个父节点会为它的第一级子节点维护一份时序。这点和刚才提到的持久顺序节点类似。

---

4 安装

4.1 linux系统安装

# 1 安装jdk并配置环境变量 & 下载zk安装包 【在改地址中查找对应的版本下载】
- http://archive.apache.org/dist/zookeeper/

# 2 下载安装包到linux服务器中 并解压缩
tar -zvxf zookeeper.tar.gz

# 3 重命名安装目录
- mv zookeeper-x.xx  zookeeper

# 4 修改zoo.cfg配置文件
- 1修改zk的conf目录下的zoo_simple.cfg，修改后重命名为zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/zookeeper/zkdata
clientPort=2181

# 5 启动zk
- 在zk的bin目录下，运行zkServer.sh
./bin/zkServer.sh start /usr/zookeeper/conf/zoo.cfg

# 6 使用jdk jps命令查看启动是否成功
- ./bin/zkCli.sh -server 192.168.0.109:2181
注意:可以通过 ./bin/zkCli.sh help 查看客户端所有可以执行的指令

5 ZK配置文件

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1

---

5.1 tickTime
集群节点之间的心跳时间，默认为两秒

---

5.2 initLimit
初始化集群式集群节点同步超时时间，为多少个心跳周期，默认为10次心跳，乘以默认心跳时间2秒，超时时间为20秒

---

5.3 syncLimit
集群在集群在运行过程中同步数据超时时间，默认为5个心跳骤起，计算方式同上

---

5.4 dataDir
ZK节点数据的存放位置，默认是/temp/zookeeper，由于temp为临时存储空间，zk建议用户自行创建目录，默认配置文件的地址仅仅作为示例

---

5.5 maxClientCnxns
最大客户端连接数，也就是zk默认的线程池初始化值

---

5.6 autopurge.snapRetainCount
zk中的数据快照机制，在某一个时刻触发，生成当前的zk节点数据快照。生成快照的数量不断累积，避免大量的数据快照占用磁盘空间，zk默认的快照合并是3个，当zk已经生成了3个数据快照，或者找过了3个，在到合并快照的时间间隔后，执行合并，如果快找时间没有到达到设定的值，则不会合并快照。

---

5.7 autopurge.purgeInterval
快照合并的时间间隔，单位小时，zk默认合并快照的时间为1小时，如果这个值为0，则表示不需要合并快照

6 客户端基本指令

# 1 ls [path]             查看特定节点下的子节点

# 2 create [path] [data]  创建一个节点，并给节点绑定数据(默认是持久性节点)
- create path data        创建持久节点(默认是持久节点)
- create -s path data     创建持久顺序节点
- create -e path data     创建临时性节点(注意：临时节点不能含有任何子节点)
- create -e -s path data  创建临时顺序节点(注意：临时节点不能含有任何子节点)

# 3 stat [path]           查看节点状态

# 4 set [path] [data]     修改节点数据

# 5 ls2 [path]            查看节点下子节点和当前节点的状态

# 6 history               查看操作历史

# 7 get [path]            获得节点上绑定的数据星系

# 8 delete [path]         删除节点(注意：删除节点不能含有子节点)

# 9 rmr [path]            递归删除节点(注意：会将当前节点下所有节点删除)

# 10 quit                 退出当前会话(会话失效)

7 节点监听机制

客户端可以今天znode节点的变化，Znode节点的变化触发相应的时间，然后清除对该节点的检测。但检测一个znode节点时候。
Zookeeper会发送通知给监听节点，一个Watch时间是一个一次性的触发器，但被设置了Watch的数据获取目录发生了变了改变的时候，则服务器将这个改变发送给设置了Watch客户端以便通知它们。

# 1 ls /path true       监听节点目录的变化
# 2 get /path tre       监听节点数据的变化

8 java操作ZK

8.1 创建项目引入依赖

<!-- zkClient依赖 -->
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>

<!-- juit依赖 -->
 <dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
    <scope>provided</scope>
 </dependency>

---

8.2 java代码示例

package cn.qzook;

import cn.qzook.domain.User;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

/**
 * @Description
 * @Author zc
 * @Date 2021/4/14 下午7:01
 */
public class TestZKClient {

    private ZkClient zkClient;

    @Before
    public void before(){
        //参数 1 ：zkServer 服务器IP地址
        //参数 2 ：会话超时时间
        //参数 3 ：连接超时时间
        //参数 4 ：序列化的方式
        zkClient = new ZkClient("192.168.0.102:2181", 60000*30, 60000,new SerializableSerializer());
    }

    @After
    public void after(){
        //释放资源
        zkClient.close();
    }


    //在zk中创建节点
    @Test
    public void testCreateNode(){
        // 1 持久节点
        zkClient.create("/node1","xiaocheng", CreateMode.PERSISTENT);

        // 2 持久顺序节点
        zkClient.create("/node1/names","xiaoli",CreateMode.PERSISTENT_SEQUENTIAL);

        // 3 临时节点
        zkClient.create("/node1/lists","xiaozhang",CreateMode.EPHEMERAL);

        // 4 临时顺序节点
        zkClient.create("/node1/lists11","xiaoxiao",CreateMode.EPHEMERAL_SEQUENTIAL);
    }

    //删除节点
    @Test
    public void delete(){
        //删除没有子节点的节点 返回值：是否删除成功
//        boolean delete = zkClient.delete("/node1");

        //递归删除节点信息，返回值：是否删除成功
        boolean recursive = zkClient.deleteRecursive("/node1");
    }

    //查看节点的字节点
    @Test
    public void testFindNOdes(){
        //获取指定路径的节点信息 返回值：当前节点的子节点信息[名称]
        List<String> children = zkClient.getChildren("/");
        for (String child : children) {
            System.out.println(child);
        }
    }

    //获取节点数据
    //注意:通过java客户端操作，必须保证数据的序列化方式一致，否则会报错[数据格式转换异常]
    @Test
    public void testFindNodeData(){
        Object o = zkClient.readData("/node1");
        System.out.println(o);
    }


    //查看当前节点的数据，并虎丘状态信息
    @Test
    public void testFindNodeDataAndStat(){
        Stat stat = new Stat();

        Object o = zkClient.readData("/node1", stat);

        System.out.println(o);
        System.out.println(stat);
    }

    //修改节点的数据
    @Test
    public void testWriteData(){
        zkClient.writeData("/node1",new User("lining",99));

        User o = zkClient.readData("/node1");

        System.out.println(o);

    }

    //监听节点数据的变化
    //注意：通过java客户端监听的节点，数据变化只能是java客户端来操作的，使用zk终端是不会触发。但是删除操作是会触发java监听事件
    @Test
    public void testWatchDataChange() throws IOException {
        zkClient.subscribeDataChanges("/node2", new IZkDataListener() {
            @Override //当前节点的数据变化时，触发当前方法
            public void handleDataChange(String dataPath, Object data) throws Exception {
                System.out.println(dataPath);
                System.out.println(data);
            }

            @Override //当前节点被删除时触发这个方法
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println(dataPath);
            }
        });

        //阻塞当前线程，防止线程终止，无法监听到节点变化
        System.in.read();
    }

    //配合上面的方法，改变节点数据
    @Test
    public void dataChange(){
        zkClient.writeData("/node2","123");
    }


    //监听节点目录的变化
    @Test
    public void testOnNodesChange() throws IOException {
        zkClient.subscribeChildChanges("/node3", new IZkChildListener() {
            @Override //监听节点的子节点发生变化，触发该方法
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                System.out.println("当前节点："+parentPath);
                System.out.println("发生变更后，当前节点的子节点列表：");
                for (String currentChild : currentChilds) {
                    System.out.print(currentChild+"\t");
                }
                System.out.println("\n=================");
            }
        });
        //阻塞当前线程，防止线程终止，无法监听到节点变化
        System.in.read();
    }
    //配合节点目录变化
    @Test
    public void changeDir(){
        zkClient.create("/node3/test1","test message",CreateMode.PERSISTENT);
        zkClient.create("/node3/test2","test message",CreateMode.PERSISTENT);
        zkClient.create("/node3/test3","test message",CreateMode.PERSISTENT);
    }
}

9 ZK集群

9.1 集群(cluster)

# 1 集群(cluster)
- 集合同一各种软件服务的多个节点同时提供服务

# 2 集群解决问题
- 单节点的并发放稳的压力
- 单点故障问题(如硬件老化，自然灾害等)

9.2 搭建集群

注意：zk集群搭建官方建议奇数个服务器，以便于zk的内部选举主节点和从节点。

# 1 创建三个dataDir目录，用于zk的持久数据存放
mkdir zkdata1 zkdata2 zkdata3

# 2 分别在三个dataDir目录下创建myid文件，文件内容为zk服务器的ID，数字类型，保证唯一
touch ./zkdata1/myid
touch ./zkdata2/myid
touch ./zkdata3/myid

# 3 在、conf目录下创建三个zk配置文件，分别为zoo1.cfg,zoo2.cfg,zoo3.cfg
- zoo1.cfg 
    tickTime=2000
    initLimit=10
    syncLimit=5
    dataDir=/root/zkdata1
    clientPort=3001
    server.1=192.16.0.102:3002:3003
    server.2=192.16.0.102:4002:4003
    server.3=192.16.0.102:5002:5003
    
- zoo2.cfg 
    tickTime=2000
    initLimit=10
    syncLimit=5
    dataDir=/root/zkdata1
    clientPort=4001
    server.1=192.16.0.102:3002:3003
    server.2=192.16.0.102:4002:4003
    server.3=192.16.0.102:5002:5003
 
- zoo3.cfg 
    tickTime=2000
    initLimit=10
    syncLimit=5
    dataDir=/root/zkdata1
    clientPort=5001
    server.1=192.16.0.102:3002:3003
    server.2=192.16.0.102:4002:4003
    server.3=192.16.0.102:5002:5003
    
# 配置文件
  1 server.x ： x 为服务器的唯一表示，对应dataDir目录myid文件中的信息
  2 192.168.0.102：为服务器所在的IP地址
  3 3002 ： 数据同步使用的端口号
  4 3003 ： 选举使用的端口号
  
  
# 4 分别启动各个zk服务器
./bin/zkServer.sh start /root/zoo1.cfg
./bin/zkServer.sh start /root/zoo2.cfg
./bin/zkServer.sh start /root/zoo3.cfg


# 5 查看各个zk服务器角色信息
./bin/zkServer.sh status /root/zoo1.cfg

# 6 客户端连接任意zk服务器进行节点操作
./bin/zkCli.sh -server 192.168.0.102:3001

# 7 停止特定的zk服务器
./bin/zkServer.sh stop /root/zoo1.cfg

10 java操作zk集群

在初始化zk客户端对象ZKClient的时候，将集群中所有的服务器地址都初始化到对象中。

ZkClient zkClient = new ZkClient("192.168.0.102:3001,192.168.0.102:4001,192.168.0.102:5001",
                                60000*30, 60000,
                                new SerializableSerializer());