ZooKeeper的基本使用

r囧r小猫 2023-01-10 14:56 217阅读 0赞

# Zookeeper #

### 文章目录 ###

*  Zookeeper
 *   *  一、ZooKeeper概述
     *  二、ZooKeeper安装
     *  三、ZooKeeper数据模型
     *  四、ZooKeeper命令行操作
     *  五、ZooKeeper JavaAPI操作
     *  六、ZooKeeper集群介绍

## 一、ZooKeeper概述 ##

1.  ZooKeeper是一个开源的分布式应用程序的协调服务，是一个树形目录的服务
2.  ZooKeeper提供的主要功能包括配置管理、分布式锁、集群管理

## 二、ZooKeeper安装 ##

前提需要Linux中已经安装了JDK7或更高版本

1.  将ZooKeeper安装包上传到Linux中
2.  执行指令`tar -zxvf apache-zookeeper-3.5.6-bin.tar.gz`解压安装包
3.  进入解压后生成的apache-zookeeper-3.5.6-bin目录，执行`mkdir data`，用于存放ZooKeeper产生 的数据
4.  `cd /opt/ZooKeeper/apache-zookeeper-3.5.6-bin/conf`，执行`mv zoo_sample.cfg zoo.cfg`
5.  `vim zoo.cfg`，修改配置文件 (开放2181端口)

![在这里插入图片描述][20210125124909254.png]

1.  启动ZooKeeper服务，先`cd /opt/ZooKeeper/apache-zookeeper-3.5.6-bin/bin`，后`./zkServer.sh start`

![在这里插入图片描述][20210125124912159.png]

注意1：ZooKeeper中的AdminService服务(一般不会用到)会占用8080端口，与Tomcat冲突， 故需要`vim zoo.cfg`，将此服务占用的端口号修改为没有被占用的端口号

![在这里插入图片描述][2021012512491695.png]

注意2：如果启动失败使用指令`./zkServer.sh start-foreground`可以查看启动日志中的错误信息

1.  停止ZooKeeper服务，先`cd /opt/ZooKeeper/apache-zookeeper-3.5.6-bin/bin`，后`./zkServer.sh stop`
2.  查看ZooKeeper状态，先`cd /opt/ZooKeeper/apache-zookeeper-3.5.6-bin/bin`，后`./zkServer.sh status`

![在这里插入图片描述][20210125125036509.png]

standalone表示zk是单节点，还没有搭建集群

1.  重启ZooKeeper服务，先`cd /opt/ZooKeeper/apache-zookeeper-3.5.6-bin/bin`，后`./zkServer.sh restart`

## 三、ZooKeeper数据模型 ##

1.  ZooKeeper是一个树形目录的服务，其数据模型和Linux的文件系统目录树类似

![在这里插入图片描述][20210125125047225.png]

1.  ZooKeeper中的每个节点都被称为ZNode，每个节点都会保存自己的值和(子)节点信息
2.  ZNode可以拥有子节点，同时也允许少量(1MB)的数据存储在该节点中
3.  ZooKeeper中的节点可以分为四大类

(1) PERSISTENT：持久化节点；此节点会一直保存直至手动删除  
(2) EPHEMERAL：临时节点；仅在命令行工具的会话中保留，会话结束，节点自动删除  
(3) PERSISTENT\_SEQUENTIAL：持久化顺序节点 (顺序节点ZK会自动的在其名称之后加一个序号)  
(4) EPHEMERAL\_SEQUENTIAL：临时顺序节点

## 四、ZooKeeper命令行操作 ##

1.  启动ZK命令行工具

先`cd /opt/ZooKeeper/apache-zookeeper-3.5.6-bin/bin`

后`./zkCli.sh -server IP地址:端口号`

![在这里插入图片描述][20210125125106573.png]

1.  常用命令 (所有路径必须以”/”打头)

(1) 关闭命令行工具：`quit`  
(2) 查看命令帮助：`help`  
(3) 查看指定目录下的子节点：`ls 目录`  
(4) 创建节点：`create /节点路径 [节点值]`，默认创建的是持久化节点  
(5) 获取节点值：`get /节点路径`  
(6) 设置节点值：`set /节点目录 节点值`  
(7) 删除单个节点：`delete /节点路径`  
(8) 删除带有子节点的节点：`deleteall /节点路径`  
(9) 创建临时节点：`create -e /节点路径 [节点值]`，执行`quit`退出命令行会话，节点删除  
(10) 创建顺序节点：`create -s /节点路径 [节点值]`

![在这里插入图片描述][20210125125151861.png]

(11) 查询节点详细信息：`ls -s /节点路径`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70]

