学习Zookeeper

以下内容是在阅读大量博客并结合个人理解后整理总结的，作为个人学习笔记分享给大家。文末将附上相关的参考资料链接。如果其中有涉及到侵权的内容，请及时告知，我会立即删除相关部分。在此，特别感谢各位前辈的无私分享和指导，也希望本文能够对读者有所帮助。

1. 初识ZooKeeper

Zookeeper是Apache Hadoop项目下的一个子项目，是一个树形目录服务。

Zookeeper翻译过来就是动物园管理员，他是用来管Hadoop(大象）、Hive(蜜蜂)、Pig(小猪)的管理员。简称zk

Zookeeper是一个分布式的、开源的分布式应用程序的协调服务。

主要为了解决分布式架构下数据一致性问题， 它是集群的管理者， 监视着集群中各个节点的状态，根据节点提交的反馈进行下一步合理操作。 最终， 将简单易用的接口和性能高效、 功能稳定的系统提供给用户。

Zookeeper提供的主要功能包括：

分布式配置中心、分布式注册中心、分布式锁、分布式队列、集群选举、分布式屏障、发布/订阅等场景。

1.1 安装ZooKeeper

这个环境安装的我费劲死了。。

安装这一类的 一定注意jdk的 版本，不然真的会踩坑很多，今后我知道了 ，如果想入门一个技术栈，先把jdk1.8安装好，并且环境变量配置好，然后在忘后学习新的东西！！！！

• JDK1.8安装教程：https://blog.csdn.net/qq_40771292/article/details/107615925
• ZooKeeper安装教程：https://gongna-au.github.io/2022-10-14-test-markdown/

不用担心教程问题，一步一步来就好，遇到问题解决问题

1.2 ZooKeeper命令操作

1.2.1 Zookeeper数据模型

ZooKeeper是一个树形目录服务，其数据模型和Unix的文件系统目录树很类似，拥有一个层次化结构。

ZooKeeper 数据模型（Data model）采用层次化的多叉树形结构，每个节点上都可以存储数据，这些数据可以是数字、字符串或者是二级制序列。并且，每个节点还可以拥有 N 个子节点，最上层是根节点以/来代表。

每个数据节点在 ZooKeeper 中被称为 znode，它是 ZooKeeper 中数据的最小单元。并且，每个 znode 都一个唯一的路径标识。由于ZooKeeper 主要是用来协调服务的，而不是用来存储业务数据的，这种特性使得 Zookeeper 不能用于存放大量的数据，每个节点的存放数据上限为1M。

和文件系统一样，我们能够自由的增加、删除znode，在一个znode下增加、删除子znode，唯一的不同在于znode是可以存储数据的。默认有四种类型的znode：

• PERSISTENT持久化节点
• EPHEMERAL临时节点：-e
• PERSISTENT_SEQUENTIAL持久化顺序节点：-S
• EPHEMERAL_SEQUENTIAL临时顺序节点：-es

1.2.2 Zookeeper 服务端常用命令

当然要在zookeeper的bin：我的路径上为 ：/usr/local/zookeeper/apache-zookeeper/bin

1. 启动ZooKeeper服务：./zkServer.sh start
2. 查看ZooKeeper服务状态：./zkServer.sh status
3. 停止ZooKeeper服务：./zkServer.sh stop
4. 重启ZooKeeper服务：./zkServer.sh restart

1.2.3 Zookeeper客户端常用命令

首先连接服务端：./zkCli.sh -server ip:port

输入quit 退出

查看子节点：ls

创建节点：create /节点名称 节点内容

create的时候不能创建相同名称节点

获取数据：get /节点名称

如果在 create的时候不设置数据，则get的时候为null
通过set /节点path value 设置节点值

删除节点： delete /节点名称

查看全部命令：help

节点 可以 创建 四大类：create -e /app1 value

• PERSISTENT持久化节点
• EPHEMERAL临时节点：-e
• PERSISTENT_SEQUENTIAL持久化顺序节点：-S
• EPHEMERAL_SEQUENTIAL临时顺序节点：-es

znode 状态信息	解释
cZxid	create ZXID，即该数据节点被创建时的事务 id
ctime	create time，znode 被创建的毫秒数(从1970 年开始)
mZxid	modified ZXID，znode 最后更新的事务 id
mtime	modified time，znode 最后修改的毫秒数(从1970 年开始)
pZxid	znode 最后更新子节点列表的事务 id，只有子节点列表变更才会更新 pZxid，子节点内容变更不会更新
cversion	znode 子节点变化号，znode 子节点修改次数，子节点每次变化时值增加 1
dataVersion	znode 数据变化号，节点创建时为 0，每更新一次节点内容(不管内容有无变化)该版本号的值增加 1
aclVersion	znode 访问控制列表(ACL )版本号，表示该节点 ACL 信息变更次数
ephemeralOwner	如果是临时节点，这个是 znode 拥有者的 sessionid。如果不是临时节，则 ephemeralOwner=0
dataLength	znode 的数据长度
numChildren	znode 子节点数量

2. ZooKeeperJavaAPl操作

2.1 Curator介绍

Curator是Apache ZooKeeper 的Java客户端库。Curator 提供了一组易于使用的API和工具，简化了与ZooKeeper 的交互，同时提供了更高级别的抽象和功能。

常见的ZooKeeperJava API：

• 原生Java API
• ZkClient
• Curator

Curator项目的目标是简化ZooKeeper客户端的使用。Curator最初是Netfix研发的，后来捐献了Apache基金会，目前是Apache的顶级项目。 官网：http://curator.apache.org/

2.2 CuratorAPI常用操作

2.2.0 引入Curator支持

<!-- zookeeper支持 -->
<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.6.4</version>
</dependency>
<!-- curator-recipes -->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>5.5.0</version>
</dependency>
<!-- curator-framework -->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>5.5.0</version>
</dependency>

2.2.1 建立连接

//超时重试（连接间隔时间和超时连接次数）
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
//连接zookeeper对象
client = CuratorFrameworkFactory.newClient("ip:port",1000,60*1000,retryPolicy);
//开始连接
client.start();

2.2.2 添加节点

//1、创建节点并赋值
String path = client.create().forPath("/zuxia","helloworld".getBytes());
System.out.println("创建节点："+path);
​
//2、创建节点带子节点（如果不给子节点赋值，子节点的值默认为当前系统的IP地址）
String path = client.create().creatingParentsIfNeeded().forPath("/zuxia/abc");
System.out.println("创建节点："+path);
​
//3、创建临时节点（当断开连接时临时节点会自动删除,withMode中的属性可选择）
String path =client.create().withMode(CreateMode.EPHEMERAL).forPath("/a","helloworld".getBytes());
System.out.println("创建节点："+path);

2.2.3 删除节点

//1、删除节点
System.out.println("删除节点："+client.delete().forPath("/wjh"));
​
//2、删除带有子节点的目录节点
System.out.println("删除子节点："+client.delete().deletingChildrenIfNeeded().forPath("/zuxia"));

2.2.4 修改节点

//给节点赋值（返回值为Stat,可写可不写）
client.setData().forPath("/ab", "hello".getBytes());

2.2.5 查询节点

//1、查询节点的数据
byte[] bytes = client.getData().forPath("/zuxia");
System.out.println(new String(bytes));
​
//2、查询节点的数据（详情信息）
Stat stats=new Stat();
System.out.println(stats);//为了区分两个结果的不同
byte[] be = client.getData().storingStatIn(stats).forPath("/zuxia");
System.out.println(stats);

2.2.6 Watch事件监听

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

• ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。

ZooKeeper提供了三种Watcher：

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

• PathChildrenCache : 监控一个ZNode的子节点.

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

2.2.6.1 NodeCache 监听事件

@Test
void testNodeCache() throws Exception {
// 1. 创建NodeCache
NodeCache nodeCache = new NodeCache(client, "/ab");
// 2. 注册监听
nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("/ab节点发生变更");byte[] dataBytes = nodeCache.getCurrentData().getData();System.out.println("节点修改后的数据：" + new String(dataBytes));}
});
// 3. 开启监听，如果设置为true，则开启监听时，加载缓冲数据
nodeCache.start(true);
while(true){}
}

