ZooKeeper实战篇之zk集群搭建、zkCli.sh操作、权限控制ACL、ZooKeeper JavaAPI使用

在看了史上最全的Zookeeper原理详解(万字长文)，了解Zookeeper的原理后，你是不是蠢蠢欲动想着手实践呢？这篇文章将手把手教你在Linux上搭建ZooKeeper集群，并调用相关API实现自己的Zookeeper应用。

文章目录

1. Linux上搭建ZooKeeper集群
- 1.1 多台服务器之间免密登录
- 1.2 ZooKeeper集群搭建
2. zkCli.sh客户端操作
- 2.1 打开客户端
- 2.2 创建节点
- 2.3 读取节点
- 2.4 更新节点
- 2.5 删除节点
3. Zookeeper 权限控制 ACL
- 3.1 world 实例
- 3.2 auth 实例
- 3.3 digest 实例
- 3.4 IP 实例
- 3.5 super用户
4. Zookeeper JAVA API的使用
- 4.1 maven坐标
- 4.2 log4j配置
- 4.3 连接Zookeeper
- 4.4 新增节点
- 4.5 查看节点
- 4.6 修改节点
- 4.7 watcher监听

1. Linux上搭建ZooKeeper集群

在这之前，欢迎参考我之前的文章对虚拟机进行相关配置：Linux切换运行级别、关闭防火墙、禁用selinux、关闭sshd、时间同步、修改时区、拍摄快照、克隆操作，否则后面可能会出现意想不到的错误。

1.1 多台服务器之间免密登录

为什么要实现多台服务器之间免密登录？

因为zookeeper之间选举也好、投票也好，互相之间都会传递消息进行通信的，为了方便未来的管理，我们要实现多台服务器之间免密登陆。

现在我们就实现在Linux上搭建ZooKeeper集群吧，下面先介绍授权的两个文件：

id_dsa.pub 存放每台服务器自己的公钥
authorized_keys 存放的也是服务器的公钥，不过除了自己的公钥外，也可以存放其它服务器的公钥。

下面我准备了四台虚拟机，主机名分别为layne1、layne2、layne3、layne4，实现四台服务器之间免密登录。

首先在每个服务器上产生自己的公钥，在每台服务器上执行以下命名，产生的公钥文件id_dsa.pub存放在/root/.ssh下：
```
ssh-keygen  -t  dsa  -P  ''  -f  ~/.ssh/id_dsa
```

在layne1上将其公钥写入到authorized_keys中

cat  ~/.ssh/id_dsa.pub  >>  ~/.ssh/authorized_keys

将layne1上的authorized_keys文件拷贝给layne2，在layne1上执行如下命令即可
```
scp  ~/.ssh/authorized_keys   layne2:/root/.ssh/
```

在layne2上将其公钥追加到authorized_keys中

cat  ~/.ssh/id_dsa.pub  >>  ~/.ssh/authorized_keys

将layne2上的authorized_keys文件拷贝给layne3
```
scp  ~/.ssh/authorized_keys   layne3:/root/.ssh/
```

在layne3上将其公钥追加到authorized_keys中

cat  ~/.ssh/id_dsa.pub  >>  ~/.ssh/authorized_keys

将layne3上的authorized_keys文件拷贝给layne4
```
scp  ~/.ssh/authorized_keys   layne4:/root/.ssh/
```

在layne4上将其公钥追加到authorized_keys中

cat  ~/.ssh/id_dsa.pub  >>  ~/.ssh/authorized_keys

将layne4的authorized_keys文件分别拷贝给layne1、layne2、layne3

scp  ~/.ssh/authorized_keys   layne1:/root/.ssh/
scp  ~/.ssh/authorized_keys   layne2:/root/.ssh/
scp  ~/.ssh/authorized_keys   layne3:/root/.ssh/

至此，我们将完成了layne1~4虚拟机之间的免密登录。

在任意虚拟机的shell命令行里，我们就可以通过ssh 主机名随意连接其他的虚拟机，而不需要输入密码。比如，在layne1上连接layne4，只需要输入以下命令：

[root@layne1 ~]# ssh layne4 # 第一次连接需要输入yes|no，不需要输入密码
The authenticity of host 'layne4 (192.168.218.54)' can't be established.
RSA key fingerprint is 2d:4c:3c:0c:2a:0f:50:bc:a2:8d:c1:2f:8a:7d:63:c4.
Are you sure you want to continue connecting (yes/no)? yes
Warning: Permanently added 'layne4,192.168.218.54' (RSA) to the list of known hosts.
Last login: Tue Feb 23 10:29:52 2021 from 192.168.218.1
[root@layne4 ~]#

1.2 ZooKeeper集群搭建

在搭建ZooKeeper集群之前，先要在所有虚拟机上安装jdk，我之前的好多博客都详细描述了jdk的安装方法，这里就不介绍了，有需要的小伙伴，可参考Linux上通过rpm安装jdk。

我安装的jdk版本是jdk-8u221-linux-x64，下面我在主机名为layne2、layne3、layne4的虚拟机搭建ZooKeeper集群。

在https://zookeeper.apache.org/releases.html上下载zookeeper的Linux压缩包。
将zookeeper的压缩包zookeeper-3.4.6.tar.gz上传到layne2上。

解压至/opt目录下

tar -zxvf zookeeper-3.4.6.tar.gz -C /opt

配置zookeeper的环境变量，执行vim /etc/profile，在末尾加入：
```
export ZOOKEEPER_HOME=/opt/zookeeper-3.4.6
export PATH=$PATH:$ZOOKEEPER_HOME/bin
```
然后执行source /etc/profile，让配置生效。

进入zookeeper的安装目录的conf下