详情介绍：  
![在这里插入图片描述][20210125125206614.png]

## 五、ZooKeeper JavaAPI操作 ##

1.  Curator介绍

Curator是ZooKeeper的java客户端库 ，目标是简化ZK客户端的使用

![在这里插入图片描述][20210125125219635.png]

1.  Curator的常用操作

建立连接、增删改查节点、Watch事件监听、分布式锁实现

1.  建立连接

(1) 在pom.xml中添加如下依赖

<dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>4.0.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>4.0.0</version>
    </dependency>

(2) 建立连接

// 方式一：
    //使用Curator首先要创建CuratorFramework接口的对象，创建此对象需要重试策略RetryPolicy的对象
    
    //1. 定义重试策略
    //RetryPolicy是一个接口，其有许多实现类，根据需要选择对应的实现类
    /** * 此策略参数1指定每次重试的间隔时间，参数2指定重试几次 */
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);
    
    //2. 创建CuratorFramework接口的对象
    /** * @param connectString String类型的ZK服务器的地址:端口，多个使用逗号隔开 * @param sessionTimeoutMs int类型会话超时时间，单位ms，默认60 * 1000，可省略 * @param connectionTimeoutMs int类型的连接超时时间，单位ms，默认15 * 1000，可省略 * @param retryPolicy 重试策略的对象 */
    CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.200.130:2181",retryPolicy);
    
    //3. 开启连接
    client.start(); //client.close()关闭连接
    
    // 方式二：
    //1. 定义重试策略
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);
    
    //2. 创建CuratorFramework接口的对象
    CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString("192.168.200.130:2181")
            .sessionTimeoutMs(60 * 1000)
            .connectionTimeoutMs(15 * 1000)
            .retryPolicy(retryPolicy)
            .namespace("test") //名称空间，所有节点的路径之前都包含/test，简化开发，可以不使用
            .build();
    
    //3. 开启连接
    client.start();

1.  创建节点

// 1. 基本创建
    /** * 1. 创建节点如果没有指定数据，则默认将当前主机的ip地址作为值存储 * 2. 创建的节点的父节点默认必须存在 * 3. 返回值是String类型的创建的节点的路径 */
    //forPath方法参数1指定节点，参数2指定值(Byte[]类型)(可以不指定)
    String path = client.create().forPath("/k1", "v1".getBytes());
    //创建出/test/k1节点，值为v1
    // 2. 创建不同类型的节点
    /** * 1. 默认是持久化节点 * 2. 使用withMode方法创建不同类型(节点的四种类型) */
    //创建临时节点，会话关闭自动消失(非命令行会话)
    client.create().withMode(CreateMode.EPHEMERAL).forPath("/k2");
    
    // 3. 创建多级节点
    /** * creatingParentsIfNeeded()如果父节点不存在则创建父节点 */
    client.create().creatingParentsIfNeeded().forPath("/k2/k2_1", "v2_1".getBytes());

1.  查询节点

// 1. 查询某一结点的值
    byte[] data = client.getData().forPath("/k1");
    System.out.println(new String(data)); //v1
    
    // 2. 查询子节点有哪些
    List<String> strings = client.getChildren().forPath("/");
    System.out.println(strings); //[k1, k2]
    //"/"表示节点/test (使用了名称空间)
    
    // 3. 查询某一节点详细信息
    //Stat是一个java bean，其中的属性就是之前所述详细信息的参数
    Stat stat = new Stat();
    //将获取到的信息存储在Stat对象的属性中
    client.getData().storingStatIn(stat).forPath("/k1");

1.  修改节点

// 1. 修改某一结点的值
    client.setData().forPath("/k1", "new_v1".getBytes());
    
    // 2. 根据某一结点的版本修改值
    // 每次修改值都会更新版本，使用此是为了防止多人修改一个节点时出错
    int version = new Stat().getVersion(); //获取当前版本
    client.setData().withVersion(version).forPath("/k2", "new_k2".getBytes());

1.  删除节点

// 1. 删除单个节点
    client.delete().forPath("/k2");
    
    // 2. 删除带有子节点的节点
    client.delete().deletingChildrenIfNeeded().forPath("/k2");
    
    // 3. 必须成功的删除
    //可能会由于网络抖动等原因导致删除失败，本质就是重试删除，直至成功
    client.delete().guaranteed().forPath("/k2");
    
    // 4. 回调删除
    //删除操作结束之后自动执行的回调函数
    client.delete().guaranteed().inBackground(new BackgroundCallback() { 
        @Override
        public void processResult(CuratorFramework client, CuratorEvent event) throws Exception { 
            System.out.println("执行了回调函数!");
        }
    }).forPath("/k2");
    
    // 关闭连接
    client.close();

