ZooKeeper

• 1. 理论基础知识
• • 1.1 概述
  • 1.2 架构组成
  • 1.3 主要特点
  • 1.4 数据结构
  • • 1.4.1 ZNode 的类型
    • 1.4.2 ZNode 的状态信息
    • 1.4.3 Watcher 机制
  • 1.5 Leader 选举机制
  • 1.6 ZAB 协议
• 2. 操作实践
• • 2.1 shell 命令行
  • • 2.1.1 常用基础命令
    • • 2.1.1.1 创建节点
      • 2.1.1.2 读取节点
      • 2.1.1.3 更新节点
      • 2.1.1.4 删除节点
  • 2.2 Java API
  • • 2.2.1 创建节点
    • 2.2.2 删除节点
    • 2.2.3 监听节点变化
    • 2.2.4 获取数据（节点是否存在、更新和删除）

---

1. 理论基础知识

 

1.1 概述

ZooKeeper 是一个致力于分布式协调服务的开源框架，主要是用来解决分布式集群中经常受困扰的一致性问题（比如避免同时操作同一数据时造成的数据脏读问题）。
从本质来看，ZooKeeper 其实是一个分布式的小文件存储系统，提供目录树的数据存储方式，并可以对树中的节点进行有效管理（也就像平常电脑系统对文件夹和文件的管理）。
ZooKeeper 还提供了给客户端监控存储在其中内部数据的功能。从而可以达成基于数据的集群管理。

---

1.2 架构组成

主要角色：
Leader —— 核心组件且是事务请求的唯一处理者。所谓 ZooKeeper 的事务，类似于 create，setData， delete 等有写操作的请求，并且需要 Leader 需要决定编号、执⾏操作，这个过程称为⼀个事务。还要负责选举投票的发起和决议并更新系统状态。
Follower —— 主要是处理非事务请求（如 read 请求），如果遇到来自 client 的事务请求，就会转发给 Leader。除此以外，Follower 还要参与 Leader 的选举。
Observer —— 观察并更新 ZooKeeper 的最新状态，且给 Leader 转发事务请求，不参与 Leader 选举。
ZooKeeper 也是 master/slave 架构，只不过他的 master 不是通过配置指定，而是通过选举策略产生。

---

1.3 主要特点

① ZooKeeper 集群半数以上节点存活，就能正常进行服务。
② ZooKeeper 遵循数据一致性原则，无论是 Follower 还是 Leader 接收 read 请求，最后每个 server 都会有一份相同的数据。
③ ZooKeeper 的更新请求会按照顺序执行。
④ 数据更新具有原子性，会很明确地反馈给客户端操作是否成功。

---

1.4 数据结构

如上图，ZooKeeper 的信息会保存在一个个的数据节点中，这些节点被称为 ZNode ，是 ZooKeeper 中的最小数据单位，而图中 ZNode 下挂载一个或者多个 ZNode 的结构被称为 ZNode Tree，类似于文件系统的层级树状结构。ZooKeeper 的节点路径标识和 Unix 文件系统路径相似，/ 表示根路径，并由一系列 / 作为路径的分隔符。数据节点可以被写入数据，也可以在下面继续创建节点。

1.4.1 ZNode 的类型

持久节点 —— 最常见的数据节点，类似于 Unix 系统的目录或文件，创建后一直会存在于服务器，只有主动删除操作时才能清除。
持久顺序节点 —— 也就是有顺序的持久节点，和持久节点不同点在于节点创建时会在后面加上一个数字后缀表示其顺序。
临时节点 —— 和持久节点相反，临时节点不会一直存在，它的生命周期和客户端会话绑定在一起，客户端会话结束时对应的临时节点也会一起删除。要注意的是临时节点下不能继续创建节点。
临时顺序节点 —— 就是有顺序、创建时节点后会有数据后缀表示顺序的临时节点。

1.4.2 ZNode 的状态信息

先了解一下 事务 ID 的概念
我们经常听到的事务是对物理和抽象的应用状态上的操作集合。而一般提到的事务是指数据库事务，包含了一系列对数据库有序的读写操作，这些数据库事务具有 ACID 特性 —— 原子性（Atomic）、一致性（Consistency）、隔离性（Isolation）、持久性（Durability）。
这里事务 ID 中的事务指的是 ZooKeeper 中的事务，也即能改变 ZooKeeper 服务器状态的操作，一般包括数据节点的创建、删除和内容更新等操作。那么对于每一个事务请求，ZooKeeper 都会分配一个全局唯一的事务 ID，用 ZXID 表示。每一个 ZXID 对应一次更新操作，从 ZXID 可以间接识别出这些操作的全局顺序。

1.4.3 Watcher 机制

ZooKeeper 主要使用 Watcher 机制来实现分布式数据的发布/订阅功能。⼀个典型的发布/订阅模型系统定义了⼀种 ⼀对多的订阅关系，能够让多个订阅者同时监听某⼀个主题对象，当这个主题对象⾃身状态变化时，会通知所有订阅者，使它们能够做出相应的处理。

---

1.5 Leader 选举机制

---

1.6 ZAB 协议

ZAB 协议全称为 ZooKeeper 原子消息广播协议（ZooKeeper Atomic Broadcast），是 ZooKeeper 专门设计的一种支持崩溃恢复和原子广播协议。
ZAB 的处理模式是所有客户端数据首先写入 Leader，然后 Leader 将数据复制给 Follower，更具体地来说，是将服务器数据的状态变更以事务 proposal 的形式广播到所有的副本进程，ZAB 能够保证每个事务操作的一个全局变更ID（也就是 ZXID）。
而这个广播过程类似于二阶段提交过程。