[root@layne2 apps]# cd $ZOOKEEPER_HOME/conf
[root@layne2 conf]# pwd
/opt/zookeeper-3.4.6/conf

复制zoo_sample.cfg文件为zoo.cfg
```
cp zoo_sample.cfg zoo.cfg
```
先介绍zoo.cfg参数说明，然后再进行配置。
- tickTime=2000 ：客户端与服务器或者服务器与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一次心跳，默认心跳时间为2000ms。通过心跳不仅能够用来监听机器的工作状态，还可以通过心跳来控制Flower跟Leader的通信时间。zookeeper的客户端和服务端之间也有和web开发里类似的session的概念，而zookeeper里最小的session过期时间通常是tickTime的两倍。
- dataDir=/tmp/zookeeper：用于保存 Zookeeper 中的数据，同时用于zookeeper集群的myid文件也存在这个文件夹里。默认路径为/tmp/zookeeper，最好不要使用/tmp，因为临时目录下，操作系统会定时清理里面的文件，可能会造成出乎意料的错误。
- dataLogDir：存放日志的目录。
- clientPort=2181：客户端连接zookeeper服务器的端口，zookeeper会监听这个端口，接收客户端的请求访问，这个端口默认是2181。
- initLimit：集群中的follower服务器(F)与leader服务器(L)之间初始化连接时最长能忍受多少个心跳时间间隔数。如果配置的是5，当已经超过 5 个心跳的时间（也就是 tickTime）长度后，ZooKeeper 服务器还没有收到客户端（即follower服务器，相对于 leader 而言的客户端）的返回信息，那么表明这个客户端连接失败，此时总的时间长度就是 5*2000=10秒。如果在设定的时间段内，半数以上的跟随者未能完成同步（即初始时的选举），领导者便会宣布放弃领导地位，进行另一次的领导选举。如果zk集群环境数量确实很大，同步数据的时间会变长，因此这种情况下可以适当调大该参数。
- syncLimit：标识 Leader 与 Follower 之间请求和应答能容忍的最多心跳数，如果配置的是4，总的时间长度就是 4*2000=8 秒。如果 follower 在设置的时间内不能与leader 进行通信，那么此 follower 将被丢弃，此时所有关联到这个跟随者的客户端将连接到另外一个跟随着。
- server.A=B：C：D：其中 A 是一个数字，表示这个是第几号服务器（和myid对应）；B 是这个服务器的ip地址（或主机名）；C 表示的是这个服务器与集群中的Leader服务器交换信息的端口（即follower与Leader交换信息的端口）；D表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口（选举时的端口）。如果是伪集群的配置方式，由于B都是一样的，所以不同的ZooKeeper实例通信端口号不能一样，要给C和D分配不同的端口号。

根据上面的参数说明，对zoo.cfg进行配置，配置如下：

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=5
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=2
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/opt/zookeeper-3.4.6/data
# zk log dir
dataLogDir=/var/log/zookeeper/datalog
# the port at which the clients will connect
clientPort=2181
# server list
server.1=layne2:2881:3881
server.2=layne3:2881:3881
server.3=layne4:2881:3881

创建/var/log/zookeeper/datalog和/opt/zookeeper-3.4.6/data目录

[root@layne2 conf]# mkdir -p /var/log/zookeeper/datalog
[root@layne2 conf]# mkdir /opt/zookeeper-3.4.6/data

在/opt/zookeeper-3.4.6/data目录下创建一个名为myid的文件，在myid中写下当前ZooKeeper的编号

[root@layne2 data]# pwd
/opt/zookeeper-3.4.6/data
[root@layne2 data]# touch myid
[root@layne2 data]# echo 1 > /opt/zookeeper-3.4.6/data/myid

将配置好Zookeeper拷贝到layne3、layne4上

scp -r /opt/zookeeper-3.4.6/ layne3:/opt/
scp -r /opt/zookeeper-3.4.6/ layne4:/opt/

在layne3和layne4上分别修改myid

echo 2 > /opt/zookeeper-3.4.6/data/myid
echo 3 > /opt/zookeeper-3.4.6/data/myid

在layne3和layne4配置Zookeeper的环境变量，并创建/var/log/zookeeper/datalog目录，参考步骤3和步骤8。

分别启动layne2、layne3、layne4上的ZooKeeper

# 进入/opt/zookeeper-3.4.6/bin目录下操作
zkServer.sh start #启动zk
zkServer.sh stop  #停止zk
zkServer.sh status  #查看zk状态
zkCli.sh # 连接ZooKeeper客户端

启动3台虚拟机上的ZooKeeper之后，如果报错Will not attempt to authenticate using SASL，报错原因是我们在实现多台服务器之间免密登陆的时候，两台服务器之间一次都没有进行连接。只要第一次连接之后，两台服务器之间才能免密登陆和授权。在layne2上执行下面三条命令，即可和layne3、layne4免密登陆。同理，layne3、layne4也如此。
```
# 在layne2、layne3、layne4全部都要执行下述命令
ssh layne2 #自己也要和自己连接一次，即自己授权自己
exit
ssh layne3
exit
ssh layne4
exit
```
再次尝试启动ZooKeeper，如果还报错，重启所有的虚拟机就好了。

以上步骤就是搭建Zookeeper集群的完整过程，如果Zookeeper启动不了，或者是启动报错，可能是以下原因造成的：

没有创建zookeeper的日志目录/var/log/zookeeper/datalog。
没有在每个服务器的myid写入正确的编号。
没有用ssh 主机名在任意两台服务器之间进行第一次连接
如果不是上面3个原因，重启一下所有的虚拟机就好了。

2. zkCli.sh客户端操作

zkCli是 Zookeeper的一个简易客户端，下面讲解通过zkCli.sh客户端操作znode节点。

