# zookeeper_learning
**Repository Path**: sun81911/zookeeper_learning
## Basic Information
- **Project Name**: zookeeper_learning
- **Description**: zookeeper知识点学习
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 0
- **Forks**: 2
- **Created**: 2020-06-12
- **Last Updated**: 2020-12-19
## Categories & Tags
**Categories**: Uncategorized
**Tags**: None
## README
## 入门
### 概述
Zookeeper是一个开源的分布式的，为分布式应用提供协调服务的Apache项目
#### 工作机制
Zookeeper从设计模式角度来理解，是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应
#### 特点
Zookeeper是由一个领导者（Leader），多个跟随者（Follower）组成的集群
集群中只要有半数以上节点存活，Zookeeper集群就能正常服务
全局数据一致指每个Server保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的
更新请求顺序进行，来自同一个Client的更新请求按其发送顺序依次执行
数据更新原子性，一次数据更新要么全部成功，要么全部失败
实时性，在一定时间范围内，Client能读到最新数据
#### 数据结构
ZooKeeper数据模型的结构与Unix文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode
每一个ZNode默认能够存储1MB的数据，每个ZNode都可以通过其路径唯一标识
#### 应用场景
提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等
##### 统一命名服务
在分布式环境下，经常需要对应用/服务进行统一命名，便于识别
##### 统一配置管理
分布式环境下，配置文件同步非常重要
一般要求一个集群中，所有节点的配置信息是一致的，对配置文件修改后，希望能够快速同步到各个节点上
配置管理可交由ZooKeeper实现
1. 可将配置信息写入ZooKeeper上的一个Znode
2. 各个客户端服务器监听这个Znode
3. 一旦Znode中的数据被修改，ZooKeeper将通知各个客户端服务器
##### 统一集群管理
分布式环境中，实时掌握每个节点的状态是必要的，可根据节点实时状态做出一些调整
ZooKeeper可以实现实时监控节点状态变化
1. 可将节点信息写入ZooKeeper上的一个ZNode
2. 监听这个ZNode可获取它的实时状态变化
##### 服务器动态上下线
客户端能实时洞察到服务器上下线的变化
##### 软负载均衡
在Zookeeper中记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户端请求
## 安装
### 本地模式安装部署
**准备工作**
1. 安装JDK
2. 拷贝Zookeeper安装包到Linux系统下
3. 解压到指定目录
```shell
tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/
```
**配置修改**
1. 将/opt/module/zookeeper-3.4.10/conf这个路径下的zoo_sample.cfg修改为zoo.cfg
```shell
cp zoo_sample.cfg zoo.cfg
```
2. 编辑zoo.cfg文件，修改dataDir路径
```shell
vim zoo.cfg
```
```
dataDir=/opt/module/zookeeper-3.4.10/zkData
```
3. 创建zkData文件夹
```shell
mkdir zkData
```
**操作 Zookeeper**
1. 启动Zookeeper
```shell
bin/zkServer.sh start
```
2. 查看进程是否启动
```shell
jps
```
3. 查看状态
```shell
bin/zkServer.sh status
```
4. 启动客户端
```shell
bin/zkCli.sh
```
5. 退出客户端
```shell
[zk: localhost:2181(CONNECTED) 0] quit
```
6. 停止Zookeeper
```shell
bin/zkServer.sh stop
```
### 配置参数分析
Zookeeper中的配置文件 zoo.cfg 中参数含义分析
**tickTime**
```
# The number of milliseconds of each tick
tickTime=2000
```
通信心跳数，Zookeeper服务器与客户端心跳时间，单位毫秒
Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳，时间单位为毫秒
它用于心跳机制，并且设置最小的 session 超时时间为两倍心跳时间，即session的最小超时时间是 `2 * tickTime`
**initLimit**
```
# The number of ticks that the initial synchronization phase can take
initLimit=10
```
Leader 和 Follower 初始通信时限
集群中的 Follower 跟随者服务器与 Leader 领导者服务器之间初始连接时能容忍的最多心跳数，tickTime的数量，用它来限定集群中的Zookeeper服务器连接到 Leader 的时限
**syncLimit**
```
# The number of ticks that can pass between sending a request and getting an acknowledgement
syncLimit=5
```
Leader 和 Follower 同步通信时限
集群中 Leader 与 Follower 之间的最大响应时间单位，假如响应超过 `syncLimit * tickTime`，Leader认为Follwer死掉，从服务器列表中删除Follwer
**dataDir**
```
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just example sakes.
dataDir=/opt/module/zookeeper-3.4.10/zkData
```
数据文件目录以及数据持久化路径，主要用于保存Zookeeper中的数据
**clientPort**
```
# the port at which the clients will connect
clientPort=2181
```
客户端连接端口，监听客户端连接的端口
## 内部原理
### 选举机制
半数机制：集群中半数以上机器存活时集群可用，所以Zookeeper适合安装奇数台服务器
Zookeeper虽然在配置文件中并没有指定Master和Slave。但是 Zookeeper 工作时，是有一个节点为Leader，其他则为Follower，Leader是通过内部的「选举机制」临时产生的
**选举过程**
假设有五台服务器组成的Zookeeper集群，它们的id从1 - 5，同时它们都是最新启动的，也就是没有历史数据，在存放数据量这一点上，都是一样的
1. 服务器1启动，此时只有它一台服务器启动了，它发出去的报文没有任何响应，所以它的选举状态一直是 LOOKING 状态
2. 服务器2启动，它与最开始启动的服务器1进行通信，互相交换自己的选举结果，由于两者都没有历史数据，所以id值较大的服务器2胜出，但是由于没有达到超过半数以上的服务器都同意选举它，所以服务器1、2还是继续保持 LOOKING 状态
3. 服务器3启动，根据理论分析，服务器3成为目前id值最大的服务器，而与上面不同的是，此时有三台服务器选举了它，超过半数，所以它成为了这次选举的Leader
4. 服务器4启动，根据前面的分析，理论上服务器4应该是服务器1、2、3、4中最大的，但是由于前面已经有半数以上的服务器选举了服务器3，所以它只能成为Follower
5. 服务器5启动，同服务器4一样
### 节点类型
**分类**
1. 持久 Persistent：客户端和服务器端断开连接后，创建的节点不删除
2. 短暂 Ephemeral：客户端和服务器端断开连接后，创建的节点自己删除
**持久化目录节点**
客户端与Zookeeper断开连接后，该节点依旧存在
**持久化顺序编号目录节点**
客户端与Zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号
**临时目录节点**
客户端与Zookeeper断开连接后，该节点被删除
**临时顺序编号目录节点**
客户端与Zookeeper断开连接后，该节点被删除，只是Zookeeper给该节点名称进行顺序编号
**注意**
创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护
在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序
### Stat 结构体
**结构体数据示例**
```
cZxid = 0x10000000e
ctime = Wed May 27 22:49:49 CST 2020
mZxid = 0x10000001b
mtime = Wed May 27 23:10:42 CST 2020
pZxid = 0x100000020
cversion = 7
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 5
numChildren = 1
```
**数据分析**
1. cZxid：创建节点的事务zxid
每次修改ZooKeeper状态都会收到一个zxid形式的时间戳，也就是ZooKeeper事务ID
事务ID是ZooKeeper中所有修改总的次序。每个修改都有唯一的zxid，如果zxid1小于zxid2，那么zxid1在zxid2之前发生
2. ctime：znode被创建的时间
3. mZxid：znode最后更新的事务zxid
4. mtime：znode最后修改的时间
5. pZxid：znode最后更新的子节点zxid
6. cversion：znode子节点变化号，znode子节点修改次数
7. dataVersion：znode数据变化号
8. aclVersion：znode访问控制列表的变化号
9. ephemeralOwner：如果是临时节点，这个是znode拥有者的session id；如果不是临时节点则为0
10. dataLength：znode的数据长度
11. numChildren：znode子节点数量
### 监听器原理
**图示分析**
**流程分析**
1. 生成一个main()线程
2. 在main线程中创建Zookeeper客户端，这时会创建两个线程，一个负责网络连接通信（connet），一个负责监听（listener）
3. 通过connect线程将注册的监听事件发送给Zookeeper
4. 在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中
5. Zookeeper监听到有数据或路径变化，就会将这个消息发送给listener线程
6. listener线程内部调用了process()方法
**常见的监听**
监听节点数据的变化
```shell
get path [watch]
```
监听子节点增减的变化
```shell
ls path [watch]
```
### 写数据流程
**图示分析**
**流程分析**
1. Client 向 ZooKeeper 的 Server1 上写数据，发送一个写请求
2. 如果Server1不是Leader，那么Server1 会把接受到的请求进一步转发给Leader，因为每个ZooKeeper的Server里面有一个是Leader
这个Leader 会将写请求广播给各个Server，比如Server1和Server2，各个Server写成功后就会通知Leader
3. 当Leader收到大多数 Server 数据写成功了，那么就说明数据写成功了
如果这里三个节点的话，只要有两个节点数据写成功了，那么就认为数据写成功了
写成功之后，Leader会告诉Server1数据写成功了
4. Server1会进一步通知 Client 数据写成功了，这时就认为整个写操作成功
## 实战操作
### 分布式安装部署
**集群规划**
在hadoop198、hadoop197和hadoop196三个节点上部署Zookeeper
**解压安装**
解压Zookeeper安装包到/opt/module/目录下
```shell
tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/
```
同步 /opt/module/zookeeper-3.4.10 目录内容到hadoop197、hadoop196服务器
```shell
xsync zookeeper-3.4.10/
```
**配置服务器编号**
在 /opt/module/zookeeper-3.4.10/zkData 目录下创建编辑一个myid的文件
```shell
vim myid
```
在文件中添加与 server 对应的编号
```
1
```
拷贝配置好的zookeeper到其他机器上，并分别在其他机器上修改myid文件中内容为`2`、`3`
```shell
xsync myid
```
也可直接使用写命令
```shell
echo "1" > myid
```
**配置zoo.cfg文件**
编辑文件
```shell
vim zoo.cfg
```
添加配置
```
# cluster
server.1=hadoop198:2888:3888
server.2=hadoop197:2888:3888
server.3=hadoop196:2888:3888
```
同步zoo.cfg配置文件
```shell
xsync zoo.cfg
```
**配置参数分析**
```
server.A=B:C:D
```
1. A 是一个数字，表示这个是第几号服务器
集群模式下配置文件myid，这个文件在dataDir目录下，这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，拿到里面的数据与 zoo.cfg 里面的配置信息比较从而判断到底是哪个server
2. B 是这个服务器的ip地址
3. C 是这个服务器与集群中的Leader服务器交换信息的端口
4. D 是若集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口
**集群操作**
分别启动Zookeeper
```shell
bin/zkServer.sh start
```
查看状态
```shell
bin/zkServer.sh status
```
### 客户端命令行操作
**部分命令**
| 命令基本语法 | 功能描述 |
| ---------------- | -------------------------------------------------- |
| help | 显示所有操作命令 |
| ls path [watch] | 使用 ls 命令来查看当前znode中所包含的内容 |
| ls2 path [watch] | 查看当前节点数据并能看到更新次数等数据 |
| create | 普通创建，-s 含有序列，-e 临时，重启或者超时消失 |
| get path [watch] | 获得节点的值 |
| set | 设置节点的具体值 |
| stat | 查看节点状态 |
| delete | 删除节点 |
| rmr | 递归删除节点 |
**具体操作**
启动客户端
```shell
bin/zkCli.sh
```
显示所有操作命令
```shell
[zk: localhost:2181(CONNECTED) 0] help
```
查看当前znode中所包含的内容
```shell
[zk: localhost:2181(CONNECTED) 1] ls /
```
查看当前节点详细数据
```shell
[zk: localhost:2181(CONNECTED) 2] ls2 /
```
分别创建2个普通节点
```shell
[zk: localhost:2181(CONNECTED) 3] create /sanguo "jinlian"
```
```shell
[zk: localhost:2181(CONNECTED) 4] create /sanguo/shuguo "liubei"
```
获得节点的值
```shell
[zk: localhost:2181(CONNECTED) 9] get /sanguo/shuguo
```
```
liubei
cZxid = 0x10000000f
ctime = Wed May 27 22:50:11 CST 2020
mZxid = 0x10000000f
mtime = Wed May 27 22:50:11 CST 2020
pZxid = 0x10000000f
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
```
创建短暂节点
```shell
[zk: localhost:2181(CONNECTED) 10] create -e /sanguo/wuguo "zhouyu"
```
1. 在当前客户端是能查看到的
```shell
[zk: localhost:2181(CONNECTED) 12] ls /sanguo
```
```
[wuguo, shuguo]
```
2. 退出当前客户端然后再重启客户端
```shell
[zk: localhost:2181(CONNECTED) 13] quit
```
```shell
bin/zkCli.sh
```
3. 再次查看根目录下短暂节点已经删除
```shell
[zk: localhost:2181(CONNECTED) 0] ls /sanguo
```
```
[shuguo]
```
创建带序号的节点
1. 先创建一个普通的根节点/sanguo/weiguo
```shell
[zk: localhost:2181(CONNECTED) 3] create /sanguo/weiguo "caocao"
```
2. 创建带序号的节点
```shell
[zk: localhost:2181(CONNECTED) 4] create -s /sanguo/weiguo/xiaoqiao "jinlian"
```
```
Created /sanguo/weiguo/xiaoqiao0000000000
```
```shell
[zk: localhost:2181(CONNECTED) 6] create -s /sanguo/weiguo/daqiao "jinlian"
```
```
Created /sanguo/weiguo/daqiao0000000001
```
**细节**：如果原来没有序号节点，序号从0开始依次递增；如果原节点下已有2个节点，则再排序时从2开始，以此类推
修改节点数据值
```shell
[zk: localhost:2181(CONNECTED) 7] set /sanguo/weiguo "simayi"
```
节点的值变化监听
1. 在hadoop196主机上注册监听/sanguo节点数据变化
```shell
[zk: localhost:2181(CONNECTED) 1] get /sanguo watch
```
2. 在hadoop197主机上修改/sanguo节点的数据
```shell
[zk: localhost:2181(CONNECTED) 0] set /sanguo "roger"
```
3. 观察hadoop196主机收到数据变化的监听
```
WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo
```
节点的子节点变化监听（路径变化）
1. 在hadoop196主机上注册监听/sanguo节点的子节点变化
```shell
[zk: localhost:2181(CONNECTED) 2] ls /sanguo watch
```
2. 在hadoop197主机/sanguo节点上创建子节点
```shell
[zk: localhost:2181(CONNECTED) 1] create /sanguo/roger "roger"
```
3. 观察hadoop196主机收到子节点变化的监听
```
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo
```
删除节点
```shell
[zk: localhost:2181(CONNECTED) 12] delete /sanguo/roger
```
递归删除节点
```shell
[zk: localhost:2181(CONNECTED) 17] rmr /sanguo/weiguo
```
查看节点状态
```shell
[zk: localhost:2181(CONNECTED) 20] stat /sanguo
```
```
cZxid = 0x10000000e
ctime = Wed May 27 22:49:49 CST 2020
mZxid = 0x10000001b
mtime = Wed May 27 23:10:42 CST 2020
pZxid = 0x100000020
cversion = 7
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 5
numChildren = 1
```
### API应用
#### 环境搭建
创建一个Maven工程
添加依赖
```xml
junit
junit
4.12
org.apache.zookeeper
zookeeper
3.4.10
```
拷贝log4j.properties文件到资源文件夹
```properties
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
```
#### 创建ZooKeeper客户端
```java
// zoo.cfg配置文件中的clientPort=2181
String connectString = "hadoop198:2181,hadoop197:2181,hadoop196:2181";
// 会话超时时间
int sessionTimeout = 2000;
// zk客户端
ZooKeeper zkClient;
@Before
public void init() throws IOException {
Watcher watcher = new Watcher() {
// 收到事件通知后的回调函数
public void process(WatchedEvent watchedEvent) {
}
};
// 获取zk客户端
zkClient = new ZooKeeper(connectString, sessionTimeout, watcher);
}
```
#### 创建子节点
```java
// 参数1: 要创建的节点的路径
// 参数2: 节点数据
// 参数3: 节点权限
// 参数4: 节点的类型
String path = zkClient.create("/roger",
"roger".getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
System.out.println(path);
```
#### 获取子节点并监听节点变化
在 zkClient 的回调函数 process 中加入
```java
Watcher watcher = new Watcher() {
// 收到事件通知后的回调函数
public void process(WatchedEvent watchedEvent) {
try {
// 参数1: 获取子节点的路径
// 参数2: 是否监听该节点的下一次变化
// 由于收到事件通知后就会执行该回调函数，故相当于每次都对下一次进行监听，实现持续监听
List children = zkClient.getChildren("/", true);
System.out.println("======================");
for (String child : children) {
System.out.println(child);
}
} catch (Exception e) {
e.printStackTrace();
}
}
};
```
此时在外触发一次回调函数并进行延时阻塞查看效果
```java
// 触发通知事件
zkClient.getChildren("/", false);
// 延时阻塞
Thread.sleep(Long.MAX_VALUE);
```
#### 判断Znode是否存在
```java
// 判断目标节点是否存在
Stat exists = zkClient.exists("/roger", false);
System.out.println(exists != null);
```
### 监听服务器节点动态上下线案例
**需求**
某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线
**需求分析**
**具体实现**
先在集群上创建/servers节点
```shell
[zk: localhost:2181(CONNECTED) 35] create /servers "servers"
```
服务器端向Zookeeper注册
```java
public class DistributeServer {
// zoo.cfg 配置文件中 clientPort=2181
String connectString = "hadoop198:2181,hadoop197:2181,hadoop196:2181";
// 会话超时时间
int sessionTimeout = 2000;
// zk客户端
ZooKeeper zkClient;
public static void main(String[] args) throws Exception {
DistributeServer server = new DistributeServer();
// 1 连接zookeeper集群
server.getConnection();
// 2 注册节点
server.register(args[0]);
// 3 逻辑处理
server.process();
}
// 逻辑处理
private void process(String hostname) throws InterruptedException {
System.out.println(hostname + " is working ...");
// 延时阻塞，避免销毁
Thread.sleep(Long.MAX_VALUE);
}
// 注册节点
private void register(String hostname) throws Exception {
zkClient.create("/servers/server",
hostname.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
// 获取连接
private void getConnection() throws Exception {
Watcher watcher = new Watcher() {
public void process(WatchedEvent watchedEvent) {
}
};
// 获取zk客户端
zkClient = new ZooKeeper(connectString, sessionTimeout, watcher);
}
}
```
客户端监控ZooKeeper集群
```java
public class DistributeClient {
// zoo.cfg 配置文件中 clientPort=2181
String connectString = "hadoop198:2181,hadoop197:2181,hadoop196:2181";
// 会话超时时间
int sessionTimeout = 2000;
// zk客户端
ZooKeeper zkClient;
public static void main(String[] args) throws Exception {
DistributeClient client = new DistributeClient();
// 1 获取zk集群连接
client.getConnection();
// 2 注册监听
client.getChildren();
// 3 逻辑处理
client.process();
}
// 逻辑处理
private void process() throws InterruptedException {
// 延时阻塞
Thread.sleep(Long.MAX_VALUE);
}
// 获取节点信息
private void getChildren() throws KeeperException, InterruptedException {
List children = zkClient.getChildren("/servers", true);
// 存储服务器节点主机
List hosts = new ArrayList();
for (String child : children) {
byte[] data = zkClient.getData("/servers/" + child, false, null);
hosts.add(new String(data));
}
// 打印在线主机
System.out.println(hosts);
}
// 获取集群连接
private void getConnection() throws IOException {
Watcher watcher = new Watcher() {
// 监听事件回调函数
public void process(WatchedEvent watchedEvent) {
// 注册监听，获取节点信息
try {
getChildren();
} catch (Exception e) {
e.printStackTrace();
}
}
};
// 获取zk客户端
zkClient = new ZooKeeper(connectString, sessionTimeout, watcher);
}
}
```