---

2. 操作实践

 

2.1 shell 命令行

ZooKeeper 拥有自己的客户端，需要用户进入到 ZooKeeper 的 bin 目录下，使用 ./zkCli.sh 启动客户端（如果需要指定服务器，./zkCli.sh -server ip:port(默认2181) 即可）。

2.1.1 常用基础命令

2.1.1.1 创建节点

操作	命令	示例
创建持久节点	create path data	create /permanent 123
创建临时节点	create -e path data	create -e /tmp 123
创建顺序节点	create -s path data	create -s /sort 123

2.1.1.2 读取节点

命令	说明
ls path	该命令可以列出当前节点下的所有子节点，但只能看第一级子节点，其实就和 linux 命令的 ls 一样
get path	该命令是获取当前节点的数据内容和属性，第一行为数据内容，剩下为属性信息，详情可见 1.4.2 的插图

2.1.1.3 更新节点

更新节点的命令为 set path data，命令执行完毕后，所属节点的数据内容会发生变化，除此以外，dataVersion 也会进行更新。

2.1.1.4 删除节点

命令为 delete path ，该命令如果当前节点下拥有子节点则无法进行删除，如果想删除节点以及其包含的子节点，需要用到命令 rmr path

---

2.2 Java API

2.2.1 创建节点

public class Create_Node_Sample {
public static void main(String[] args) {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181");
	System.out.println("ZooKeeper session established.");
	//createParents的值设置为true，可以递归创建节点
	zkClient.createPersistent("/lg-zkClient/lg-c1",true);
	System.out.println("success create znode.");
 }
}

2.2.2 删除节点

public class Del_Data_Sample {
public static void main(String[] args) throws Exception {
	String path = "/lg-zkClient/lg-c1";
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 5000);
	zkClient.deleteRecursive(path);
	System.out.println("success delete znode.");
 }
}

2.2.3 监听节点变化

public class Get_Child_Change {
	public static void main(String[] args) throws InterruptedException {
		//获取到zkClient
		final ZkClient zkClient = new ZkClient("linux121:2181");
		//zkClient对指定⽬录进⾏监听(不存在⽬录:/lg-client)，指定收到通知之后的逻辑
		//对/lag-client注册了监听器，监听器是⼀直监听
		zkClient.subscribeChildChanges("/lg-client", new IZkChildListener() {
		//该⽅法是接收到通知之后的执⾏逻辑定义
		public void handleChildChange(String path, List<String> childs)
		throws Exception {
		//打印节点信息
		System.out.println(path + " childs changes ,current childs " + childs);
		 }
		 });
		//使⽤zkClient创建节点，删除节点，验证监听器是否运⾏
		zkClient.createPersistent("/lg-client");
		Thread.sleep(1000); //只是为了⽅便观察结果数据
		zkClient.createPersistent("/lg-client/c1");
		Thread.sleep(1000);
		zkClient.delete("/lg-client/c1");
		Thread.sleep(1000);
		zkClient.delete("/lg-client");
		Thread.sleep(Integer.MAX_VALUE);
		/*
		1 监听器可以对不存在的⽬录进⾏监听
		2 监听⽬录下⼦节点发⽣改变，可以接收到通知，携带数据有⼦节点列表
		3 监听⽬录创建和删除本身也会被监听到
		*/
 	}
}

2.2.4 获取数据（节点是否存在、更新和删除）

//使⽤监听器监听节点数据的变化
public class Get_Data_Change {
	public static void main(String[] args) throws InterruptedException {
		// 获取zkClient对象
		final ZkClient zkClient = new ZkClient("linux121:2181");
		//设置⾃定义的序列化类型,否则会报错！！
		zkClient.setZkSerializer(new ZkStrSerializer());
		//判断节点是否存在，不存在创建节点并赋值
		final boolean exists = zkClient.exists("/lg-client1");
		if (!exists) {
			zkClient.createEphemeral("/lg-client1", "123");
		 	}
		//注册监听器，节点数据改变的类型，接收通知后的处理逻辑定义
		zkClient.subscribeDataChanges("/lg-client1", new IZkDataListener() {
		public void handleDataChange(String path, Object data) throws
		Exception {
			//定义接收通知之后的处理逻辑
			System.out.println(path + " data is changed ,new data " +
			data);
			 }
			 
		//数据删除--》节点删除
			public void handleDataDeleted(String path) throws Exception {
				System.out.println(path + " is deleted!!");
			 }
		 });
		//更新节点的数据，删除节点，验证监听器是否正常运⾏
		final Object o = zkClient.readData("/lg-client1");
		System.out.println(o);
		zkClient.writeData("/lg-client1", "new data");
		Thread.sleep(1000);
		//删除节点
		zkClient.delete("/lg-client1");
		Thread.sleep(Integer.MAX_VALUE);
	 }
}

public class ZkStrSerializer implements ZkSerializer {
	//序列化，数据--》byte[]
	public byte[] serialize(Object o) throws ZkMarshallingError {
		return String.valueOf(o).getBytes();
	 }
	//反序列化，byte[]--->数据
	public Object deserialize(byte[] bytes) throws ZkMarshallingError {
		return new String(bytes);
	 }
}