# Zookeeper
Zookeeper从设计模式角度来理解:是一个基 于观察者模式设计的分布式服务管理框架,它负责存储和管理大家都关心的数据
,然后接受观察者的注册
,一旦这些数据的状态发生变化,Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者
做出相应的反应。
# 一、Zookeeper入门
# 1.1 Zookeeper特点
- Zookeeper:一个领导者(Leader),多个跟随者(Follower)组成的集群。
- 集群中只要有
半数以上
节点存活,Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。 - 全局数据一致:每个Server保存一份相同的数据副本,Client无论连接到哪个Server,数据都是一致的。 、
- 更新请求顺序执行,来自同一个Client的更新请求按其发送顺序依次执行。
- 数据更新原子性,一次数据更新要么成功,要么失败。
- 实时性,在一定时间范围内,Client能读到最新数据。
# 1.2 数据结构
ZooKeeper 数据模型的结构与 Unix 文件系统很类似,整体上可以看作是一棵树,每个节点称做一个ZNode。每一个 ZNode 默认能够存储 1MB 的数据,每个ZNode都可以通过其路径唯一标识。
# 1.3 应用场景
提供的服务包括:统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。
- 统一命名服务

- 统一配置管理

- 统一群集管理

- 服务器动态上下线

- 软负载均衡