1.  Watch事件监听

(1) ZK允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZK服务端会 将事件通知到感兴趣的客户端中，该机制是ZK实现分布式协调服务的重要特性

(2) ZK中引入了Watcher机制实现发布/订阅功能，能够让多个订阅者同时监听某一个对象，当 此对象状态发生变化的时候，会通知所有订阅者

(3) ZK提供了三种Watcher：

i. NodeCache：监听某一特定的节点  
ii. PathChildrenCache：监听某一结点的所有子节点 (并不监听此节点本身)  
iii. TreeCache：监听某一(子)树的所有结点 (NodeCache + PathChildrenCache)

//演示一：NodeCache
    //1. 创建NodeCache对象
    NodeCache nodeCache = new NodeCache(client, "/k1");
    //2. 注册监听 (无法监听特定的添加或者删除等事件，除了get操作其余均会触发此事件)
    nodeCache.getListenable().addListener(new NodeCacheListener() { 
        @Override
        public void nodeChanged() throws Exception { 
            System.out.println("数据变化了");
            //获取节点修改过后的值
            byte[] data = nodeCache.getCurrentData().getData();
            System.out.println("新值是：" + new String(data));
        }
    });
    // 3. 开启监听，如果设置为true，则开启监听时，如果之前的缓存记录中有满足监听条件的会显示出来
    nodeCache.start(true);
    for (;;) { 
        //死循环，可以持续监听
    }
    //运行结果：在命令行工具修改对应节点的值，在控制台打印出对应的语句

//演示二：PathChildrenCache
    // 1. 创建PathChildrenCache对象，设置为true，监听开启时加载之前的缓存数据
    PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/", true);
    
    // 2. 注册监听
    pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() { 
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { 
    
            System.out.println("节点发生变更！");
            System.out.println("event对象的值：" + event);
    
            //获取节点变更的类型，Type是内部枚举类
            PathChildrenCacheEvent.Type type = event.getType();
            System.out.println("获取到的变更类型：" + type);
    
            //判断变更类型是否为update，还有ADDED、REMOVED等类型
            if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) { 
                
                //获取修改后的值，连续使用两个getData()
                System.out.println("节点的值被修改成为：" + new String(event.getData().getData()));
            }
        }
    });
    
    // 3. 开启监听
    pathChildrenCache.start();
    for (;;) { 
        //死循环，可以持续监听
    }
    //在命令行工具中将/test/k1的值修改为newValue
    
    运行结果：
    首先在控制台按照childEvent()中的指定格式打印出之前满足条件的缓存记录

![在这里插入图片描述][20210125125334706.png]

//演示三：TreeCache
    // 1. 创建TreeCache对象
    TreeCache treeCache = new TreeCache(client, "/k1");
    // 2. 注册监听
    treeCache.getListenable().addListener(new TreeCacheListener() { 
        @Override
        public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception { 
            System.out.println("变更后的具体数组都在event对象中：" + event);
        }
    });
    // 3. 开启监听
    treeCache.start();
    for (;;) { 
        //死循环，可以持续监听
    }

1.  分布式锁实现

(1) 以前进行单机应用开发时，如果涉及到并发问题，通常采用synchronized或Lock的方式解 决并发问题，此时多线程运行在一个JVM下，没有任何问题

(2) 在应用处于分布式集群工作的情况下(属于多JVM下的工作环境)，上述两种方式无法解决跨 JVM的并发问题，即一台JVM加锁无法影响到另一台JVM

(3) 使用分布式锁的机制处理跨机器的进程之间的数据同步问题

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 1]

(4) ZK分布式锁的原理

核心思想：客户端获取锁，创建节点；使用完锁，删除节点

① 当客户端要获取锁时，在/lock节点(节点可任意)下创建一个临时顺序节点

i. 临时：保证锁一定会释放；客户端可能宕机导致无法删除持久化节点  
ii. 顺序：要找到比自己序号小的节点(根据下述得出)

② 要获取锁的所有客户端都获取到/lock节点下的所有子节点，某个客户端如果发现自己 创建的子节点序号最小，则认为该客户端获取到了锁，使用完锁，将该子节点删除

③ 如果发现自己创建的子节点并非/lock节点下的序号最小子节点，则表示该客户端并没 有获取到锁，此时该客户端需要找到比自己创建的节点序号小的一个邻居节点，同时对 其注册事件监听器，监听删除事件

④ 如果发现自己监听的节点被删除，则只有该客户端的Watcher会收到相应通知，此时再 次判断自己创建的节点是否为/lock节点下的序号最小子节点，如果是，表示该客户端 获取到了锁，如果不是则重复上述步骤获取比自己序号小的相邻节点，并注册监听

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 2]

