Zookeeper命令操作（初始Zookeeper、JavaAPI操作、分布式锁实现、模拟12306售票分布式锁、Zookeeper集群搭建、选举投票）

1、初始Zookeeper

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

2、Zookeeper命令操作

2.1、Zookeeper数据模型

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

2.2、Zookeeper 服务端常用命令

在安装目录的zkServer.sh是作为linux系统的启动服务端，zkServer.cmd是作为wind系统的启动服务端，wind系统下可以直接双击启动（ZookeeperClient和ZookeeperJavaAPI可以作为客户端对服务端进行连接）

Linux系统操作，进入到解压的可执行文件/bin目录下进行执行命令

启动服务端

./zkServer.sh start

查看服务端状态

./zkServer.sh status

停止服务端

./zkServer.sh stop

重启服务端

./zkServer.sh restart

2.3、Zookeeper 客户端常用命令

需要启动服务端再进行连接

启动服务

./zkCli.sh -server localhost:2181
# 连接本机的可以不需要指定服务地址和端口号
./zkCli.sh

退出客户端

quit

查看节点下面包含的节点信息

所有的节点都是一级一级的往下查看的

ls / # 查看根节点
ls /dubbo # 查看dubbo节点
ls /dubbo/com.zcl.controller.UserServer # 查看dubbo下面的节点

添加数据【持久化】、获取数据、修改数据、删除数据

create /app [数据] # 在根目录下创建app节点，并带上数据，节点不能重复
get /app # 获取app节点数据
set /app # 修改app节点 数据
delete /app # 删除app节点数据
deleteall /app # 删除所有节点包括有子节点的
help # 查看命令帮助手册

创建顺序节点【关闭客户端就会没有】

临时顺序节点和持久顺序节点的编号都是连在一起的

# 创建临时节点app
create -e /app
# 创建持久的顺序节点app
create -s /app
# 创建临时的顺序节点app
create -es /app

3、Zookeeper JavaAPI操作

3.1、Curator 介绍

Curator 是 Apache ZooKeeper 的Java客户端库
常见的ZooKeeper Java API ：
- 原生Java API
- ZkClient
- Curator
Curator 项目的目标是简化 ZooKeeper 客户端的使用。
官网：http://curator.apache.org/

低版本的zookeeper不可以使用高版本的Curator，高版本的Curator可以操作低版本的zookeeper

3.2、Curator API 常用操作

创建maven项目工程

导入依赖

<dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.10</version><scope>test</scope></dependency><!--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><!--日志--><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.21</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.21</version></dependency>
</dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins>
</build>

创建日志文件

log4j.rootLogger=off,stdoutlog4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%d{yyyy-MM-dd HH/:mm/:ss}]%-5p %c(line/:%L) %x-%m%n

下面的测试文件都在tset中测试

3.2.1、建立连接的两种方式

CuratorFramework build = null;
/*** 创建连接的两种方式*/
@Before
public void test1(){// 重试策略ExponentialBackoffRetry retry = new ExponentialBackoffRetry(3000, 10);// 1、第一种方式/** Create a new client** @param connectString       连接字符串 zkserver 地址:端口 也可以配置集群* @param sessionTimeoutMs    会话超时时间 单位ms* @param connectionTimeoutMs 连接超时时间 单位ms* @param retryPolicy         失败重试策略* @return client*//*CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.26.131:2181",60 * 1000, 15 * 1000, retry);// 开启连接client.start();*/// 2、第二种连接方式build = CuratorFrameworkFactory.builder().connectString("192.168.26.131:2181").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retry).build();// 支持名称空间.namespace("节点名称")// 开启连接build.start();
}

3.2.2、添加节点【基本创建、带有数据创建、节点类型、多级创建】

/*** 创建节点 create 持久 临时 顺序 数据*/
@Test
public void createTest() throws Exception {// 1、基本创建节点【如果没有指定节点数据，会默认将当前客户端的ip作为数据存储】// System.out.println(client.create().forPath("/app1"));// 2、带有数据的创建节点【是字节】// client.create().forPath("/app2","hello".getBytes());// 3、创建节点的类型【临时的】// client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3","hello".getBytes());// 4、创建多级节点【如果父节点不存在则创建父节点】client.create().creatingParentsIfNeeded().forPath("/app4/p1");
}

节点类型有下面四种：

PERSISTENT：默认的持久类

PERSISTENT_SEQUENTIAL：持久顺序节点型