# 二、Zookeeper本地安装
# 2.1 安装JDK
# 2.2 解压到指定目录
[root@hdp1 opt]# tar xzf ~/apache-zookeeper-3.7.0-bin.tar.gz
[root@hdp1 opt]# mv apache-zookeeper-3.7.0-bin/ zookeeper-3.7.0
2
# 2.3 配置
# 样例配置改名
[root@hdp1 conf]# mv zoo_sample.cfg zoo.cfg
# 创建数据存储目录
[root@hdp1 zookeeper-3.7.0]# mkdir zkData
# 修改数据存储目录
[root@hdp1 conf]# vim zoo.cfg
dataDir=/opt/zookeeper-3.7.0/zkData
2
3
4
5
6
7
# 2.4 启动
# 启动服务器
[tpxcer@hdp1 zookeeper-3.7.0]$ bin/zkServer.sh start
[tpxcer@hdp1 zookeeper-3.7.0]$ jps -l
36724 sun.tools.jps.Jps
36421 org.apache.zookeeper.server.quorum.QuorumPeerMain
# 启动客户端
[tpxcer@hdp1 zookeeper-3.7.0]$ bin/zkCli.sh
# 查看状态
[tpxcer@hdp1 zookeeper-3.7.0]$ bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.7.0/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: standalone
2
3
4
5
6
7
8
9
10
11
12
13
# 2.5 配置参数解读
Zookeeper中的配置文件zoo.cfg中参数含义解读如下:
- tickTime = 2000:通信心跳时间,Zookeeper服务器与客户端心跳时间,单位毫秒
- initLimit = 10:LF初始通信时限,Leader和Follower
初始连接
时能容忍的最多心跳数(tickTime的数量) - syncLimit = 5:LF同步通信时限 Leader和Follower之间通信时间如果超过syncLimit * tickTime,Leader认为Follwer死掉,从服务器列表中删除Follwer。
- dataDir:保存Zookeeper中的数据 注意:默认的tmp目录,容易被Linux系统定期删除,所以一般不用默认的tmp目录。
- clientPort = 2181:客户端连接端口,通常不做修改。
# 三、Zookeeper群集安装
# 3.1 群集规划
系统:CentOS 7
本文规划
hdp1 | hdp2 | hdp3 |
---|---|---|
zk | zk | zk |
# 3.2 配置服务器编号
# 在/opt/zookeeper-3.7.0/zkData目录下创建一个 myid 的文件
[tpxcer@hdp1 zkData]$ vim myid
# 在文件中添加与 server 对应的编号(注意:上下不要有空行,左右不要有空格)
[tpxcer@hdp1 zkData]$ cat myid
0
2
3
4
5
拷贝配置好的 zookeeper 到其他机器上并分别修改 myid 文件中的编号
# 3.3 配置zoo.cfg文件
编辑zoo.cfg文件添加如下配置
[tpxcer@hdp1 zookeeper-3.7.0]$ vim conf/zoo.cfg
#######################cluster##########################
server.0=hdp1:2888:3888
server.1=hdp2:2888:3888
server.2=hdp3:2888:3888
2
3
4
5
6
配置参数解读server.A=B:C:D
A 是一个数字,表示这个是第几号服务器;
集群模式下配置一个文件 myid,这个文件在 dataDir 目录下,这个文件里面有一个数据 就是 A 的值,Zookeeper 启动时读取此文件,拿到里面的数据与 zoo.cfg 里面的配置信息比 较从而判断到底是哪个 server。
B 是这个服务器的地址;
C 是这个服务器 Follower 与集群中的 Leader 服务器交换信息的端口;
D 是万一集群中的 Leader 服务器挂了,需要一个端口来重新进行选举,选出一个新的Leader,而这个端口就是用来执行选举时服务器相互通信的端口。
2
3
4
5
# 3.4 ZK 集群启动停止脚本
[root@hdp1 script]# cat zk.sh
#!/bin/bash
if [ $# -lt 1 ]
then
echo "No Args Input..."
exit ;
fi
case $1 in
"start"){
for i in hdp1 hdp2 hdp3
do
echo ---------- zookeeper $i 启动 ------------
ssh $i "/opt/zookeeper-3.7.0/bin/zkServer.sh start"
done
};;
"stop"){
for i in hdp1 hdp2 hdp3
do
echo ---------- zookeeper $i 停止 ------------
ssh $i "/opt/zookeeper-3.7.0/bin/zkServer.sh stop"
done
};;
"status"){
for i in hdp1 hdp2 hdp3
do
echo ---------- zookeeper $i 状态 ------------
ssh $i "/opt/zookeeper-3.7.0/bin/zkServer.sh status"
done
};;
esac
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# 四、客户端命令行操作
客户端指定服务器
[tpxcer@hdp1 zookeeper-3.7.0]$ bin/zkCli.sh -server hdp2:2181
# 4.1 znode 节点数据信息
- 查看当前znode中所包含的内容
[zk: hdp2:2181(CONNECTED) 0] ls /
[zookeeper]
2
- 查看当前节点详细数据
[zk: hdp2:2181(CONNECTED) 1] ls -s /
[zookeeper]
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1
(1)czxid:创建节点的事务 zxid
每次修改ZooKeeper状态都会产生一个ZooKeeper事务ID。事务ID是ZooKeeper中所 有修改总的次序。每次修改都有唯一的 zxid,如果 zxid1 小于 zxid2,那么 zxid1 在 zxid2 之 前发生。
(2)ctime:znode 被创建的毫秒数(从 1970 年开始)
(3)mzxid:znode 最后更新的事务 zxid
(4)mtime:znode 最后修改的毫秒数(从 1970 年开始)
(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 子节点数量
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# 4.2 节点类型(持久/短暂/有序号/无序号)
持久(Persistent):客户端和服务器端断开连接后,创建的节点不删除 短暂(Ephemeral):客户端和服务器端断开连接后,创建的节点自己删除
- 分别创建2个普通节点(永久节点 + 不带序号)
[zk: hdp2:2181(CONNECTED) 2] create /bihell "zookeeper"
Created /bihell
[zk: hdp2:2181(CONNECTED) 3] ls /
[bihell, zookeeper]
[zk: hdp2:2181(CONNECTED) 4] create /bihell/bigdata "bigdata"
Created /bihell/bigdata
[zk: hdp2:2181(CONNECTED) 5] ls /bihell
[bigdata]
2
3
4
5
6
7
8
- 获得节点的值
[zk: hdp2:2181(CONNECTED) 6] get -s /bihell
zookeeper
cZxid = 0x300000002
ctime = Mon Sep 13 16:53:16 CST 2021
mZxid = 0x300000002
mtime = Mon Sep 13 16:53:16 CST 2021
pZxid = 0x300000003
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 9
numChildren = 1
2
3
4
5
6
7
8
9
10
11
12
13
- 创建带序号的节点(永久节点 + 带序号)
[zk: hdp2:2181(CONNECTED) 10] create -s /bihell/hdfs "hdfs"
Created /bihell/hdfs0000000001
[zk: hdp2:2181(CONNECTED) 11] create -s /bihell/hdfs "hdfs"
Created /bihell/hdfs0000000002
2
3
4
- 创建短暂节点(短暂节点 + 不带序号 or 带序号)
[zk: hdp2:2181(CONNECTED) 12] create -e /bihell/hadoop "hadoop"
Created /bihell/hadoop
[zk: hdp2:2181(CONNECTED) 13] create -e -s /bihell/hadoop "hadoop"
Created /bihell/hadoop0000000004
2
3
4
- 修改节点数据值
[zk: hdp2:2181(CONNECTED) 14] set /bihell "bihell"
[zk: hdp2:2181(CONNECTED) 15] get /bihell
bihell
2
3
# 五、监听器原理
监听原理
- 首先要有一个main()线程
- 在main线程中创建Zookeeper客户端,这时就会创建两个线 程,一个负责网络连接通信(connet),一个负责监听(listener)。
- 通过connect线程将注册的监听事件发送给Zookeeper。
- 在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中。
- Zookeeper监听到有数据或路径变化,就会将这个消息发送给listener线程。
- listener线程内部调用了process()方法。
常见的监听
- 监听节点数据的变化 get path [watch]
- 监听子节点增减的变化 ls path [watch]
# 5.1 节点的值变化监听
- 在 hdp1 主机上注册监听
/bihell
节点数据变化
[zk: localhost:2181(CONNECTED) 0] get -w /bihell
bihell
2
- 在 hdp3 主机上修改
/bihell
节点的数据
[zk: localhost:2181(CONNECTED) 0] set /bihell "test"
- 观察 hdp1 主机收到数据变化的监听
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/bihell
2
3
在hdp3再多次修改/bihell的值,hdp1上不会再收到监听。因为注册 一次,只能监听一次。想再次监听,需要再次注册。
# 5.2 节点的子节点变化监听(路径变化)
- 在 hdp1 主机上注册监听/bihell 节点的子节点变化
[zk: localhost:2181(CONNECTED) 0] ls -w /bihell
[bigdata, hdfs0000000001, hdfs0000000002]
2
- 在 hdp3 主机/bihell 节点上创建子节点
[zk: localhost:2181(CONNECTED) 1] create /bihell/yarn "yarn"
- 观察 hdp1 主机收到子节点变化的监听
WATCHER::
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/bihell
Created /bihell/yarn
2
3
4
节点的路径变化,也是注册一次,生效一次。想多次生效,就需要多次注册。
# 5.3 节点删除与查看
- 删除节点
[zk: localhost:2181(CONNECTED) 2] delete /bihell/hdfs0000000002
- 查看节点状态
[zk: localhost:2181(CONNECTED) 4] stat /bihell
cZxid = 0x300000002
ctime = Mon Sep 13 16:53:16 CST 2021
mZxid = 0x30000000c
mtime = Mon Sep 13 17:30:34 CST 2021
pZxid = 0x300000010
cversion = 9
dataVersion = 2
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 3
2
3
4
5
6
7
8
9
10
11
12
- 递归删除节点
[zk: localhost:2181(CONNECTED) 5] deleteall /bihell
# 六、客户端 API 操作
# 6.1 IDEA 环境搭建
- 创建一个工程:zookeeper
- 添加POM依赖
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8.2</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.7.0</version>
</dependency>
</dependencies>
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
- 日志配置:需要在项目的 src/main/resources 目录下,新建一个文件,命名为“log4j.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
2
3
4
5
6
7
8
# 6.2 创建 ZooKeeper 客户端
private String connectString = "hdp1:2181,hdp2:2181,hdp3:2181";
private int sessionTimeout = 2000;
private ZooKeeper zkClient;
@Before
public void init() throws IOException {
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
System.out.println("-------------------------------");
List<String> children = null;
try {
children = zkClient.getChildren("/", true);
for (String child : children) {
System.out.println(child);
}
System.out.println("-------------------------------");
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 6.3 创建子节点
@Test
public void create() throws KeeperException, InterruptedException {
String nodeCreated = zkClient.create("/bihell", "create_bihell".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
2
3
4
# 6.4 获取子节点并监听节点变化
@Test
public void getChildren() throws KeeperException, InterruptedException {
List<String> children = zkClient.getChildren("/", true);
for (String child : children) {
System.out.println(child);
}
// 延时
Thread.sleep(Long.MAX_VALUE);
}
2
3
4
5
6
7
8
9
10
11
# 6.5 判断 Znode 是否存在
@Test
public void exist() throws KeeperException, InterruptedException {
Stat stat = zkClient.exists("/bihell", false);
System.out.println(stat==null? "not exist " : "exist");
}
2
3
4
5
6
7
# 七、 服务器动态上下线监听案例
- 先在集群上创建/servers 节点
[zk: localhost:2181(CONNECTED) 11] create /servers "servers"
- 服务器端向 Zookeeper 注册代码
public class DistributeServer {
private String connectString = "hdp1:2181,hdp2:2181,hdp3:2181";
private int sessionTimeout = 2000;
private ZooKeeper zk;
public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
DistributeServer server = new DistributeServer();
// 1 获取zk连接
server.getConnect();
// 2 注册服务器到zk集群
server.regist(args[0]);
// 3 启动业务逻辑(睡觉)
server.business();
}
private void business() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
private void regist(String hostname) throws KeeperException, InterruptedException {
String create = zk.create("/servers/"+hostname, hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(hostname +" is online") ;
}
private void getConnect() throws IOException {
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
}
});
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
- 客户端代码
public class DistributeClient {
private String connectString = "hdp1:2181,hdp2:2181,hdp3:2181";
private int sessionTimeout = 2000;
private ZooKeeper zk;
public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
DistributeClient client = new DistributeClient();
// 1 获取zk连接
client.getConnect();
// 2 监听/servers下面子节点的增加和删除
client.getServerList();
// 3 业务逻辑(睡觉)
client.business();
}
private void business() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
private void getServerList() throws KeeperException, InterruptedException {
List<String> children = zk.getChildren("/servers", true);
ArrayList<String> servers = new ArrayList<>();
for (String child : children) {
byte[] data = zk.getData("/servers/" + child, false, null);
servers.add(new String(data));
}
// 打印
System.out.println(servers);
}
private void getConnect() throws IOException {
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
try {
getServerList();
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
# 八、分布式锁案例
什么叫做分布式锁呢? 比如说"进程 1"在使用该资源的时候,会先去获得锁,"进程 1"获得锁以后会对该资源保持独占,这样其他进程就无法访问该资源,"进程 1"用完该资源以后就将锁释放掉,让其 他进程来获得锁,那么通过这个锁机制,我们就能保证了分布式系统中多个进程能够有序的 访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。
...to do
# 九、Curator 框架实现分布式锁案例
...to do