2.1 打开客户端

在Zookeeper服务端开启的情况下，运行客户端，使用命令：zkCli.sh

若连接不同的主机，可使用命令：zkCli.sh -server ip:port，如zkCli.sh -server 192.168.218.52:2181，此处的端口是zoo.cfg配置文件中clientPort。

连接客户端以后，可以使用help命令来查看客户端的操作

[zk: localhost:2181(CONNECTED) 1] help
ZooKeeper -server host:port cmd argsstat path [watch]set path data [version]ls path [watch]delquota [-n|-b] pathls2 path [watch]setAcl path aclsetquota -n|-b val pathhistory redo cmdnoprintwatches on|offdelete path [version]sync pathlistquota pathrmr pathget path [watch]create [-s] [-e] path data acladdauth scheme authquit getAcl pathclose connect host:port

2.2 创建节点

使用create命令，可以创建一个Zookeeper节点，格式为：

create [-s] [-e] path data acl

[-s] [-e]：-s 和 -e 都是可选的，-s 代表顺序节点， -e 代表临时节点，注意其中 -s 和 -e 可以同时使用，都不使用，则代表普通节点。需要注意的是，临时节点不能再创建子节点。
path：指定要创建节点的路径，比如 /zk01。
data：要在此节点存储的数据。
acl：访问权限相关，默认是 world，相当于全世界都能访问，请看后面第3节Zookeeper权限控制ACL。

①创建永久顺序节点

[zk: 192.168.218.52:2181(CONNECTED) 5] create -s /zk01-seq 123 # 创建顺序节点
Created /zl01-seq0000000009

可以看到创建的zk01-seq节点后面添加了一串数字以示区别。

②创建临时顺序节点

[zk: 192.168.218.52:2181(CONNECTED) 1] create -s -e /zk01-tmp-seq 1234
Created /zk01-tmp-seq0000000013

③创建普通临时节点

[zk: 192.168.218.52:2181(CONNECTED) 10] create -e /zk01-tmp 456
Created /zk01-tmp

临时节点在客户端会话结束后，就会自动删除，下面使用quit命令退出客户端

[zk: 192.168.218.52:2181(CONNECTED) 11] quit
Quitting...
2021-03-01 21:06:58,605 [myid:] - INFO  [main:ZooKeeper@684] - Session: 0x177ed505b850002 closed

再次使用客户端连接服务端，并使用ls / 命令查看根目录下的节点

[zk: 192.168.218.52:2181(CONNECTED) 0] ls /
[zookeeper, zk01-seq0000000011]

可以看到根目录下已经不存在zk01-tmp临时节点了。

④创建普通永久节点

[zk: 192.168.218.52:2181(CONNECTED) 2] create /zk01-permanent 123
Created /zk01-permanent

可以看到普通节点不同于顺序节点，不会自动在后面添加一串数字。

2.3 读取节点

与读取相关的命令ls、ls2、get和stat命令。

①ls命令

ls 命令用于查看某个路径下的znode节点（只能查看第一级目录的所有子节点），格式为ls path，例如：

ls /
ls /zookeeper

[zk: 192.168.218.52:2181(CONNECTED) 3] ls / # 查看根目录下的znode节点
[zookeeper, zk01-seq0000000011, zk01-tmp-seq0000000013, zk01-permanent]
[zk: 192.168.218.52:2181(CONNECTED) 4] ls /zk01-permanent # 查看zk01-permanent目录下的znode节点
[]

②ls2命令

ls2 命令也是用于查看某个路径下的znode节点，格式同ls，但它能同时显示该路径节点的信息。

[zk: 192.168.218.52:2181(CONNECTED) 5] ls2 / #查看根目录下的znode节点，并显示根节点的信息
[zookeeper, zk01-seq0000000011, zk01-tmp-seq0000000013, zk01-permanent]
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x600000022
cversion = 26
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 4

③get 命令

get 命令用于获取某个znode节点数据和状态信息。其格式为：

get path [watch]

path：代表路径
[watch]：对该节点进行事件监听，该参数为可选参数。

以下示例我们同时开启两个终端，对zk01节点进行监听：

[zk: 192.168.218.52:2181(CONNECTED) 9] create /zk01 zk01Content # 创建zk01普通持久节点
Created /zk01
[zk: 192.168.218.52:2181(CONNECTED) 10] get /zk01 watch #在当前终端上对zk01事件监听
zk01Content
cZxid = 0x600000023
ctime = Mon Mar 01 21:20:26 CST 2021
mZxid = 0x600000023
mtime = Mon Mar 01 21:20:26 CST 2021
pZxid = 0x600000023
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 11
numChildren = 0
[zk: localhost:2181(CONNECTED) 0] set /zk01 zk01ABC #在终端二上修改zk01节点的数据
cZxid = 0x600000023
ctime = Mon Mar 01 21:20:26 CST 2021
mZxid = 0x600000024
mtime = Mon Mar 01 21:24:07 CST 2021
pZxid = 0x600000023
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 7
numChildren = 0

此时，会在第一个终端上会输出NodeDataChanged 事件：

[zk: 192.168.218.52:2181(CONNECTED) 11]
WATCHER::WatchedEvent state:SyncConnected type:NodeDataChanged path:/zk01

④stat 命令

stat 命令用于查看某个节点状态信息。该命令除了不输出节点的内容之后，输出的其他信息和get命令一致。

其格式为：

get path [watch]

path：代表路径
[watch]：对该节点进行事件监听，该参数为可选参数。

