Zookeeper（从入门到掌握）看完这一篇就够了

文章目录

一、初识 Zookeeper
- 1.Zookeeper 概念
- 2.Zookeeper 数据模型
- 3.Zookeeper 服务端常用命令
- 4.Zookeeper 客户端常用命令
二、ZooKeeper JavaAPI 操作
- 1.Curator 介绍
- 1.Curator API 常用操作
- - （1）建立连接
  - （2）添加节点
  - （3）删除节点
  - （4）修改节点
  - （5）查询节点
  - （6）Watch事件监听
  - （7）分布式锁实现
  - （7）ZooKeeper分布式锁原理
  - （8）Curator实现分布式锁API
三、ZooKeeper集群
- - （1）Leader选举：
  - （2）ZooKeeper 集群搭建
  - （3）Zookeeper 集群角色

一、初识 Zookeeper

1.Zookeeper 概念

Zookeeper 是 Apache Hadoop 项目下的一个子项目，是一个树形目录服务。
Zookeeper 翻译过来就是动物园管理员，他是用来管 Hadoop（大象）、Hive(蜜蜂)、Pig(小猪)的管理员。简称zk
Zookeeper 是一个分布式的、开源的分布式应用程序的协调服务。
Zookeeper 提供的主要功能包括：
- 配置管理
- 分布式锁
- 集群管理

2.Zookeeper 数据模型

