暂无图片
暂无图片
暂无图片
暂无图片
暂无图片
首页 / ZooKeeper实战篇之zk集群搭建、zkCli.sh操作、权限控制ACL、ZooKeeper JavaAPI使用

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

雷恩Layne 2021-08-02
2328

在了解Zookeeper原理详解[1]原理后，你是不是蠢蠢欲动想着手实践呢？这篇文章将手把手教你在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，实现四台服务器之间免密登录。

  1. 首先在每个服务器上产生自己的公钥，在每台服务器上执行以下命名，产生的公钥文件id_dsa.pub存放在/root/.ssh
    下：

    ssh-keygen  -t  dsa  -P  ''  -f  ~/.ssh/id_dsa

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

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

  3. 将layne1上的authorized_keys文件拷贝给layne2，在layne1上执行如下命令即可

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

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

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

  5. 将layne2上的authorized_keys文件拷贝给layne3

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

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

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

  7. 将layne3上的authorized_keys文件拷贝给layne4

    scp  ~/.ssh/authorized_keys   layne4:/root/.ssh/

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

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

  9. 将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[2]。

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

  1. 在https://zookeeper.apache.org/releases.html
    上下载zookeeper的Linux压缩包。

  2. 将zookeeper的压缩包zookeeper-3.4.6.tar.gz
    上传到layne2上。

  3. 解压至/opt
    目录下

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

  4. 配置zookeeper的环境变量，执行vim etc/profile
    ，在末尾加入：

    export ZOOKEEPER_HOME=/opt/zookeeper-3.4.6
    export PATH=$PATH:$ZOOKEEPER_HOME/bin

    然后执行source etc/profile
    ，让配置生效。

  5. 进入zookeeper的安装目录的conf下

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

  6. 复制zoo_sample.cfg
    文件为zoo.cfg

    cp zoo_sample.cfg zoo.cfg

  7. 先介绍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分配不同的端口号。

  8. 根据上面的参数说明，对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

  9. 创建/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

  10. 在/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

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

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

  2. 在layne3和layne4上分别修改myid

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

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

  4. 分别启动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客户端

  5. 启动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

  6. 再次尝试启动ZooKeeper，如果还报错，重启所有的虚拟机就好了。

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

  1. 没有创建zookeeper的日志目录/var/log/zookeeper/datalog
    。
  2. 没有在每个服务器的myid写入正确的编号。
  3. 没有用ssh 主机名
    在任意两台服务器之间进行第一次连接
  4. 如果不是上面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 args
 stat path [watch]
 set path data [version]
 ls path [watch]
 delquota [-n|-b] path
 ls2 path [watch]
 setAcl path acl
 setquota -n|-b val path
 history 
 redo cmdno
 printwatches on|off
 delete path [version]
 sync path
 listquota path
 rmr path
 get path [watch]
 create [-s] [-e] path data acl
 addauth scheme auth
 quit 
 getAcl path
 close 
 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(c)、删除权限 delete(d)、读权限 read(r)、写权限 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查询出来的结果包含所有前面添加的三个认证用户。

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

  1. setAcl命令中的id值是无效的，当使用addauth命令授权多个用户后，再用setAcl设置ACL时，会把当前会话所有addauth的用户都被会加入到acl中。
  2. 通过addauth命令(addauth digest <username>:<password>
    )授权的用户只在当前会话(session)有效    。
  3. setAcl命令设置权限后是永久式的，即使当前会话退出也不会消失。
  4. 如果在当前会话中，用户没有通过addauth授权就用setAcl设置acl权限时会失败。
  5. 使用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有如下特性：

  1. 授权是针对单个特定用户。
  2. 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 超级管理员[3]，我没有尝试，应该也可行。

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() {
    @Override
    public 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
     */
    @Test
    public 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
    ：传递上下文参数

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

    @Test
    public 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);
    }

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

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

    @Test
    public void getNodeDataAsync(){
        String result = null;
        try {
            zooKeeper.getData("/zk01", null, new AsyncCallback.DataCallback() {
                @Override
                public 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();
        }
    }

列出所有的子节点

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

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

    @Test
    public 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
    ：传递上下文参数

下面是删除节点的例子

    @Test
    public 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

    @Test
    public void testGetDataWather(){
        String result = "";
        try {
            //在读取数据时添加一个监听事件
            byte[] data = zooKeeper.getData("/zk01", new Watcher() {
                @Override
                public 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
     */
    @Test
    public 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);

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

    /**
     * 注册了自定义的监听对象，走自定义的。
     */
    @Test
    public void isExistWatcher2(){
        Stat stat = null;
        try {
            stat = zooKeeper.exists("/zk01/node2", new Watcher() {
                @Override
                public 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事件。

    @Test
    public void getChildsWatcher(){
        try {
            List<String> children = zooKeeper.getChildren("/zk01", new Watcher() {
                @Override
                public 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[4]

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

end

我是雷恩Layne，欢迎关注。如果在学习过程中有什么疑问，欢迎随时和我沟通交流。一起学习，一起努力，向着更好的明天进发！

长按图片一键关注

参考资料

[1]

Zookeeper原理详解: https://blog.csdn.net/qq_37555071/article/details/114609145

[2]

Linux中jdk安装方式: https://wxler.github.io/2021/02/10/112210/#rpm%E5%AE%89%E8%A3%85jdk

[3]

zookeeper设置super用户: https://blog.csdn.net/u010900754/article/details/78498291

[4]

zookeeper demo: https://github.com/wxler/zookeeperAPI.git


java
文章转载自雷恩Layne，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

评论