[zk: 192.168.218.52:2181(CONNECTED) 12] stat /zk01 #查看zk01节点的信息
cZxid = 0x600000023
ctime = Mon Mar 01 21:20:26 CST 2021
mZxid = 0x600000024
mtime = Mon Mar 01 21:24:07 CST 2021
pZxid = 0x600000023
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 7
numChildren = 0

2.4 更新节点

使用set命令，可以更新指定节点的数据内容，其格式：

set path data [version]

path：节点路径。
data：需要存储的数据。
[version]：可选项，版本号(可用作乐观锁)。

[zk: 192.168.218.52:2181(CONNECTED) 13] get /zk01
zk01ABC
cZxid = 0x600000023
ctime = Mon Mar 01 21:20:26 CST 2021
mZxid = 0x600000024
mtime = Mon Mar 01 21:24:07 CST 2021
pZxid = 0x600000023
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 7
numChildren = 0

可以看到，zk01节点dataVersion为1，下面只有正确的版本号才能设置成功：

[zk: 192.168.218.52:2181(CONNECTED) 14] set /zk01 123 0 #带上版本号，只有正确的版本才能执行
version No is not valid : /zk01
[zk: 192.168.218.52:2181(CONNECTED) 15] set /zk01 456 1
cZxid = 0x600000023
ctime = Mon Mar 01 21:20:26 CST 2021
mZxid = 0x600000026
mtime = Mon Mar 01 21:29:11 CST 2021
pZxid = 0x600000023
cversion = 0
dataVersion = 2
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0
[zk: 192.168.218.52:2181(CONNECTED) 16] set /zk01 789 3
version No is not valid : /zk01
[zk: 192.168.218.52:2181(CONNECTED) 17] set /zk01 555  #不加版本号，均可以执行成功
cZxid = 0x600000023
ctime = Mon Mar 01 21:20:26 CST 2021
mZxid = 0x600000028
mtime = Mon Mar 01 21:30:30 CST 2021
pZxid = 0x600000023
cversion = 0
dataVersion = 3
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0

2.5 删除节点

delete 命令用于删除某节点。格式为：

delete path [version]

path：节点路径。
[version]：可选项，版本号（同 set 命令）。

[zk: 192.168.218.52:2181(CONNECTED) 18] delete /zk01-permanent #删除zk01-permanent节点

若删除节点存在子节点，那么无法删除该节点，必须先删除子节点，再删除父节点。

3. Zookeeper 权限控制 ACL

Zookeeper 的 ACL（Access Control List，访问控制表）权限在生产环境是特别重要的，ACL 权限可以针对节点设置相关读写等权限，保障数据安全性。我们以zkCli.sh客户端为例，来说明zookeeper对ACL的设置。

ACL通过[scheme:id:permissions] 来构成权限列表。

scheme：代表采用的某种权限机制，包括 world、auth、digest、ip、super 几种。
id：代表允许访问的用户。
permissions：权限组合字符串，由 cdrwa 组成，其中每个字母代表支持不同权限，创建权限 create©、删除权限 delete(d)、读权限 read®、写权限 write(w)、管理权限admin(a)。

需要注意的是，zookeeper对权限的控制是znode级别的，不具有继承性，即子节点不继承父节点的权限。

ACL 命令有三个，分别是：

getAcl 命令：获取某个节点的 acl 权限信息。
setAcl 命令：设置某个节点的 acl 权限信息。
addauth 命令：输入认证授权信息，注册时输入明文密码，加密形式保存。

3.1 world 实例

这是默认方式，代表开放式权限。当创建一个新的节点(znode)，而又没有设置任何权限时，就是这个值，例如：

[zk: localhost:2181(CONNECTED) 51] create /node mynode
Created /node
[zk: localhost:2181(CONNECTED) 52] getAcl /node
'world,'anyone
: cdrwa

可以看到，/node节点的ACL属于是world schema的，因为它没有设置ACL属性，这样任何人都可以访问这个节点。

设置某一节点的权限命令为setAcl，语法格式为：

setAcl <path> scheme:<id>:<acl>

setAcl命令中的id域是可忽略的，可以填任意值，或者空串，例如：setAcl <path> auth::crdwa。如果这个域是忽略的，会把所有已经授权的认证用户都加进来。

如果要手工设置world schema，那么此时的id域只允许一个值，即anyone，格式如下：

setAcl /node world:anyone:crdwa

下面，设置/node节点 permissions 权限部分为 crwa

[zk: localhost:2181(CONNECTED) 53] setAcl /node world:anyone:crwa
cZxid = 0x60000002a
ctime = Mon Mar 01 21:58:20 CST 2021
mZxid = 0x60000002a
mtime = Mon Mar 01 21:58:20 CST 2021
pZxid = 0x60000002a
cversion = 0
dataVersion = 0
aclVersion = 1
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: localhost:2181(CONNECTED) 54] getAcl /node #再次查看权限，发现permissions部分已经改变。
'world,'anyone
: crwa

3.2 auth 实例

auth 用于给用户授予权限，授权之前需要先创建用户。

语法格式为：

addauth digest <user>:<password>

下面，给lucy用户授权/node节点的权限：

[zk: localhost:2181(CONNECTED) 6] setAcl /node auth:lucy:123456:cdrwa
Acl is not valid : /node
[zk: localhost:2181(CONNECTED) 7] addauth digest user1:123456  #只有授权之后的用户才能设置权限
[zk: localhost:2181(CONNECTED) 8] setAcl /node auth:lucy:123456:cdrwa
cZxid = 0x600000032
ctime = Mon Mar 01 22:12:43 CST 2021
mZxid = 0x600000032
mtime = Mon Mar 01 22:12:43 CST 2021
pZxid = 0x600000032
cversion = 0
dataVersion = 0
aclVersion = 1
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: localhost:2181(CONNECTED) 9] getAcl /node #查看权限，发现scheme和id部分已经改变
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: cdrwa