ZooKeeper 是一个树形目录服务,其数据模型和Unix的文件系统目录树很类似，拥有一个层次化结构。
这里面的每一个节点都被称为： ZNode，每个节点上都会保存自己的数据和节点信息。
节点可以拥有子节点，同时也允许少量（1MB）数据存储在该节点之下。
节点可以分为四大类：
PERSISTENT     持久化节点
EPHEMERAL     临时节点：-e
PERSISTENT_SEQUENTIAL     持久化顺序节点：-s
EPHEMERAL_SEQUENTIAL     临时顺序节点：-es
![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/a2b336f205e14b73b77d6ad1372404f7.png

3.Zookeeper 服务端常用命令

启动 ZooKeeper 服务: ./zkServer.sh start
查看 ZooKeeper 服务状态: ./zkServer.sh status
停止 ZooKeeper 服务: ./zkServer.sh stop
重启 ZooKeeper 服务: ./zkServer.sh restart

4.Zookeeper 客户端常用命令

连接ZooKeeper服务端：./zkCli.sh –server ip:port
断开连接：quit
查看命令帮助：help
显示指定目录下节点：ls 目录
创建节点：create /节点path value
获取节点值：get /节点path
设置节点值：set /节点path value
删除单个节点：delete /节点path
删除带有子节点的节点：deleteall /节点path
创建临时节点：create -e /节点path value
创建顺序节点：create -s /节点path value
查询节点详细信息：ls –s /节点path
- czxid：节点被创建的事务ID
- ctime: 创建时间
- mzxid: 最后一次被更新的事务ID
- mtime: 修改时间
- pzxid：子节点列表最后一次被更新的事务ID
- cversion：子节点的版本号
- dataversion：数据版本号
- aclversion：权限版本号
- ephemeralOwner：用于临时节点，代表临时节点的事务ID，如果为持久节点则为0
- dataLength：节点存储的数据的长度
- numChildren：当前节点的子节点个数

二、ZooKeeper JavaAPI 操作

1.Curator 介绍

Curator 是 Apache ZooKeeper 的Java客户端库。
常见的ZooKeeper Java API ：

原生Java API
ZkClient
Curator

Curator 项目的目标是简化 ZooKeeper 客户端的使用。
Curator 最初是 Netfix 研发的,后来捐献了 Apache 基金会,目前是 Apache 的顶级项目。
官网：http://curator.apache.org/
要注意Curator和zookeeper的版本对应关系，参照官网

1.Curator API 常用操作

（1）建立连接

引入依赖

	<!--curator--><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

创建测试类

package com.example.curator;import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.junit.Test;public class CuratorTest {//建立连接@Testpublic void testConnect() {/*** @param connectString         连接字符串，zk server地址和端口 "127.0.0.1:2181"* @param sessionTimeoutMs      会话超时时间，单位是毫秒ms* @param connectionTimeoutMs   连接超时时间，单位是毫秒ms* @param retryPolicy           重试策略*//*//重试策略RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);//第一种方式CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", 60 * 1000, 15 * 1000, retryPolicy);//开启连接client.start();*///第二种方式,链式创建RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retryPolicy).namespace("smsService").build();}
}

（2）添加节点

    @Testpublic void testCreate() throws Exception {CuratorFramework client = getClient();//基本创建String path = client.create().forPath("/app1");System.out.println(path);//带数据的创建client.create().forPath("/app1","haha".getBytes(StandardCharsets.UTF_8));//设置节点类型,临时类型，默认类型持久化client.create().withMode(CreateMode.EPHEMERAL).forPath("/app1");//创建多级节点  /app4/ios//creatingParentsIfNeeded,如果父节点不存在则创建父节点client.create().creatingParentsIfNeeded().forPath("/app4/ios");client.close();}

（3）删除节点

    @Testpublic void testDelete() throws Exception {CuratorFramework client = getClient();//删除单个节点client.delete().forPath("/app1");//删除带有子节点的节点client.delete().deletingChildrenIfNeeded().forPath("/app1");//必须删除成功client.delete().guaranteed().forPath("/app1");//回调client.delete().guaranteed().inBackground(new BackgroundCallback() {@Overridepublic void processResult(CuratorFramework client, CuratorEvent event) throws Exception {System.out.println("被删除了" + event);}}).forPath("/app1");client.close();}

（4）修改节点

    @Testpublic void testSet() throws Exception {CuratorFramework client = getClient();//简单修改数据client.setData().forPath("/app1", "wawa".getBytes(StandardCharsets.UTF_8));//判断版本信息一致再修改Stat stat = new Stat();client.getData().storingStatIn(stat).forPath("/app1");int version = stat.getVersion();client.setData().withVersion(version).forPath("/app1", "haha".getBytes(StandardCharsets.UTF_8));client.close();}

（5）查询节点

    @Testpublic void testSearch() throws Exception {CuratorFramework client = getClient();//查询数据  getbyte[] bytes = client.getData().forPath("/app1");System.out.println(new String(bytes));//查询子节点  lsList<String> path = client.getChildren().forPath("/");System.out.println(path);//查询节点状态信息  ls -sStat stat = new Stat();client.getData().storingStatIn(stat).forPath("/app1");System.out.println("====" +stat);client.close();}

（6）Watch事件监听

ZooKeeper 允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。
ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。
ZooKeeper 原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便
需要开发人员自己反复注册Watcher，比较繁琐。
Curator引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。
ZooKeeper提供了三种Watcher：

NodeCache : 只是监听某一个特定的节点

    @Testpublic void testNodeCache() throws Exception {CuratorFramework client = getClient();//删除NodeCache对象NodeCache nodeCache = new NodeCache(client, "/app");//注册监听nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化了");//拿到变化后的数据byte[] data = nodeCache.getCurrentData().getData();System.out.println(new String(data));}});//开启监听,如果设置为true，则开启监听时，加载缓冲数据nodeCache.start(true);client.close();}

PathChildrenCache : 监控一个ZNode的子节点.，感知不到自己的变化。

    @Testpublic void testPathChildrenCache() throws Exception {CuratorFramework client = getClient();//删除监听对象PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app", true);//注册监听pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {System.out.println("子节点变化了" + event);}});//开启监听,如果设置为true，则开启监听时，加载缓冲数据pathChildrenCache.start();client.close();}

TreeCache : 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合

    @Testpublic void testTreeCache() throws Exception {CuratorFramework client = getClient();//删除监听对象TreeCache treeCache = new TreeCache(client, "/");//注册监听treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {System.out.println("有节点变化了" + event);}});//开启监听,如果设置为true，则开启监听时，加载缓冲数据treeCache.start();client.close();}

（7）分布式锁实现

在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程间的代码同步问题，这时多线程的运行都是在同一个JVM之下，没有任何问题。
但当我们的应用是分布式集群工作的情况下，属于多JVM下的工作环境，跨JVM之间已经无法通过多线程的锁解决同步问题。
那么就需要一种更加高级的锁机制，来处理种跨机器的进程之间的数据同步问题——这就是分布式锁。
请添加图片描述

（7）ZooKeeper分布式锁原理

核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除该节点。

客户端获取锁时，在lock节点下创建临时顺序节点。
然后获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。
如果发现自己创建的节点并非lock所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。
如果发现比自己小的那个节点被删除，则客户端的Watcher会收到相应通知，此时再次判断自己创建的节点是否是lock子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

（8）Curator实现分布式锁API

在Curator中有五种锁方案：

InterProcessSemaphoreMutex：分布式排它锁（非可重入锁）
InterProcessMutex：分布式可重入排它锁
InterProcessReadWriteLock：分布式读写锁
InterProcessMultiLock：将多个锁作为单个实体管理的容器
InterProcessSemaphoreV2：共享信号量

三、ZooKeeper集群

（1）Leader选举：

Serverid：服务器ID
比如有三台服务器，编号分别是1,2,3。
编号越大在选择算法中的权重越大。
Zxid：数据ID
服务器中存放的最大数据ID.值越大说明数据越新，在选举算法中数据越新权重越大。
在Leader选举的过程中，如果某台ZooKeeper
获得了超过半数的选票，
则此ZooKeeper就可以成为Leader了。

（2）ZooKeeper 集群搭建

搭建要求

真实的集群是需要部署在不同的服务器上的，但是在我们测试时同时启动很多个虚拟机内存会吃不消，所以我们通常会搭建伪集群，也就是把所有的服务都搭建在一台虚拟机上，用端口进行区分。

我们这里要求搭建一个三个节点的Zookeeper集群（伪集群）。

准备工作

重新部署一台虚拟机作为我们搭建集群的测试服务器。

（1）安装JDK 【此步骤省略】。

（2）Zookeeper压缩包上传到服务器
（3）将Zookeeper解压，建立/usr/local/zookeeper-cluster目录，将解压后的Zookeeper复制到以下三个目录

/usr/local/zookeeper-cluster/zookeeper-1

/usr/local/zookeeper-cluster/zookeeper-2

/usr/local/zookeeper-cluster/zookeeper-3

[root@localhost ~]# mkdir /usr/local/zookeeper-cluster
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-1
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-2
[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-3

（4）创建data目录，并且将 conf下zoo_sample.cfg 文件改名为 zoo.cfg

mkdir /usr/local/zookeeper-cluster/zookeeper-1/data
mkdir /usr/local/zookeeper-cluster/zookeeper-2/data
mkdir /usr/local/zookeeper-cluster/zookeeper-3/datamv  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
mv  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
mv  /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

（5）配置每一个Zookeeper 的dataDir 和 clientPort 分别为2181 2182 2183

修改/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfgclientPort=2181
dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data

修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfgclientPort=2182
dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data

修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfgclientPort=2183
dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data

1.3 配置集群

（1）在每个zookeeper的 data 目录下创建一个 myid 文件，内容分别是1、2、3 。这个文件就是记录每个服务器的ID

echo 1 >/usr/local/zookeeper-cluster/zookeeper-1/data/myid
echo 2 >/usr/local/zookeeper-cluster/zookeeper-2/data/myid
echo 3 >/usr/local/zookeeper-cluster/zookeeper-3/data/myid

（2）在每一个zookeeper 的 zoo.cfg配置客户端访问端口（clientPort）和集群服务器IP列表。

集群服务器IP列表如下

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfgserver.1=192.168.149.135:2881:3881
server.2=192.168.149.135:2882:3882
server.3=192.168.149.135:2883:3883

解释：server.服务器ID=服务器IP地址：服务器之间通信端口：服务器之间投票选举端口

1.4 启动集群

启动集群就是分别启动每个实例。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start

在这里插入图片描述

启动后我们查询一下每个实例的运行状态

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

先查询第一个服务

请添加图片描述

Mode为follower表示是跟随者（从）

再查询第二个服务Mod 为leader表示是领导者（主）
在这里插入图片描述

查询第三个为跟随者（从）
在这里插入图片描述

模拟集群异常

（1）首先我们先测试如果是从服务器挂掉，会怎么样

把3号服务器停掉，观察1号和2号，发现状态并没有变化

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh stop/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

在这里插入图片描述

由此得出结论，3个节点的集群，从服务器挂掉，集群正常

（2）我们再把1号服务器（从服务器）也停掉，查看2号（主服务器）的状态，发现已经停止运行了。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh stop/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

在这里插入图片描述

由此得出结论，3个节点的集群，2个从服务器都挂掉，主服务器也无法运行。因为可运行的机器没有超过集群总数量的半数。

（3）我们再次把1号服务器启动起来，发现2号服务器又开始正常工作了。而且依然是领导者。

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

在这里插入图片描述

（4）我们把3号服务器也启动起来，把2号服务器停掉,停掉后观察1号和3号的状态。

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh stop/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

在这里插入图片描述

发现新的leader产生了~

由此我们得出结论，当集群中的主服务器挂了，集群中的其他服务器会自动进行选举状态，然后产生新得leader

（5）我们再次测试，当我们把2号服务器重新启动起来启动后，会发生什么？2号服务器会再次成为新的领导吗？我们看结果

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

在这里插入图片描述