EPHEMERAL：临时的

EPHEMERAL_SEQUENTIAL：临时顺序型

3.2.3、查询节点

普通查询：client.getData().forPath("");

查询子节点：client.getChildren().forPath(）

查询节点状态： client.getData().storingStatIn(stat).forPath("")【需要一个Stat stat = new Stat()对象】

/*** get查询节点*/
@Test
public void getTest() throws Exception {byte[] bytes = client.getData().forPath("/app1");System.out.println(new String(bytes));
}/*** getChildren查询子节点*/
@Test
public void lsTest() throws Exception {List<String> strings = client.getChildren().forPath("/app4/p1");System.out.println(strings);
}/*** 查询子节点的状态信息*/
@Test
public void lsTest2() throws Exception {// 创建类型信息的空对象Stat stat = new Stat();client.getData().storingStatIn(stat).forPath("/app1");System.out.println(stat);
}

3.2.4、修改节点信息

基本修改：client.setData().forPath()

更具版本修改：client.setData().withVersion(version).forPath()

withVersion()是通过查询节点信息获取的，不可以自定义，为了防止其他客户端或线程进行干扰，修改完版本也会变

/*** 修改节点*/
@Test
public void testSet() throws Exception {// 创建类型信息的空对象Stat stat = new Stat();client.getData().storingStatIn(stat).forPath("/app1");// 获取节点的版本号int version = stat.getVersion();System.out.println(version);// client.setData().forPath("/app1","12".getBytes()); // 基本修改// 根据版本修改client.setData().withVersion(version).forPath("/app1","12".getBytes());
}

3.2.5、删除节点

删除单个节点：client.delete().forPath()

删除带有子节点的节点：client.delete().deletingChildrenIfNeeded().forPath()

必须成功的删除：为了防止网络抖动

回调

/*** 删除节点的多种方式及*/
@Test
public void testDelete() throws Exception {// 1、删除一个节点// client.delete().forPath("/app1");// 2、删除带有子节点的接待你// client.delete().deletingChildrenIfNeeded().forPath("/app4");// 3、必须删除成功，重试删除// client.delete().guaranteed().forPath("/app2");// 4、删除回调函数client.delete().guaranteed().inBackground((client, event) -> {System.out.println("我是回调函数");System.out.println(event);}).forPath("/app2");
}

3.2.6、Watch事件监听

ZooKeeper 允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。
ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。
ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。
Curator引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。

ZooKeeper提供了三种Watcher：

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

@Test
public void testNodeCache() throws Exception {// 1、创建nodecache对象final NodeCache nodeCache = new NodeCache(client,"/app1");// 2、注册监听nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化");// 获取修改后的节点信息byte[] data = nodeCache.getCurrentData().getData();System.out.println("获取修改后的节点："+new String(data));}});// 3、开启监听,如果设置为true则开启监听器，加载缓存数据nodeCache.start(true);// 保证监听一直运行，在测试中的使用方法while(true){}
}

运行测试文件，当节点信息被其他客户端修改的时候会被监听到并打印输出信息和修改后的信息

PathChildrenCache : 监控一个ZNode的子节点

如果是监听自身的节点发送变化是监听不到的

/*** 监控一个ZNode的子节点* @throws Exception*/
@Test
public void testChildrenCache2() throws Exception {// 1、创建nodecache对象PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app2", true);// 2、注册监听pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {System.out.println("子节点变化了");System.out.println(event);// 监听子节点的数据变更，并且拿到变更后的数据// 获取监听类型信息PathChildrenCacheEvent.Type type = event.getType();// 判断类型信息if(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){System.out.println("执行修改的操作信息");// 获取修改后的数据byte[] data = event.getData().getData();System.out.println(new String(data));}}});// 3、开启监听pathChildrenCache.start();// 保证监听一直运行，在测试中的使用方法while(true){}
}

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

自身变化和子节点变化都可以被监听到

/*** 以监控整个树上的所有节点* @throws Exception*/
@Test
public void testTreeCache3() throws Exception {// 1、创建nodecache对象TreeCache treeCache = new TreeCache(client, "/app2");// 2、注册监听treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {System.out.println("节点变化了，可能是自身也可能是子节点");System.out.println(event);}});// 3、开启监听treeCache.start();// 保证监听一直运行，在测试中的使用方法while(true){}
}

4、分布式锁实现

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

4.1、ZooKeeper分布式锁原理

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

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

4.2、模拟12306售票分布式锁

创建12306的售票实体类分布式锁

package com.zcl.Curator;import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;import java.util.concurrent.TimeUnit;/*** 项目名称：Curator* 描述：模拟12306售票类** @author zhong* @date 2022-06-15 21:02*/
public class Ticket12306 implements Runnable{/*** 数据库中存储的票数*/private int tickets = 10;/*** 模拟锁*/private InterProcessMutex lock;public Ticket12306() {ExponentialBackoffRetry retry = new ExponentialBackoffRetry(3000, 10);CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.26.131:2181").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retry).build();// 开去连接client.start();// 初始化锁lock = new InterProcessMutex(client,"/lock");}@Overridepublic void run() {while (true) {// 获取锁try {lock.acquire(3, TimeUnit.SECONDS);if(tickets > 0){System.out.println(Thread.currentThread() + ":"+tickets);Thread.sleep(300);tickets--;}} catch (Exception e) {e.printStackTrace();}finally {// 释放锁try {lock.release();} catch (Exception e) {e.printStackTrace();}}}}
}

创建启动方法，通过线程来模拟

/*** 项目名称：Curator* 描述：zookeeper分布式锁** @author zhong* @date 2022-06-15 21:02*/
public class LockTest {public static void main(String[] args) {Ticket12306 ticket12306 = new Ticket12306();// 创建客户端Thread t1 = new Thread(ticket12306,"携程");Thread t2 = new Thread(ticket12306,"飞猪");t1.start();t2.start();}
}

运行方法，查看输出信息，并不会出现出售相同的票

5、Zookeeper集群搭建

5.1、Zookeeper集群介绍

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

5.2、Zookeeper集群搭建

5.2.1、集群搭建要求

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

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

5.2.2、准备工作

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

（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 /opt/zooKeeper/apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-1
[root@localhost /]# cp -r /opt/zooKeeper/apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-2
[root@localhost /]# cp -r /opt/zooKeeper/apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-3
[root@localhost /]# cd /usr/local/zookeeper-cluster

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

[root@localhost zookeeper-cluster]# ll
总用量 0
drwxr-xr-x. 7 root root 146 6月  15 22:01 zookeeper-1
drwxr-xr-x. 7 root root 146 6月  15 22:01 zookeeper-2
drwxr-xr-x. 7 root root 146 6月  15 22:01 zookeeper-3
[root@localhost zookeeper-cluster]# mkdir /usr/local/zookeeper-cluster/zookeeper-1/data
[root@localhost zookeeper-cluster]# mkdir /usr/local/zookeeper-cluster/zookeeper-2/data
[root@localhost zookeeper-cluster]# mkdir /usr/local/zookeeper-cluster/zookeeper-3/data[root@localhost zookeeper-cluster]# mv /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
[root@localhost zookeeper-cluster]# mv /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
[root@localhost zookeeper-cluster]# 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

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

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

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

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

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

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

所有的保存退出都是按esc键然后输入:wq+回车完成保存退出

5.2.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.cfg
# 改为自己的ip地址
server.1=192.168.26.131:2881:3881
server.2=192.168.26.131:2882:3882
server.3=192.168.26.131:2883:3883

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

5.2.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

角色：follower也就是跟随者，二号服务为leader

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

/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

角色选举可以查看上面讲解

5.2.5、模拟集群故障观察选举过程

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

关闭三号服务，查看2、3号服务器的情况
```
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh stop
```
由此得出结论，3个节点的集群，从服务器挂掉，集群正常
我们再把1号服务器（从服务器）也停掉，查看2号（主服务器）的状态，发现已经停止运行了。
```
/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh stop/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status
```
由此得出结论，3个节点的集群，2个从服务器都挂掉，主服务器也无法运行。因为可运行的机器没有超过集群总数量的半数。

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

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

我们把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

我们再次测试，当我们把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
```
我们会发现，2号服务器启动后依然是跟随者（从服务器），3号服务器依然是领导者（主服务器），没有撼动3号服务器的领导地位。

由此我们得出结论，当领导者产生后，再次有新服务器加入集群，不会影响到现任领导者

6、Zookeeper集群角色

在Zookeeper集群服务中有三个角色：

Leader领导者
- 处理事务请求
- 集群内部各服务器的调度
Follower跟随着
- 处理客户端非事务处理，转发事务处理给leader器
- 参与Leader选举投票
Observer观察者
- 处理客户端非事务请求，转发事务请求给leader服务器