再来看一个例子，会有奇怪的现象发生：

[zk: localhost:2181(CONNECTED) 10] quit  #退出
Quitting...
2021-03-01 22:31:30,855 [myid:] - INFO  [main:ZooKeeper@684] - Session: 0x177ed505b850004 closed
2021-03-01 22:31:30,856 [myid:] - INFO  [main-EventThread:ClientCnxn$EventThread@512] - EventThread shut down
[root@layne2 bin]# zkCli.sh  #连接客户端，重新生成session
[zk: localhost:2181(CONNECTED) 0] addauth digest user1:123456
[zk: localhost:2181(CONNECTED) 1] addauth digest user2:123456
[zk: localhost:2181(CONNECTED) 2] addauth digest user3:123456
[zk: localhost:2181(CONNECTED) 3] getAcl /node
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: cdrwa
[zk: localhost:2181(CONNECTED) 4] setAcl /node auth:user2:crdwa
cZxid = 0x600000032
ctime = Mon Mar 01 22:12:43 CST 2021
mZxid = 0x600000032
mtime = Mon Mar 01 22:12:43 CST 2021
pZxid = 0x600000032
cversion = 0
dataVersion = 0
aclVersion = 2
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: localhost:2181(CONNECTED) 5] getAcl /node
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: cdrwa
'digest,'user2:hZG2W+NR7DCvADzOkGR6JGLqoTY=
: cdrwa
'digest,'user3:SzpfOOuDCdri8p4n7oIaFCZpXeE=
: cdrwa

这个例子中，我们先添加了三个授权用户user1、user2、user3，然后通过setAcl设置ACL，命令中指定了id为user2，可以看到，最后通过getAcl查询出来的结果包含所有前面添加的三个认证用户。

下面做几点总结（重要）：

setAcl命令中的id值是无效的，当使用addauth命令授权多个用户后，再用setAcl设置ACL时，会把当前会话所有addauth的用户都被会加入到acl中。
通过addauth命令(addauth digest <username>:<password>)授权的用户只在当前会话(session)有效。
setAcl命令设置权限后是永久式的，即使当前会话退出也不会消失。
如果在当前会话中，用户没有通过addauth授权就用setAcl设置acl权限时会失败。
使用setAcl来设置acl权限后，经过addauth授权其它的用户，如果再使用setAcl设置权限，则会覆盖之前的acl权限信息，而且只会针对当前会话中的授权用户来设置acl权限。

所以这种授权方式更倾向于用作测试开发环境，而不是产品环境中。

3.3 digest 实例

这就是最普通的用户名:密码的验证方式，在一般业务系统中最常用。其语法格式如下：

setAcl <path> digest:<user>:<password(密文)>:<acl>

和auth实例相比，digest 实例的密码是经过sha1及base64处理的密文。

密码可以通过如下shell的方式生成：

echo -n <user>:<password> | openssl dgst -binary -sha1 | openssl base64

也可以通过zookeeper的库文件生成：

# 方式一：通过linux自带的命令工具生成密码的密文
[root@layne2 bin]# echo -n user1:123456 | openssl dgst -binary -sha1 | openssl base64
HYGa7IZRm2PUBFiFFu8xY2pPP/s=# 方式二：通过zookeeper的库文件生成密码的密文
[root@layne2 bin]# java -cp /opt/zookeeper-3.4.6/zookeeper-3.4.6.jar:/opt/zookeeper-3.4.6/lib/slf4j-api-1.6.1.jar \
>   org.apache.zookeeper.server.auth.DigestAuthenticationProvider \
>   user1:123456
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
user1:123456->user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=

把上面输出的HYGa7IZRm2PUBFiFFu8xY2pPP/s=传递给diges实例下setAcl使用的password域。

[zk: localhost:2181(CONNECTED) 10] quit #退出
Quitting...
2021-03-01 22:31:30,855 [myid:] - INFO  [main:ZooKeeper@684] - Session: 0x177ed505b850004 closed
2021-03-01 22:31:30,856 [myid:] - INFO  [main-EventThread:ClientCnxn$EventThread@512] - EventThread shut down
[root@layne2 bin]# zkCli.sh #连接客户端，重新生成session
[zk: localhost:2181(CONNECTED) 0] getAcl /node
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: cdrwa
'digest,'user2:hZG2W+NR7DCvADzOkGR6JGLqoTY=
: cdrwa
'digest,'user3:SzpfOOuDCdri8p4n7oIaFCZpXeE=
: cdrwa
[zk: localhost:2181(CONNECTED) 1] setAcl /node digest:user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=:rwdca
Authentication is not valid : /node   #digest scheme用户也必须经过授权
[zk: localhost:2181(CONNECTED) 2] addauth digest user1:123abc456 #用户的密码一定是生成密文的密码才行
[zk: localhost:2181(CONNECTED) 3] setAcl /node digest:user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=:rwdca
Authentication is not valid : /node
[zk: localhost:2181(CONNECTED) 4] addauth digest user1:123456 #这次是生成密文的密码
[zk: localhost:2181(CONNECTED) 5] setAcl /node digest:user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=:rwdca
cZxid = 0x600000032
ctime = Mon Mar 01 22:12:43 CST 2021
mZxid = 0x600000032
mtime = Mon Mar 01 22:12:43 CST 2021
pZxid = 0x600000032
cversion = 0
dataVersion = 0
aclVersion = 3
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: localhost:2181(CONNECTED) 6] getAcl /node
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: cdrwa
[zk: localhost:2181(CONNECTED) 7] addauth digest user2:123456
[zk: localhost:2181(CONNECTED) 8] setAcl /node digest:user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=:rwdca   #授权user2用户后，再次用setAcl为user1设置权限，看是否同时给user2设置权限
cZxid = 0x600000032
ctime = Mon Mar 01 22:12:43 CST 2021
mZxid = 0x600000032
mtime = Mon Mar 01 22:12:43 CST 2021
pZxid = 0x600000032
cversion = 0
dataVersion = 0
aclVersion = 4
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: localhost:2181(CONNECTED) 9] getAcl /node #可以看到，没有为user2设置权限
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: cdrwa
[zk: localhost:2181(CONNECTED) 10]

