# Zookeeper

Zookeeper从设计模式角度来理解:是一个基 于观察者模式设计的分布式服务管理框架,它负责存储和管理大家都关心的数据,然后接受观察者的注册,一旦这些数据的状态发生变化,Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。

# 一、Zookeeper入门

# 1.1 Zookeeper特点

  1. Zookeeper:一个领导者(Leader),多个跟随者(Follower)组成的集群。
  2. 集群中只要有半数以上节点存活,Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。
  3. 全局数据一致:每个Server保存一份相同的数据副本,Client无论连接到哪个Server,数据都是一致的。 、
  4. 更新请求顺序执行,来自同一个Client的更新请求按其发送顺序依次执行。
  5. 数据更新原子性,一次数据更新要么成功,要么失败。
  6. 实时性,在一定时间范围内,Client能读到最新数据。

# 1.2 数据结构

ZooKeeper 数据模型的结构与 Unix 文件系统很类似,整体上可以看作是一棵树,每个节点称做一个ZNode。每一个 ZNode 默认能够存储 1MB 的数据,每个ZNode都可以通过其路径唯一标识。

# 1.3 应用场景

提供的服务包括:统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。

  1. 统一命名服务
  1. 统一配置管理
  1. 统一群集管理
  1. 服务器动态上下线
  1. 软负载均衡

# 二、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
1
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
1
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
1
2
3
4
5
6
7
8
9
10
11
12
13

# 2.5 配置参数解读

Zookeeper中的配置文件zoo.cfg中参数含义解读如下:

  1. tickTime = 2000:通信心跳时间,Zookeeper服务器与客户端心跳时间,单位毫秒
  2. initLimit = 10:LF初始通信时限,Leader和Follower初始连接时能容忍的最多心跳数(tickTime的数量)
  3. syncLimit = 5:LF同步通信时限 Leader和Follower之间通信时间如果超过syncLimit * tickTime,Leader认为Follwer死掉,从服务器列表中删除Follwer。
  4. dataDir:保存Zookeeper中的数据 注意:默认的tmp目录,容易被Linux系统定期删除,所以一般不用默认的tmp目录。
  5. 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
1
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
1
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,而这个端口就是用来执行选举时服务器相互通信的端口。
1
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
1
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
1

# 4.1 znode 节点数据信息

  1. 查看当前znode中所包含的内容
[zk: hdp2:2181(CONNECTED) 0] ls /
[zookeeper]
1
2
  1. 查看当前节点详细数据
[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 子节点数量
1
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):客户端和服务器端断开连接后,创建的节点自己删除

  1. 分别创建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]
1
2
3
4
5
6
7
8
  1. 获得节点的值
[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
1
2
3
4
5
6
7
8
9
10
11
12
13
  1. 创建带序号的节点(永久节点 + 带序号)
[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
1
2
3
4
  1. 创建短暂节点(短暂节点 + 不带序号 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
1
2
3
4
  1. 修改节点数据值
[zk: hdp2:2181(CONNECTED) 14] set /bihell "bihell"
[zk: hdp2:2181(CONNECTED) 15] get /bihell
bihell
1
2
3

# 五、监听器原理

监听原理

  1. 首先要有一个main()线程
  2. 在main线程中创建Zookeeper客户端,这时就会创建两个线 程,一个负责网络连接通信(connet),一个负责监听(listener)。
  3. 通过connect线程将注册的监听事件发送给Zookeeper。
  4. 在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中。
  5. Zookeeper监听到有数据或路径变化,就会将这个消息发送给listener线程。
  6. listener线程内部调用了process()方法。

常见的监听

  1. 监听节点数据的变化 get path [watch]
  2. 监听子节点增减的变化 ls path [watch]

# 5.1 节点的值变化监听

  1. 在 hdp1 主机上注册监听/bihell 节点数据变化
[zk: localhost:2181(CONNECTED) 0]  get -w /bihell
bihell
1
2
  1. 在 hdp3 主机上修改/bihell节点的数据
[zk: localhost:2181(CONNECTED) 0] set /bihell "test" 
1
  1. 观察 hdp1 主机收到数据变化的监听
WATCHER::

WatchedEvent state:SyncConnected type:NodeDataChanged path:/bihell 
1
2
3

在hdp3再多次修改/bihell的值,hdp1上不会再收到监听。因为注册 一次,只能监听一次。想再次监听,需要再次注册。

# 5.2 节点的子节点变化监听(路径变化)

  1. 在 hdp1 主机上注册监听/bihell 节点的子节点变化
[zk: localhost:2181(CONNECTED) 0] ls -w /bihell
[bigdata, hdfs0000000001, hdfs0000000002]
1
2
  1. 在 hdp3 主机/bihell 节点上创建子节点
[zk: localhost:2181(CONNECTED) 1] create /bihell/yarn "yarn" 
1
  1. 观察 hdp1 主机收到子节点变化的监听
WATCHER::

WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/bihell
Created /bihell/yarn
1
2
3
4

节点的路径变化,也是注册一次,生效一次。想多次生效,就需要多次注册。

# 5.3 节点删除与查看

  1. 删除节点
[zk: localhost:2181(CONNECTED) 2] delete /bihell/hdfs0000000002
1
  1. 查看节点状态
[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
1
2
3
4
5
6
7
8
9
10
11
12
  1. 递归删除节点
[zk: localhost:2181(CONNECTED) 5] deleteall /bihell
1

# 六、客户端 API 操作

# 6.1 IDEA 环境搭建

  1. 创建一个工程:zookeeper
  2. 添加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> 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
  1. 日志配置:需要在项目的 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 
1
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();
                }
            }
        });
    }
1
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);
    }
1
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);
    }
1
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");
    }
1
2
3
4
5
6
7

# 七、 服务器动态上下线监听案例

  1. 先在集群上创建/servers 节点
[zk: localhost:2181(CONNECTED) 11] create /servers "servers"
1
  1. 服务器端向 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) {

            }
        });
    }
}
1
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
  1. 客户端代码
 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();
                }
            }
        });
    }
}
1
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

更新时间: 1/27/2022, 6:04:00 PM