(5) Curator实现分布式锁有五种方案：

i. InterProcessSemaphoreMutex：分布式排它锁  
ii. InterProcessMutex：分布式可重入排它锁 (使用较多)  
iii. InterProcessReadWriteLock：分布式读写锁  
iv. InterProcessMultiLock：将多个锁作为单个实体管理的容器  
v. InterProcessSemaphoreV2：共享信号量

(6) 模仿卖票机制

public class Ticket12306 implements Runnable{ 
    
        //票数
        private int tickets = 10;
    
        //定义分布式可重入排它锁
        private InterProcessMutex lock ;
    
        @Override
        public void run() { 
            while(true){ 
                //获取锁
                try { 
                    lock.acquire();
                    if(tickets > 0){ 
                        System.out.println(Thread.currentThread().getName()+":"+tickets);
                        Thread.sleep(100);
                        tickets--;
                    }
                } catch (Exception e) { 
                    e.printStackTrace();
                }finally { 
                    //释放锁
                    try { 
                        lock.release();
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        //构造函数
        public Ticket12306(){ 
            //重试策略
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);
            //创建客户端对象
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString("192.168.200.130:2181")
                    .sessionTimeoutMs(60 * 1000)
                    .connectionTimeoutMs(15 * 1000)
                    .retryPolicy(retryPolicy)
                    .build();
            //建立连接
            client.start();
            //为分布式可重入排它锁赋值，客户端创建的子节点将会存放在/lock目录下
            lock = new InterProcessMutex(client,"/lock");
        }
    }

public class TicketsTest { 
        public static void main(String[] args) throws Exception { 
            Ticket12306 ticket12306 = new Ticket12306();
            // 创建两个线程
            Thread t1 = new Thread(ticket12306, "抢票线程1：");
            Thread t2 = new Thread(ticket12306, "抢票线程2：");
            t1.start();
            t2.start();
        }
    }
    // 运行结束后在/lock节点下创建了两个临时顺序子节点

![在这里插入图片描述][20210125125420210.png]

## 六、ZooKeeper集群介绍 ##

1.  ZK集群中的角色

(1) Leader领导者

i. 处理事务请求(增删改)  
ii. 集群内部各服务器的调度者

(2) Follower 追随者

i. 处理非事务请求(查询)，转发事务请求给Leader服务器  
ii. 参与Leader选举投票

(3) Observer 观察者

i. 处理非事务请求(查询)  
ii. 转发事务请求给Leader服务器  
iii. 用于分担Follower追随者的压力

1.  Leader选举规则

(1) 服务器ID

zoo.cfg配置文件中，集群配置中服务  
器ID越大，此服务器选举中权重越大

(2) 数据ID

服务器中存放的最大数据ID值越大，此服务器选举中权重越大

(3) 如果某台服务器获得了超过半数的选票，此服务器成为Leader

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 3]

1.  ZK集群搭建，详见ZooKeeper集群搭建.PDF
2.  注意

(1) 集群中服务器的数量最好是奇数个，且>=3个

(2) 从服务器挂掉，不会影响集群的正常运行

(3) 可运行的机器要超过集群总数量的半数

(4) 主服务器挂掉，其余服务器会重新进行选举出Leader

(5) 产生了Leader之后，当有新的服务器加入，不会影响到现有Leader地位

[20210125124909254.png]: /images/20221119/c8dca8d1427f49919efdb6a92657d8fa.png
[20210125124912159.png]: /images/20221119/054ac5c08b72423783dc9beb7480dfdb.png
[2021012512491695.png]: https://img-blog.csdnimg.cn/2021012512491695.png
[20210125125036509.png]: https://img-blog.csdnimg.cn/20210125125036509.png
[20210125125047225.png]: https://img-blog.csdnimg.cn/20210125125047225.png
[20210125125106573.png]: https://img-blog.csdnimg.cn/20210125125106573.png
[20210125125151861.png]: https://img-blog.csdnimg.cn/20210125125151861.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210125125154111.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA==,size_16,color_FFFFFF,t_70
[20210125125206614.png]: https://img-blog.csdnimg.cn/20210125125206614.png
[20210125125219635.png]: https://img-blog.csdnimg.cn/20210125125219635.png
[20210125125334706.png]: https://img-blog.csdnimg.cn/20210125125334706.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210125125354830.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210125125402447.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA==,size_16,color_FFFFFF,t_70
[20210125125420210.png]: https://img-blog.csdnimg.cn/20210125125420210.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/2021012512543160.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA==,size_16,color_FFFFFF,t_70