和auth比较，digest有如下特性：

授权是针对单个特定用户。
setAcl使用的密码不是明文，是sha1摘要值，无法反推出用户密码内容。

3.4 IP 实例

限制 IP 地址的访问权限，比如把权限设置给 IP 地址为 192.168.218.54 后，IP 为 192.168.218.52 已经没有访问权限。

IP地址也可以为主机名。主机名可以是单个主机名，也可以是域名。IP可以是单个IP地址，也可以是IP地址段

[zk: localhost:2181(CONNECTED) 10] create /testnode tnode
Created /testnode
[zk: localhost:2181(CONNECTED) 11] getAcl /testnode
'world,'anyone
: cdrwa
[zk: localhost:2181(CONNECTED) 15] setAcl /testnode ip:192.168.218.54:cdrwa
cZxid = 0x60000003e
ctime = Mon Mar 01 23:18:00 CST 2021
mZxid = 0x60000003f
mtime = Mon Mar 01 23:18:55 CST 2021
pZxid = 0x60000003e
cversion = 0
dataVersion = 1
aclVersion = 1
ephemeralOwner = 0x0
dataLength = 23
numChildren = 0
[zk: localhost:2181(CONNECTED) 16] getAcl /testnode
'ip,'192.168.218.54
: cdrwa
[zk: localhost:2181(CONNECTED) 17] get /testnode
Authentication is not valid : /testnode

这时，通过192.168.218.54连接192.168.218.52中的zkCli.sh，就有访问权限：

[root@layne4 version-2]# zkCli.sh -server 192.168.218.52:2181
[zk: 192.168.218.52:2181(CONNECTED) 0] getAcl /testnode
'ip,'192.168.218.54
: cdrwa
[zk: 192.168.218.52:2181(CONNECTED) 1] get /testnode
tnode
cZxid = 0x60000003e
ctime = Mon Mar 01 23:18:00 CST 2021
mZxid = 0x60000003f
mtime = Mon Mar 01 23:18:55 CST 2021
pZxid = 0x60000003e
cversion = 0
dataVersion = 1
aclVersion = 1
ephemeralOwner = 0x0
dataLength = 23
numChildren = 0

3.5 super用户

设置一个超级用户，这个超级用户的设置必须在zookeeper内部，在zookeeper启动之前设置好。在这种scheme情况下，超级用户具有超级权限，可以做任何事情(cdrwa)，不需要授权。

我们通过digest scheme方式只为user1设置d权限：

[zk: localhost:2181(CONNECTED) 23] setAcl /node digest:user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=:d
cZxid = 0x600000032
ctime = Mon Mar 01 22:12:43 CST 2021
mZxid = 0x600000032
mtime = Mon Mar 01 22:12:43 CST 2021
pZxid = 0x600000032
cversion = 0
dataVersion = 0
aclVersion = 5
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: localhost:2181(CONNECTED) 24] get /node
Authentication is not valid : /node
[zk: localhost:2181(CONNECTED) 25] getAcl /node
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: d
[zk: localhost:2181(CONNECTED) 26] addauth digest root:123456
[zk: localhost:2181(CONNECTED) 28] get /node
Authentication is not valid : /node

可以看到，usr1用户已经不能访问/node结点信息。同样的，root用户也不能。

现在，我们为root用户添加为super用户：

1、生成root用户的密文：

[root@layne2 bin]# echo -n root:123456 | openssl dgst -binary -sha1 | openssl base64
u53OoA8hprX59uwFsvQBS3QuI00=

2、设置zookeeper环境变量SERVER_JVMFLAGS

export SERVER_JVMFLAGS="-Dzookeeper.DigestAuthenticationProvider.superDigest=root:u53OoA8hprX59uwFsvQBS3QuI00="

3、重启zookeeper

4、连接zkCli.sh客户端

5、再次访问/node结点

[zk: localhost:2181(CONNECTED) 1] getAcl /node
'digest,'user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=
: d
[zk: localhost:2181(CONNECTED) 2] get /node
Authentication is not valid : /node
[zk: localhost:2181(CONNECTED) 3] addauth digest root:123456
[zk: localhost:2181(CONNECTED) 4] get /node
mynode
cZxid = 0x600000032
ctime = Mon Mar 01 22:12:43 CST 2021
mZxid = 0x600000032
mtime = Mon Mar 01 22:12:43 CST 2021
pZxid = 0x600000032
cversion = 0
dataVersion = 0
aclVersion = 5
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

可以看到，给root添加授权后，就能访问/node结点了，因为这时root在zookeeper集群里面被配置成了超级用户。

在第2步，直接在Linux的bash命令行输入设置zookeeper环境变量SERVER_JVMFLAGS只对当前有效，如果Linux系统重启，就会失效。可以将该命令写入/etc/profile文件里，保证重启电脑后也不会失效。

即在/etc/profile的最后一行加入下面的内容，并执行source /etc/profile让配置立即生效。

export SERVER_JVMFLAGS="-Dzookeeper.DigestAuthenticationProvider.superDigest=root:u53OoA8hprX59uwFsvQBS3QuI00="

还有另一种方法设置Spuer用户，可以参考ACL super 超级管理员，我没有尝试，应该也可行。