2.2.6.2 PathChildrenCache 监听事件

@Test
void testPathChildrenCache() throws Exception {
//创建监听对象(监听指定节点下的)
PathChildrenCache pathChildrenCache= new PathChildrenCache(client, "/zuxia", true);
//注册监听事件
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework cf, PathChildrenCacheEvent event) throws Exception {System.out.println("节点发生变化了");PathChildrenCacheEvent.Type type = event.getType();//当前判断的是当节点发生更新时进入改方法，可以选择添加或者删除的方法if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){byte[] bytes = event.getData().getData();System.out.println("节点修改后的数据"+new String(bytes));}}
});
//开启监听
pathChildrenCache.start();
while (true){}
}

2.2.6.3 TreeCache 监听事件

@Test
void testTreeCache() throws Exception {//创建监听对象TreeCache treeCache = new TreeCache(client, "/zuxia");//注册监听treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {System.out.println("节点发生变化了");TreeCacheEvent.Type type = treeCacheEvent.getType();if (type.equals(TreeCacheEvent.Type.NODE_ADDED)){System.out.println("节点添加了");}}});//开启监听treeCache.start();while (true){}
}

2.3 分布式锁

一种更加高级的锁机制，来处理种跨机器的进程之间的数据同步问题——这就是分布式锁

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