4. Zookeeper JAVA API的使用

4.1 maven坐标

  <dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.11</version></dependency><dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.4.6</version></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.16</version></dependency></dependencies>

4.2 log4j配置

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/zookeeperAPI.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

4.3 连接Zookeeper

// 用于等待 SyncConnected 事件触发后继续执行当前线程
private CountDownLatch countDownLatch = new CountDownLatch(1);
//session的实效时间
private static final int SESSION_TIMEOUT = 30000;
//创建Logger对象，按照文件log4j.properties中指定的格式输出日志到控制台
private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperApiDemo.class);
//zookeeper连接对象
private ZooKeeper zooKeeper;
private Watcher watcher = new Watcher() {@Overridepublic void process(WatchedEvent event) {if(Event.KeeperState.SyncConnected == event.getState()){countDownLatch.countDown();String msg=String.format("process info,eventType:%s,eventState:%s,eventPath:%s",event.getType(),event.getState(),event.getPath());LOGGER.info(msg);}}
};@Before
public void connect() throws IOException {zooKeeper = new ZooKeeper("192.168.218.52:2181,192.168.218.53:2181,192.168.218.54:2181",SESSION_TIMEOUT,watcher);try {countDownLatch.await();LOGGER.info("Zookeeper session establish success,sessionID="+Long.toHexString(zooKeeper.getSessionId()));} catch (InterruptedException e) {e.printStackTrace();LOGGER.debug("Zookeeper session establish fail");}
}@After
public void close(){if(zooKeeper!=null){try {zooKeeper.close();} catch (InterruptedException e) {e.printStackTrace();}}
}

ZooKeeper构造函数的参数：

connectionString ：zookeeper主机（注意端口2181）
sessionTimeout ：会话超时（以毫秒为单位)
watcher ：实现“监视器”对象，zookeeper集合通过监视器对象返回连接状态。

当new一个zookeeper对象后，zookeeper的连接过程可能会受到网络、zookeeper集群等各种问题的影响，连接的过程可能会比较慢。因此，为了提高程序的执行性能，可以在watcher监视器里面使用并发工具类CountDownLatch，这个工具类在初始化的时候指定一个int类型的值，通过调用countDown方法，这个值会减一，当减到0时，所有的await线程都会被叫醒。所以，每次在使用zookeeper之前，使用countDownLatch.await()来确保每次使用zookeeper对象之前，zookeeper客户端都能成功连接到集群。

4.4 新增节点

// 同步方式
create(String path, byte[] data, List<ACL> acl, CreateMode createMode)
// 异步方式
create(String path, byte[] data, List<ACL> acl, CreateMode createMode,
AsyncCallback.StringCallback callBack,Object ctx)

path：znode路径
data：要存储在指定znode路径中的数据
acl：要创建的节点的访问控制列表。 zookeeper API提供了一个静态接口ZooDefs.Ids来获取一些基本的acl列表
createMode：节点的类型，这是一个枚举类型
callBack：异步回调接口
ctx：传递上下文参数