1. 客户端获取锁时，在lock节点下创建临时顺序节点。

2. 然后获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。

3. 如果发现自己创建的节点并非lock所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。

4. 如果发现比自己小的那个节点被删除，则客户端的Watcher会收到相应通知，此时再次判断自己创建的节点是否是lock子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

2.4 模拟12306售票案例

•在Curator中有五种锁方案：

•InterProcessSemaphoreMutex：分布式排它锁（非可重入锁）

•InterProcessMutex：分布式可重入排它锁

•InterProcessReadWriteLock：分布式读写锁

•InterProcessMultiLock：将多个锁作为单个实体管理的容器

•InterProcessSemaphoreV2：共享信号量

package com.wjh;
​
import org.apache.curator.RetryPolicy;
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;
​
public class TickTest implements Runnable{private int x=10;//票数//创建分布式可重入排它锁对象private InterProcessMutex lock;private CuratorFramework client;
​//当前方法的构造方法public TickTest() {//超时重试（连接间隔时间和超时连接次数）RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);//连接zookeeper对象client = CuratorFrameworkFactory.newClient("ip:port",1000,60*1000,retryPolicy);//开始连接client.start();//创建分布式可重入排它锁对象连接zookeeper注册中心客户端//客户端中不用创建，这里会自动创建lock = new InterProcessMutex(client, "/lock");}
​@Overridepublic void run() {try {//设置锁lock.acquire(3, TimeUnit.SECONDS);while (true) {if(x>0){//输出的调用线程的对象以及票数的数量System.out.println(Thread.currentThread()+"票数:" + x);//间隔200毫秒输出一次Thread.sleep(200);x--;}}} catch (Exception e) {throw new RuntimeException(e);}finally {try {//释放锁lock.release();} catch (Exception e) {throw new RuntimeException(e);
​}}}
}
​

3. ZooKeeper集群搭建

3.1 zookeeper集群介绍

Leader选举：

• Serverid ：服务器 ID

  • 比如有三台服务器，编号分别是1,2,3。
  • 编号越大在选择算法中的权重越大。

• Zxid ：数据 ID

  • 服务器中存放的最大数据ID.值越大说明数据 越新，在选举算法中数据越新权重越大。

• 在 Leader 选举的过程中，如果某台 ZooKeeper获得了超过半数的选票，则此ZooKeeper就可以成为Leader了

4. Zookeeper核心理论

对于集群来说，多加几台服务器就行（当然还得解决session共享，负载均衡等问题），而对于分布式来说，你首先需要将业务进行拆分，然后再加服务器，同时还要去解决分布式带来的一系列问题。比如各个分布式组件如何协调起来，如何减少各个系统之间的耦合度，如何处理分布式事务，如何去配置整个分布式系统，如何解决各分布式子系统的数据不一致问题等等。ZooKeeper 主要就是解决这些问题的。

许多著名的开源项目用到了 ZooKeeper，比如：

• Kafka : ZooKeeper 主要为 Kafka 提供 Broker 和 Topic 的注册以及多个 Partition 的负载均衡等功能。
• Hbase : ZooKeeper 为 Hbase 提供确保整个集群只有一个 Master 以及保存和提供 regionserver 状态信息（是否在线）等功能。
• Hadoop : ZooKeeper 为 Namenode 提供高可用支持。
• Dubbo：阿里巴巴集团开源的分布式服务框架，它使用 ZooKeeper 来作为其命名服务，维护全局的服务地址列表。

4.1 ZooKeeper的三种运行模式

ZooKeeper 有三种运行模式：单机模式、伪集群模式和集群模式。

• 单机模式：这种模式一般适用于开发测试环境，一方面我们没有那么多机器资源，另外就是平时的开发调试并不需要极好的稳定性。
• 集群模式：一个 ZooKeeper 集群通常由一组机器组成，一般 3 台以上就可以组成一个可用的 ZooKeeper 集群了。组成 ZooKeeper 集群的每台机器都会在内存中维护当前的服务器状态，并且每台机器之间都会互相保持通信。
  -伪集群模式：这是一种特殊的集群模式，即集群的所有服务器都部署在一台机器上。当你手头上有一台比较好的机器，如果作为单机模式进行部署，就会浪费资源，这种情况下，ZooKeeper 允许你在一台机器上通过启动不同的端口来启动多个 ZooKeeper 服务实例，从而以集群的特性来对外服务。

继续学习：
https://blog.csdn.net/dream_ambition/article/details/136027023