下面创建临时结点/zk001

    /**创建节点：*CreateMode：* PERSISTENT 普通持久节点* PERSISTENT_SEQUENTIAL:顺序持久节点* EPHEMERAL ：普通临时* EPHEMERAL_SEQUENTIAL：顺序临时节点* Access Control List: 访问控制列表* OPEN_ACL_UNSAFE: ANYONE CAN VISIT*/@Testpublic void createNode(){String result = null;try {result = zooKeeper.create("/zk001",//节点的全路径"zk001-data".getBytes(),//节点中的数据->字节数据ZooDefs.Ids.OPEN_ACL_UNSAFE,//指定访问控制列表CreateMode.EPHEMERAL //指定创建节点的类型);Thread.sleep(10000);} catch (Exception e) {LOGGER.error(e.getMessage());}LOGGER.info("create node success,result={}",result);}

4.5 查看节点

查询节点有两层，第一个是相当于zkCli的get，就是获取某个节点的内容。还有一个就是类似于ls，列出子节点。

获取获取某个节点的内容可以通过zookeeper的getData方法，getData方法有多个重载，主要就是分为直接获取和异步获取，异步获取多了一个回掉，直接获取则直接返回获取的结果。

// 同步方式
getData(String path, Watcher watcher, Stat stat)
// 异步方式
getData(String path, Watcher watcher， AsyncCallback.DataCallback callBack，
Object ctx)
// 同步方式
getData(String path, boolean watch, Stat stat)
// 异步方式
getData(String path, boolean watch， AsyncCallback.DataCallback callBack，
Object ctx)

path：znode路径
watcher：使用新的注册的监视器，该参数允许传入null
watch：当watch为true时，则使用系统默认的Watcher，系统默认的Watcher是在zookeeper的构造函数中传递的Watcher。如果watch为false，则表明不注册Watcher。
stat ：返回znode的元数据
callBack：异步回调接口
ctx：传递上下文参数

同步方式获取某个节点的内容

    @Testpublic void getNodeData(){String result = null;try {//注意，是结点的全路径byte[] data = zooKeeper.getData("/zk01", null, null);result = new String(data);} catch (Exception e) {LOGGER.error(e.getMessage());Assert.fail();}LOGGER.info("getNodeData={}",result);}

异步方式获取某个节点的内容

这里要注意，一定要休眠，否则在看不到结果之前可能程序就停掉了。

    @Testpublic void getNodeDataAsync(){String result = null;try {zooKeeper.getData("/zk01", null, new AsyncCallback.DataCallback() {@Overridepublic void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {LOGGER.info("getNodeDataAsync={}",new String(bytes));}},null);Thread.sleep(3000);} catch (Exception e) {LOGGER.error(e.getMessage());Assert.fail();}}

列出所有的子节点

    //获取所有的子节点@Testpublic void getChilds(){try {List<String> children = zooKeeper.getChildren("/zk01", true);for(String node:children){LOGGER.info("================{}",node);}} catch (Exception e) {LOGGER.error(e.getMessage());}}

获取所有子节点，并打印其信息

    @Testpublic void getChilds2(){try {List<String> children = zooKeeper.getChildren("/zk01", true);Stat stat=null;for(String node:children){stat=new Stat();//封装结点的信息LOGGER.info("================{}",node);byte[] data=zooKeeper.getData("/zk01/"+node,null,stat);System.out.println(new String(data)+", stat:"+stat);}} catch (Exception e) {LOGGER.error(e.getMessage());}}

4.6 修改节点

// 同步方式
setData(String path, byte[] data, int version)
// 异步方式
setData(String path, byte[] data, int version， AsyncCallback.StatCallback
callBack， Object ctx)

path：znode路径
data：要存储在指定znode路径中的数据
version：这里的version指的是znode节点的dataVersion的值，每次数据的修改都会更新这个值，主要是为了保证一致性。通俗来讲就是如果你指定的version比保持的version值小，则表示已经有其他线程所更新了，你也就不能更新成功了，否则则可以更新成功。如果你不管别的线程有没有更新成功都要更新这个节点的值，则version可以指定为-1。
callBack：异步回调接口
ctx：传递上下文参数

下面是删除节点的例子

    @Testpublic void deleteNode(){try {//-1指的是无论你的结点是什么版本，都将删除zooKeeper.delete("/zk06/test-0000000008",-1);} catch (Exception e) {LOGGER.error(e.getMessage());Assert.fail();}}

4.7 watcher监听

客户端注册 Watcher，注册 watcher 有 3 种方式，getData、exists、getChildren，可以触发观察的操作有：create、delete、setData，下面分别进行测试：

（1）对于getData

    @Testpublic void testGetDataWather(){String result = "";try {//在读取数据时添加一个监听事件byte[] data = zooKeeper.getData("/zk01", new Watcher() {@Overridepublic void process(WatchedEvent event) {LOGGER.info("testGetDataWather watch type:{}", event.getType());//只能监听一次，想要持续监听可以通过循环或递归的方式//testGetDataWather();}}, null);result = new String(data);Thread.sleep(30000);} catch (Exception e) {LOGGER.error(e.getMessage());Assert.fail();}LOGGER.info("result = {}",result);}

主要/zk01节点中的数据改变，就会输出testGetDataWather watch type:NodeDataChanged，但只会出发一次，想要持续监听可以通过循环或递归的方式。

（2）对于exists

使用系统默认的Watcher是在zookeeper的构造函数中传递的Watcher

    /*** watch:true:表示使用系统默认的Watcher是在zookeeper的构造函数中传递的Watcher* watch:false:不使用Watcher*/@Testpublic void isExistWatcher1(){Stat stat = null;try {stat = zooKeeper.exists("/zk01/node1", true);} catch (Exception e) {LOGGER.error(e.getMessage());}//如果stat不为null，继续往后执行Assert.assertNotNull(stat);try {zooKeeper.delete("/zk01/node1",-1);} catch (Exception e) {LOGGER.error(e.getMessage());}}

出发事件时，将输出process info,eventType:NodeDeleted,eventState:SyncConnected,eventPath:/zk01/node1，因为zookeeper的构造函数中传递的Watcher的内容是：

String msg=String.format("process info,eventType:%s,eventState:%s,eventPath:%s",event.getType(),event.getState(),event.getPath());
LOGGER.info(msg);

然后，使用自定义的监听对象

    /*** 注册了自定义的监听对象，走自定义的。*/@Testpublic void isExistWatcher2(){Stat stat = null;try {stat = zooKeeper.exists("/zk01/node2", new Watcher() {@Overridepublic void process(WatchedEvent event) {LOGGER.info("isExistWatcher2  wather type:{}",event.getType());}});} catch (Exception e) {LOGGER.error(e.getMessage());}//如果stat不为null，继续往后执行Assert.assertNotNull(stat);try {zooKeeper.setData("/zk01/node2","isExistWatcher2_edited".getBytes(),-1);} catch (Exception e) {LOGGER.error(e.getMessage());}try {zooKeeper.delete("/zk01/node2",-1);} catch (Exception e) {LOGGER.error(e.getMessage());}}

输出的结果：isExistWatcher2 wather type:NodeDataChanged 或者 isExistWatcher2 wather type:NodeDeleted，不过一般是第一个操作触发。

（3）对于getChildren

对于getChildren只有子节点创建和删除时，才能触发watcher事件，子节点数据改变不会触发该事件，只有在子节点创建和删除时，才能触发watcher事件。

    @Testpublic void getChildsWatcher(){try {List<String> children = zooKeeper.getChildren("/zk01", new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {LOGGER.info("getChildsWatcher  wather type:{}",watchedEvent.getType());}});for(String node:children){LOGGER.info("================{}",node);//  /zk06/test-0000000001}Thread.sleep(30000);} catch (Exception e) {LOGGER.error(e.getMessage());}}

当子节点创建和删除时，会输出：getChildsWatcher wather type:NodeChildrenChanged

本文所有的demo的github地址为：https://github.com/wxler/zookeeperAPI.git

另外，我也通过zookeeper使用RMI远程调用，通过三个IP实现简单的负载均衡，也在上述github地址中。

【参考资料】

https://www.cnblogs.com/leesf456/p/6022357.html
https://blog.csdn.net/wx_it/article/details/105862972
https://www.runoob.com/w3cnote/zookeeper-acl.html