zookeeper(一)

1. zookeeper简介

1.1 分布式概述

早期我们使用单体架构，即所有服务部署在一台服务器的一个进程中，随着互联网的发展，逐步演进为分布式架构，多个服务分别部署在不同机器的不同进程中。

image-20200730181013482

1.2 zookeeper概述

zookeeper是一个开源的分布式协调服务，提供分布式数据一致性解决方案，分布式应用程序可以实现数据发布订阅、负载均衡、命名服务、集群管理分布式锁、分布式队列等功能。

zookeeper提供了分布式数据一致性解决方案，那什么是分布式数据一致性？首先我们谈谈什么叫一致性？

image-20200730181136559

如图在上图中有用户user在DB1中修改balance=900,如果user下一次read请求到DB2数据，此时DB1的数据还没及时更新到DB2中，就会造成整个数据库集群数据不一致。

数据一致性分为强一致性和最终一致性，强一致性指的如果数据不一致，就不对外提供数据服务，保证用户读取的数据始终是一致的。数据强一致性只需要通过锁机制即可解决，在案例中通过在DB2没有从DB1同步数据之前上锁，不对外提供读操作。只有当同步完成以后才对外提供服务。而最终一致性要求数据最终同步即可，没有实时性要求。

1.3 CAP原则

CAP在分布式系统中主要指的是一致性（(Consistency）、可用性（Availability）和分区容错性（Partition tolerance）

  • 一致性

    一致性指的是强一致性

  • 可用性

    系统提供的服务一直处于可用状态，用户的操作请求在指定的响应时间内响应请求，超出时间范围，认为系统不可用

  • 分区容错性

    分布式系统在遇到任何网络分区故障的时候，仍需要能够保证对外提供一致性和可用性服务，除非是整个网络都发生故障。

在一个分布式系统中不可能同时满足一致性、可用性、分区容错性，最多满足两个，对于分布式互联网应用而言，必须保证P，所以要么满足AP模型、要么满足CP模型

image-20200730182433524

1.4 一致性协议

事务需要跨多个分布式节点时，为了保证事务的ACID特性，需要选举出一个协调者来协调分布式各个节点的调度，基于这个思想衍生了多种一致性协议:

1.4.1 2PC 二阶段提交

顾名思义，二阶段提交叫事务的提交过程分为两个阶段：

image-20200730183051786

  • 阶段一 提交事务请求

    1 、协调者向所有的参与者节点发送事务内容，询问是否可以执行事务操作，并等待其他参与者节点的反馈

    2 、各参与者节点执行事务操作

    3 、各参与者节点反馈给协调者，事务是否可以执行

  • 阶段二 事务提交

    根据一阶段各个参与者节点反馈的ack,如果所有参与者节点反馈ack，则执行事务提交，否则中断事务

    • 事务提交：

      1 、协调者向各个参与者节点发送commit请求

      2 、参与者节点接受到commit请求后，执行事务的提交操作

      3 、各参与者节点完成事务提交后，向协调者返送提交commit成功确认消息

      4 、协调者接受各个参与者节点的ack后，完成事务commit

    • 中断事务：

      1 、发送回滚请求

      2 、各个参与者节点回滚事务

      3 、反馈给协调者事务回滚结果

      4 、协调者接受各参与者节点ack后回滚事务

二阶段提交存在的问题：

  • 同步阻塞

    二阶段提交过程中，所有参与事务操作的节点处于同步阻塞状态，无法进行其他的操作

  • 单点问题

    一旦协调者出现单点故障，无法保证事务的一致性操作

  • 脑裂导致数据不一致

    如果分布式节点出现网络分区，某些参与者未收到commit提交命令。则出现部分参与者完成数据提交。未收到commit的命令的参与者则无法进行事务提交，整个分布式系统便出现了数据不一致性现象。

1.4.2 3PC 三阶段提交

3PC是2PC的改进版，实质是将2PC中提交事务请求拆分为两步，形成了CanCommit、PreCommit、doCommit三个阶段的事务一致性协议

image-20200730184255383

  • 阶段一 : CanCommit

    1 、事务询问

    2 、各参与者节点向协调者反馈事务询问的响应

  • 阶段二 : PreCommit

    根据阶段一的反馈结果分为两种情况

    • 执行事务预提交

      1 、发送预提交请求

      协调者向所有参与者节点发送preCommit请求，进入prepared阶段

      2 、事务预提交

      各参与者节点接受到preCommit请求后，执行事务操作

      3 、各参与者节点向协调者反馈事务执行

    • 中断事务

      任意一个参与者节点反馈给协调者响应No时，或者在等待超时后，协调者还未收到参与者的反馈，就中断事务，中断事务分为两步：

      1 、协调者向各个参与者节点发送abort请求

      2 、参与者收到abort请求，或者等待超时时间后，中断事务

  • 阶段三 : doCommit

    • 执行提交

      1 、发送提交请求

      协调者向所有参与者节点发送doCommit请求

      2 、事务提交

      各参与者节点接受到doCommit请求后，执行事务提交操作

      3 、反馈事务提交结果

      各参与者节点完成事务提交以后，向协调者发送ack

      4 、事务完成

      协调者接受各个参与者反馈的ack后，完成事务

    • 中断事务

      1 、参与者接受到abort请求后，执行事务回滚

      2 、参与者完成事务回滚以后，向协调者发送ack

      3 、协调者接受回滚ack后，回滚事务

3PC相较于2PC而言，解决了协调者挂点后参与者无限阻塞和单点问题，但是仍然无法解决网络分区问题

1.4.3 Paxos算法

Paxos算法是Leslie Lamport 1990年提出的一种一致性算法，该算法是一种提高分布式系统容错性的一致性算法，解决了3PC中网络分区的问题，paxos算法可以在节点失效、网络分区、网络延迟等各种异常情况下保证所有节点都处于同一状态，同时paxos算法引入了“过半”理念，即少数服从多数原则。

paxos有三个版本：

  • Basic Paxos
  • Multi Paxos
  • Fast Paxos

在paxos算法中，有四种种角色，分别具有三种不同的行为，但多数情况，一个进程可能同时充当多种角色。

  • client：系统外部角色，请求发起者，不参与决策
  • proposer：提案提议者
  • acceptor：提案的表决者，即是否accept该提案，只有超过半数以上的acceptor接受了提案，该提案才被认为被“选定”
  • learners：提案的学习者，当提案被选定后，其同步执行提案，不参与决策

image-20200730195421460

Paxos算法分为两个阶段：prepare阶段、accept阶段

  • prepare阶段

    1 、proposer提出一个提案，编号为N,发送给所有的acceptor。

    2 、每个表决者都保存自己的accept的最大提案编号maxN，当表决者收到prepare(N)请求时，会比较N与maxN的值，若N小于maxN,则提案已过时，拒绝prepare(N)请求。若N大于等于maxN，则接受提案，并将该表决者曾经接受过的编号最大的提案Proposal(myid,maxN,value)反馈给提议者：其中myid表示表决者acceptor的标识id，maxN表示接受过的最大提案编号maxN，value表示提案内容。若当前表决者未曾accept任何提议，会将proposal(myid,null,null)反馈给提议者。

  • accept阶段

    1 、提议者proposal发出prepare(N),若收到超过半数表决者acceptor的反馈，proposal将真正的提案内容proposal(N,value)发送给所有表决者。

    2 、表决者acceptor接受提议者发送的proposal(N,value)提案后，会将自己曾经accept过的最大提案编号maxN和反馈过的prepare的最大编号，若N大于这两个编号，则当前表决者accept该提案，并反馈给提议者。否则拒绝该提议。

    3 、若提议者没有收到半数以上的表决者accept反馈，则重新进入prepare阶段，递增提案编号，重新提出prepare请求。若收到半数以上的accept，则其他未向提议者反馈的表决者称为learner，主动同步提议者的提案。

image-20200730202021534

正常流程

image-20200730202128379

单点故障，部分节点失败

image-20200730202145076

proposer失败

image-20200730202159173

Basic Paxos算法存在活锁问题（liveness）或dueling，而且较难实现

image-20200730202211999

Multi Paxos: 唯一的proposor，即leader

image-20200730202224216

简化角色

image-20200730202235533

1.4.4 ZAB协议(Fast Paxos)

由于paxos算法实现起来较难，存在活锁和全序问题（无法保证两次最终提交的顺序），所以zookeeper并没有使用paxos作为一致性协议，而是使用了ZAB协议。

ZAB（zookeeper atomic broadcast）:是一种支持崩溃恢复的原子广播协议，基于Fast paxos实现

ZooKeeper使用单一主进程Leader用于处理客户端所有事务请求,，即写请求。当服务器数据发生变更好，集群采用ZAB原子广播协议，以事务提交proposal的形式广播到所有的副本进程，每一个事务分配一个全局的递增的事务编号xid。

若客户端提交的请求为读请求时，则接受请求的节点直接根据自己保存的数据响应。若是写请求，且当前节点不是leader，那么该节点就会将请求转发给leader，leader会以提案的方式广播此写请求，如果超过半数的节点同意写请求，则该写请求就会提交。leader会通知所有的订阅者同步数据。

image-20200730211313432

zookeeper的三种角色：

为了避免zk的单点问题，zk采用集群方式保证zk高可用

  • leader

    leader负责处理集群的写请求，并发起投票，只有超过半数的节点同意后才会提交该写请求

  • follower

    处理读请求，响应结果。转发写请求到leader，在选举leader过程中参与投票

  • observer

    observer可以理解为没有投票权的follower，主要职责是协助follower处理读请求。那么当整个zk集群读请求负载很高时，为什么不增加follower节点呢？原因是增加follower节点会让leader在提出写请求提案时，需要半数以上的follower投票节点同意，这样会增加leader和follower的通信通信压力，降低写操作效率。

zookeeper两种模式：

  • 恢复模式

    当服务启动或领导崩溃后，zk进入恢复状态，选举leader，leader选出后，将完成leader和其他机器的数据同步，当大多数server完成和leader的同步后，恢复模式结束

  • 广播模式

    一旦Leader已经和多数的Follower进行了状态同步后，进入广播模式。进入广播模式后，如果有新加入的服务器，会自动从leader中同步数据。leader在接收客户端请求后，会生成事务提案广播给其他机器，有超过半数以上的follower同意该提议后，再提交事务。

    注意在ZAB的事务的二阶段提交中，移除了事务中断的逻辑，follower要么ack，要么放弃，leader无需等待所有的follower的ack。

zxid

zxid是 64 位长度的Long类型，其中高 32 位表示纪元epoch，低 32 位表示事务标识xid。即zxid由两部分构成：epoch和xid

每个leader都会具有不同的epoch值，表示一个纪元，每一个新的选举开启时都会生成一个新的epoch，新的leader产生，会更新所有的zkServer的zxid的epoch，xid是一个依次递增的事务编号。

leader选举算法：

image-20200730212416540

启动过程

  • 每一个server发出一个投票给集群中其他节点
  • 收到各个服务器的投票后，判断该投票有效性，比如是否是本轮投票，是否是 looking状态
  • 处理投票，pk别人的投票和自己的投票 比较规则xid>myid “取大原则”
  • 统计是否超过半数的接受相同的选票
  • 确认leader，改变服务器状态
  • 添加新server，leader已经选举出来，只能以follower身份加入集群中

崩溃恢复过程

  • leader挂掉后，集群中其他follower会将状态从FOLLOWING变为LOOKING,重新进入leader选举
  • 同上启动过程

image-20200730212528662

消息广播算法：

一旦进入广播模式，集群中非leader节点接受到事务请求，首先会将事务请求转发给服务器，leader服务器为其生成对应的事务提案proposal,并发送给集群中其他节点，如果过半则事务提交；

image-20200730212939123

  • leader接受到消息后，消息通过全局唯一的 64 位自增事务id，zxid标识
  • leader发送给follower的提案是有序的，leader会创建一个FIFO队列，将提案顺序写入队列中发送给follower
  • follower接受到提案后，会比较提案zxid和本地事务日志最大的zxid，若提案zxid比本地事务id大，将提案记录到本地日志中，反馈ack给leader，否则拒绝
  • leader接收到过半ack后，leader向所有的follower发送commit，通知每个follower执行本地事务

2. zookeeper环境搭建

zookeeper安装以linux环境为例，windows省略

2.1 单机环境

2.1.1 安装jdk

略……

2.1.2 zookeeper压缩包上传到linux

Alt+P 进入SFTP ，输入put d:\zookeeper-3.4.6.tar.gz 上传，d:\zookeeper-3.4.6.tar.gz是本地存放zookeeper的路径或者rz上传

2.1.3 解压缩压缩包

1
tar -zxvf zookeeper-3.4.6.tar.gz

2.1.4 创建 data 文件夹

进入 zookeeper-3.4.13 目录,创建data文件夹并修改conf文件夹下的zoo_sample.cfg为zoo.cfg

1
2
3
4
mkdir data

cd conf
mv zoo_sample.cfg zoo.cfg

2.1.5 修改zoo.cfg中的data属性

1
dataDir=/root/zookeeper-3.4.6/data

2.1.6 zookeeper服务启动

进入bin目录，启动服务输入命令

1
./zkServer.sh start

输出以下内容表示启动成功

image-20200730214246835

关闭服务输入命令

1
./zkServer.sh stop

输出以下提示信息

image-20200730214255293

查看状态

1
./zkServer.sh status

如果启动状态，提示

image-20200730214313289

如果未启动状态，提示

image-20200730214349044

2.2 集群环境

真实的集群是需要部署在不同的服务器上的，但是在我们测试时启动多个虚拟机的内存消耗太大，所以我们通常会搭建伪集群 ，也就是把所有的服务都搭建在一台虚拟机上，用端口进行区分。

2.2.1 准备工作

（ 1 ）安装JDK 【此步骤省略】。

（ 2 ）Zookeeper压缩包上传到服务器

（ 3 ）将Zookeeper解压到 /usr/local/zookeeper-cluster，复制三份zookeeper并改名为 zookeeper-1、zookeeper-2、zookeeper-3

（ 4 ）在解压后的Zookeeper目录下创建data目录

1
2
3
4
5
6
7
8
cd /usr/local/zookeeper-cluster/zookeeper-1
mkdir data

cd /usr/local/zookeeper-cluster/zookeeper-2
mkdir data

cd /usr/local/zookeeper-cluster/zookeeper-3
mkdir data

（ 5 ） 配置每一个Zookeeper 的dataDir（zoo.cfg） clientPort 分别为 2181 2182 2183

修改 /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

1
2
clientPort=2181
dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data

修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

1
2
clientPort=2182
dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data

修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

1
2
clientPort=2183
dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data

2.2.2 配置集群

（ 1 ）在每个zookeeper的 data 目录下创建一个 myid 文件，内容分别是 1 、 2 、 3 。这个文件就是记录每个服务器的ID

1
touch myid

（ 2 ）在每一个zookeeper 的 zoo.cfg配置客户端访问端口（clientPort）和集群服务器IP列表。

1
2
3
4
server.1=192.168.25.129:2881:3881
server.2=192.168.25.129:2882:3882
server.3=192.168.25.129:2883:3883
server.服务器ID=服务器IP地址：服务器之间通信端口：服务器之间投票选举端口

2.2.3 启动集群

依次启动三个zk实例，其中有一个leader和两个follower

2.2.4 observer角色及其配置

observer角色特点：

  • 不参与集群的leader选举
  • 不参与集群中写数据时的ack反馈

为了使用observer角色，在任何想变成observer角色的配置文件中加入如下配置：

1
peerType=observer

并在所有server的配置文件中，配置成observer模式的server的那行配置追加:observer，例如：

1
server.3=192.168.25.129:2883:3883:observer

2.2.5 zookeeperAPI连接集群

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
/**
 * zookeeper连接
 *
 * @author wgy
 */
public class ZookeeperConnection {

    public static void main(String[] args) {
        try {
            // 计数器对象
            CountDownLatch countDownLatch = new CountDownLatch(1);
            // arg1:服务器的ip和端口
            // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
            // arg3:监视器对象
            ZooKeeper zooKeeper = new ZooKeeper("192.168.142.128:2181,192.168.142.128:2182,192.168.142.128:2183", 5000, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                        System.out.println("连接创建成功!");
                        countDownLatch.countDown();
                    }
                }
            });
            // 主线程阻塞等待连接对象的创建成功
            countDownLatch.await();
            // 会话编号
            System.out.println(zooKeeper.getSessionId());
            zooKeeper.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3. zookeeper基本使用

3.1 数据结构

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

image-20200730220144122

Znode节点类型

  • 持久化目录节点（PERSISTENT）

    客户端与zookeeper断开连接后，该节点依旧存在

  • 持久化顺序编号目录节点（PERSISTENT_SEQUENTIAL）

    客户端与zookeeper断开连接后，该节点依旧存在，Zookeeper会给该节点按照顺序编号

  • 临时目录节点（EPHEMERAL）

    客户端与zookeeper断开连接后，该节点被删除

  • 临时顺序编号目录节点（EPHEMERAL_SEQUENTIAL）

    客户端与zookeeper断开连接后，该节点被删除，Zookeeper会给该节点按照顺序编号

节点的状态stat：

用来描述当前节点的创建、修改记录，包括cZxid、ctime等

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
#在zookeeper shell中使用stat命令查看指定路径节点的stat信息：
stat /ns-1/tenant

cZxid = 0x6a0000000a
ctime = Wed Mar 27 09:56:44 CST 2019
mZxid = 0x6a0000000a
mtime = Wed Mar 27 09:56:44 CST 2019
pZxid = 0x6a0000000e
cversion = 2
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 2

#属性说明：
cZxid：数据节点创建时的事务 ID
ctime：数据节点创建时的时间
mZxid：数据节点最后一次更新时的事务 ID
mtime：数据节点最后一次更新时的时间
pZxid：数据节点的子节点最后一次被修改时的事务 ID
cversion：子节点的更改次数
dataVersion：节点数据的更改次数
aclVersion：节点的 ACL 的更改次数
ephemeralOwner：如果节点是临时节点，则表示创建该节点的会话的
SessionID；如果节点是持久节点，则该属性值为 0
dataLength：数据内容的长度
numChildren：数据节点当前的子节点个数

3.2 命令行使用

通过zkClient进入zookeeper客户端命令行，输入help查看zookeeper客户端的指令

1
2
3
./zkcli.sh

help

image-20200730220426812

如上图列出了zookeeper所有的客户端命令行，下面主要讲解常见的几个命令行

  • 使用 ls 命令来查看当前znode中所包含的内容

    1
    2
    3
    ls path

    ls /

  • 查看当前节点数据并能看到更新次数等数据

    1
    2
    3
    ls2 path

    ls2 /

  • 创建节点 -s 含有序列 -e 临时

    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
    create [-s] [-e] path data #其中-s 为有序节点，-e 临时节点

    #创建持久化节点并写入数据：
    create /test "testvalue"

    #创建持久化有序节点，此时创建的节点名为指定节点名 + 自增序号
    create -s /a "aaa"
    Created /a0000000000

    create -s /b "bbb"
    Created /b0000000001

    create -s /c "ccc"
    Created /c0000000002

    #创建临时节点，临时节点会在会话过期后被删除
    create -e /tmp "tmp"
    Created /tmp

    #创建临时有序节点，临时节点会在会话过期后被删除
    create -s -e /aa "aaa"
    Created /aa0000000004

    create -s -e /bb "bbb"
    Created /bb0000000005

  • 获得节点的值

    1
    2
    3
    get path

    get /test

  • 设置节点的值

    1
    2
    3
    4
    5
    6
    7
    set path data [version]

    set /test "testvalueupdate"

    #也可以基于版本号进行更改，此时类似于乐观锁机制，当你传入的数据版本号(dataVersion) 和当前节点的数据版本号不符合时，zookeeper 会拒绝本次修改
    set /test "3456" 1
    version No is not valid : /test

  • 查看节点状态

    1
    2
    3
    stat path

    stat /test

  • 删除节点

    1
    2
    3
    4
    5
    6
    7
    delete path [version]

    delete /test

    #和更新节点数据一样，也可以传入版本号，当你传入的数据版本号 (dataVersion)和当前节点的数据版本号不符合时，zookeeper 不会执行删除操作
    delete /test 0
    version No is not valid : /test #无效的版本号

  • 递归删除节点

    1
    2
    3
    4
    5
    create /test "testvalue"

    create /test/test001 "testvalue001"

    rmr /test

  • 监听器get path [watch]

    1
    2
    3
    4
    5
    #使用 get path [watch] 注册的监听器能够在节点内容发生改变的时候，向客户端发出通知。需要注意的是 zookeeper 的触发器是一次性的 (One-time trigger)，即触发一次后就会立即失效

    get /test watch
    set /test 45678
    WATCHER::WatchedEvent state:SyncConnected type:NodeDataChanged path:/test #节点值改变

  • 监听器stat path [watch]

    1
    2
    3
    4
    #使用 stat path [watch] 注册的监听器能够在节点状态发生改变的时候，向客户端发出通知
    stat /test watch
    set /test 112233
    WATCHER::WatchedEvent state:SyncConnected type:NodeDataChanged path:/hadoop #节点值改变

  • 监听器ls\ls2 path [watch]

    1
    2
    3
    4
    #使用 ls path [watch] 或 ls2 path [watch] 注册的监听器能够监听该节点下所有子节点的增加和删除操作
    ls /test watch
    create /test/yarn "aaa"
    WatchedEvent state:SyncConnected type:NodeCreated path:/test/yarn

3.3 api使用

maven依赖

1
2
3
4
5
6
7
8
9
10
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.13</version>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
</dependency>
API 说明
ZooKeeper zk = new ZooKeeper(String connectString, int sessionTimeout,Watcher watcher) 创建zookeeper连接，connectString表示连接的zookeeper服务器的地址，sessionTimeOut指定会话的的超时时间，Watcher配置监听
String create(String path, byte[] data, List acl,CreateMode createMode) 创建一个给定的目录节点 path, 并给它设置数据，CreateMode 标识有四种形式的目录节点，分别是 PERSISTENT：持久化目录节点，这个目录节点存储的数据不会丢失；PERSISTENT_SEQUENTIAL：顺序自动编号的目录节点，这种目录节点会根据当前已近存在的节点数自动加 1 ，然后返回给客户端已经成功创建的目录节点名；EPHEMERAL：临时目录节点，一旦创建这个节点的客户端与服务器端口也就是 session 超时，这种节点会被自动删除；EPHEMERAL_SEQUENTIAL：临时自动编号节点
Stat exists(String path, boolean watch) 判断某个 path 是否存在，并设置是否监控这个目录节点，这里的watcher 是在创建 ZooKeeper 实例时指定的 watcher，exists方法还有一个重载方法，可以指定特定的watcher
Stat exists(String path,Watcher watcher) 重载方法，这里给某个目录节点设置特定的 watcher，Watcher 在ZooKeeper 是一个核心功能，Watcher 可以监控目录节点的数据变化以及子目录的变化，一旦这些状态发生变化，服务器就会通知所有设置在这个目录节点上的 Watcher，从而每个客户端都很快知道它所关注的目录节点的状态发生变化，而做出相应的反应
void delete(String path, int version 删除 path 对应的目录节点，version 为 -1 可以匹配任何版本，也就删除了这个目录节点所有数据
List getChildren(String path, boolean watch) 获取指定 path 下的所有子目录节点，同样 getChildren方法也有一个重载方法可以设置特定的 watcher 监控子节点的状态
Stat setData(String path, byte[] data, int version) 给 path 设置数据，可以指定这个数据的版本号，如果 version 为 -1 怎可以匹配任何版本
byte[] getData(String path, boolean watch,Stat stat) 获取这个 path 对应的目录节点存储的数据，数据的版本等信息可以通过 stat 来指定，同时还可以设置是否监控这个目录节点数据的状态

代码：

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
/**
 * zookeeper API测试
 *
 * @author wgy
 */
public class ZkApiTest {

    @Test
    public void test() throws Exception {
        //1. 创建zookeeper连接
        ZooKeeper zooKeeper = new ZooKeeper("192.168.142.128:2181", 30000, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
                System.out.println("触发了" + watchedEvent.getType() + "的事件");
            }
        });
        //2. 创建父节点
        String path = zooKeeper.create("/test", "testValue".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(path);
        //3. 创建子节点
        String childrenPath = zooKeeper.create("/test/children", "childrenValue".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(childrenPath);
        //4. 获取节点的值（父节点和子节点）
        byte[] data = zooKeeper.getData("/test", false, null);
        System.out.println(new String(data));

        List<String> children = zooKeeper.getChildren("/test", false);
        for (String child : children) {
            System.out.println(child);
        };
        //5. 修改节点的值
        Stat stat = zooKeeper.setData("/test", "testUpdate".getBytes(), -1);
        System.out.println(stat);
        //6. 判断某个节点是否存在
        Stat exists = zooKeeper.exists("/test/children", false);
        System.out.println(exists);
        //7. 删除节点
        zooKeeper.delete("/test/children", -1);
    }
}

4. zookeeper的acl权限控制

4.1 概述

zookeeper 类似文件系统，client 可以创建节点、更新节点、删除节点，那么如何做到节点的权限的控制呢？zookeeper的access control list 访问控制列表可以做到这一点。

acl 权限控制，使用scheme：id：permission: 来标识，主要涵盖 3 个方面：

  • 权限模式（scheme）：授权的策略
  • 授权对象（id）：授权的对象
  • 权限（permission）：授予的权限

其特性如下：

  • zooKeeper的权限控制是基于每个znode节点的，需要对每个节点设置权限
  • 每个znode支持设置多种权限控制方案和多个权限
  • 子节点不会继承父节点的权限，客户端无权访问某节点，但可能可以访问它的子节点
1
2
// 将节点权限设置为Ip:192.168.60.130的客户端可以对节点进行增、删、改、查、管理权限
setAcl /test2 ip:192.168.60.130:crwda

4.2 权限模式

采用何种方式授权

方案 描述
world 只有一个用户：anyone，代表登录zokeeper所有人（默认）
ip 对客户端使用IP地址认证
auth 使用已添加认证的用户认证
digest 使用“用户名:密码”方式认证

4.3 授权的对象

给谁授予权限

授权对象ID是指，权限赋予的实体，例如：IP 地址或用户。

4.4 授予的权限

授予什么权限

create、delete、read、writer、admin也就是 增、删、改、查、管理权限，这5种权限简写为cdrwa，注意:这5种权限中，delete是指对子节点的删除权限，其它4种权限指对自身节点的操作权限

权限 ACL简写 描述
create c 可以创建子节点
delete d 可以删除子节点（仅下一级节点）
read r 可以读取节点数据及显示子节点列表
write w 可以设置节点数据
admin a 可以设置节点访问控制列表权限

4.5 授权的相关命令

命令 使用方式 描述
getAcl getAcl 读取ACL权限
setAcl setAcl 设置ACL权限
addauth addauth 添加认证用户

4.6 案例

4.6.1 world授权模式

1
2
3
4
5
6
7
8
9
10
11
#命令
setAcl <path> world:anyone:<acl>

#案例
create /node1 "node1"

getAcl /node1
'world,'anyone #world方式对所有用户进行授权
: cdrwa #增、删、改、查、管理

setAcl /node1 world:anyone:cdrwa

4.6.2 IP授权模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#命令
setAcl <path> ip:<ip>:<acl>

#案例
#注意：远程登录zookeeper命令:./zkCli.sh -server ip

create /node2 "node2"

setAcl /node2 ip:192.168.60.129:cdrwa

getAcl /node2
'ip,'192.168.60.129
: cdrwa

#使用IP非 192.168.60.129 的机器
get /node2
Authentication is not valid : /node2 #没有权限

4.6.3 Auth授权模式

1
2
3
4
5
6
7
8
9
10
11
12
13
#命令
addauth digest <user>:<password> #添加认证用户
setAcl <path> auth:<user>:<acl>

#案例
create /node3 "node3"
#添加认证用户
addauth digest wgy:123456
setAcl /node3 auth:wgy:cdrwa

getAcl /node3
'digest,'wgy:860WKAFm6CvqEZcdivw5c3Q3i6Y=
: cdrwa

4.6.4 Digest授权模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#命令
setAcl <path> digest:<user>:<password>:<acl>

#这里的密码是经过SHA1及BASE64处理的密文，在SHELL中可以通过以下命令计算：
echo -n <user>:<password> | openssl dgst -binary -sha1 | openssl base64

#案例
create /node4 "node4"
setAcl /node4 digest:wgy:860WKAFm6CvqEZcdivw5c3Q3i6Y=:cdrwa

getAcl /node4
'digest,'wgy:860WKAFm6CvqEZcdivw5c3Q3i6Y=
: cdrwa

get /node4
Authentication is not valid : /node4 #没有权限
#添加认证用户
addauth digest wgy:123456

4.6.5 多种模式授权

1
2
3
4
5
6
7
8
#同一个节点可以同时使用多种模式授权

create /node5 "node5"

#添加认证用户
addauth digest wgy:123456 

setAcl /node5 ip:192.168.60.129:cdra,auth:wgy:cdrwa,digest:haha:3RTWRMbFGscZqJBONcrAlMYMrv0=:cdrwa

4.7 acl 超级管理员

zookeeper的权限管理模式有一种叫做super，该模式提供一个超管可以方便的访问任何权限的节点

假设这个超管是：super:admin，需要先为超管生成密码的密文

1
echo -n super:admin | openssl dgst -binary -sha1 | openssl base64

那么打开zookeeper目录下的/bin/zkServer.sh服务器脚本文件，找到如下一行

1
nohup $JAVA "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}"

这就是脚本中启动zookeeper的命令，默认只有以上两个配置项，我们需要加一个超管的配置项

1
"-Dzookeeper.DigestAuthenticationProvider.superDigest=super:xQJmxLMiHGwaqBvst5y6rkB6HQs="

那么修改以后这条完整命令变成了

1
2
nohup $JAVA "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" "-Dzookeeper.DigestAuthenticationProvider.superDigest=super:xQJmxLMiHGwaqBvst5y6rkB6HQs=" \
-cp "$CLASSPATH" $JVMFLAGS $ZOOMAIN "$ZOOCFG" > "$_ZOO_DAEMON_OUT" 2>&1 < /dev/null &

之后启动zookeeper,输入如下命令添加权限

1
2
#添加认证用户
addauth digest super:admin

5. zookeeper javaAPI

znode是zooKeeper集合的核心组件，zookeeper API提供了一小组方法使用zookeeper集合来操纵znode的所有细节。

客户端应该遵循以步骤，与zookeeper服务器进行清晰和干净的交互。

  • 连接到zookeeper服务器。zookeeper服务器为客户端分配会话ID。
  • 定期向服务器发送心跳。否则，zookeeper服务器将过期会话ID，客户端需要重新连接。
  • 只要会话ID处于活动状态，就可以获取/设置znode。
  • 所有任务完成后，断开与zookeeper服务器的连接。如果客户端长时间不活动，则zookeeper服务器将自动断开客户端。

5.1 连接到ZooKeeper

1
2
3
4
5
ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

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

案例:

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
/**
 * zookeeper连接
 *
 * @author wgy
 */
public class ZookeeperConnection {

    public static void main(String[] args) {
        try {
            // 计数器对象
            CountDownLatch countDownLatch = new CountDownLatch(1);
            // arg1:服务器的ip和端口
            // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
            // arg3:监视器对象
            ZooKeeper zooKeeper = new ZooKeeper("192.168.142.128:2181", 5000, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                        System.out.println("连接创建成功!");
                        countDownLatch.countDown();
                    }
                }
            });
            // 主线程阻塞等待连接对象的创建成功
            countDownLatch.await();
            // 会话编号
            System.out.println(zooKeeper.getSessionId());
            zooKeeper.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

5.2 新增节点

1
2
3
4
5
6
7
8
9
10
11
// 同步方式
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路径。例如，/node1 /node1/node11
data - 要存储在指定znode路径中的数据
acl - 要创建的节点的访问控制列表。zookeeper API提供了一个静态接口ZooDefs.Ids 来获取一些基本的acl列表。例如，ZooDefs.Ids.OPEN_ACL_UNSAFE返回打开znode的acl列表。
createMode - 节点的类型,这是一个枚举。
callBack-异步回调接口
ctx-传递上下文参数

案例:

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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
/**
 * zookeeper新增节点
 *
 * @author wgy
 */
public class ZKCreate {

    String IP = "192.168.142.128:2181";
    ZooKeeper zooKeeper;

    @Before
    public void before() throws Exception {
        // 计数器对象
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // arg1:服务器的ip和端口
        // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
        // arg3:监视器对象
        zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("连接创建成功!");
                    countDownLatch.countDown();
                }
            }
        });
        // 主线程阻塞等待连接对象的创建成功
        countDownLatch.await();
    }

    @After
    public void after() throws Exception {
        zooKeeper.close();
    }

    /**
     * 持久化节点
     *
     * @throws Exception
     */
    @Test
    public void create1() throws Exception {
        // arg1:节点的路径
        // arg2:节点的数据
        // arg3:权限列表  world:anyone:cdrwa
        // arg4:节点类型  持久化节点
        zooKeeper.create("/create/node1", "node1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     * 指定权限节点
     *
     * @throws Exception
     */
    @Test
    public void create2() throws Exception {
        // Ids.READ_ACL_UNSAFE world:anyone:r
        zooKeeper.create("/create/node2", "node2".getBytes(), ZooDefs.Ids.READ_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     * world授权模式
     *
     * @throws Exception
     */
    @Test
    public void create3() throws Exception {
        // world授权模式
        // 权限列表
        List<ACL> acls = new ArrayList<ACL>();
        // 授权模式和授权对象
        Id id = new Id("world", "anyone");
        // 权限设置
        acls.add(new ACL(ZooDefs.Perms.READ, id));
        acls.add(new ACL(ZooDefs.Perms.WRITE, id));
        zooKeeper.create("/create/node3", "node3".getBytes(), acls, CreateMode.PERSISTENT);
    }

    /**
     * ip授权模式
     *
     * @throws Exception
     */
    @Test
    public void create4() throws Exception {
        // ip授权模式
        // 权限列表
        List<ACL> acls = new ArrayList<ACL>();
        // 授权模式和授权对象
        Id id = new Id("ip", "192.168.60.130");
        // 权限设置
        acls.add(new ACL(ZooDefs.Perms.ALL, id));
        zooKeeper.create("/create/node4", "node4".getBytes(), acls, CreateMode.PERSISTENT);
    }

    /**
     * auth授权模式
     *
     * @throws Exception
     */
    @Test
    public void create5() throws Exception {
        // auth授权模式
        // 添加授权用户
        zooKeeper.addAuthInfo("digest", "wgy:123456".getBytes());
        zooKeeper.create("/create/node5", "node5".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
    }

    @Test
    public void create6() throws Exception {
        // auth授权模式
        // 添加授权用户
        zooKeeper.addAuthInfo("digest", "wgy:123456".getBytes());
        // 权限列表
        List<ACL> acls = new ArrayList<ACL>();
        // 授权模式和授权对象
        Id id = new Id("auth", "wgy");
        // 权限设置
        acls.add(new ACL(ZooDefs.Perms.READ, id));
        zooKeeper.create("/create/node6", "node6".getBytes(), acls, CreateMode.PERSISTENT);
    }

    /**
     * digest授权模式
     *
     * @throws Exception
     */
    @Test
    public void create7() throws Exception {
        // digest授权模式
        // 权限列表
        List<ACL> acls = new ArrayList<ACL>();
        // 授权模式和授权对象
        Id id = new Id("digest", "haha:3RTWRMbFGscZqJBONcrAlMYMrv0=");
        // 权限设置
        acls.add(new ACL(ZooDefs.Perms.ALL, id));
        zooKeeper.create("/create/node7", "node7".getBytes(), acls, CreateMode.PERSISTENT);
    }

    /**
     * 持久化顺序节点
     *
     * @throws Exception
     */
    @Test
    public void create8() throws Exception {
        // 持久化顺序节点
        // Ids.OPEN_ACL_UNSAFE world:anyone:cdrwa
        String result = zooKeeper.create("/create/node8", "node8".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
        System.out.println(result);
    }

    /**
     * 临时节点
     *
     * @throws Exception
     */
    @Test
    public void create9() throws Exception {
        //  临时节点
        // Ids.OPEN_ACL_UNSAFE world:anyone:cdrwa
        String result = zooKeeper.create("/create/node9", "node9".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println(result);
    }

    /**
     * 临时顺序节点
     *
     * @throws Exception
     */
    @Test
    public void create10() throws Exception {
        // 临时顺序节点
        // Ids.OPEN_ACL_UNSAFE world:anyone:cdrwa
        String result = zooKeeper.create("/create/node10", "node10".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(result);
    }

    @Test
    public void create11() throws Exception {
        // 异步方式创建节点
        zooKeeper.create("/create/node11", "node11".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new AsyncCallback.StringCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, String name) {
                // 0 代表创建成功
                System.out.println(rc);
                // 节点的路径
                System.out.println(path);
                // 节点的路径
                System.out.println(name);
                // 上下文参数 I am context
                System.out.println(ctx);
            }
        }, "I am context");
        Thread.sleep(10000);
        System.out.println("结束");
    }
}

5.3 更新节点

1
2
3
4
5
6
7
8
9
10
// 同步方式
setData(String path, byte[] data, int version)
// 异步方式
setData(String path, byte[] data, int version，AsyncCallback.StatCallback callBack， Object ctx)

path- znode路径
data - 要存储在指定znode路径中的数据。
version- znode的当前版本。每当数据更改时，ZooKeeper会更新znode的版本号。
callBack-异步回调接口
ctx-传递上下文参数

案例:

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
57
58
59
60
61
62
63
64
/**
 * 更新节点
 *
 * @author wgy
 */
public class ZKSet {

    String IP = "192.168.142.128:2181";
    ZooKeeper zooKeeper;

    @Before
    public void before() throws Exception {
        // 计数器对象
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // arg1:服务器的ip和端口
        // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
        // arg3:监视器对象
        zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("连接创建成功!");
                    countDownLatch.countDown();
                }
            }
        });
        // 主线程阻塞等待连接对象的创建成功
        countDownLatch.await();
    }

    @After
    public void after() throws Exception {
        zooKeeper.close();
    }

    @Test
    public void set1() throws Exception {
        // arg1:节点的路径
        // arg2:修改的数据
        // arg3:数据版本号 -1代表版本号不参与更新
        Stat stat = zooKeeper.setData("/set/node1", "node11".getBytes(), 2);
        // 当前节点的版本号
        System.out.println(stat.getVersion());
    }

    @Test
    public void set2() throws Exception {
        zooKeeper.setData("/set/node1", "node14".getBytes(), -1, new AsyncCallback.StatCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, Stat stat) {
                // 0代表修改成功
                System.out.println(rc);
                // 节点的路径
                System.out.println(path);
                // 上下文参数对象 I am Context
                System.out.println(ctx);
                // 属性描述对象
                System.out.println(stat.getVersion());
            }
        }, "I am Context");
        Thread.sleep(10000);
        System.out.println("结束");
    }
}

5.4 删除节点

1
2
3
4
5
6
7
8
9
// 同步方式
delete(String path, int version)
// 异步方式
delete(String path, int version, AsyncCallback.VoidCallback callBack,Object ctx)

path - znode路径。
version - znode的当前版本
callBack-异步回调接口
ctx-传递上下文参数

案例:

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
57
58
59
60
/**
 * 删除节点
 *
 * @author wgy
 */
public class ZKDelete {

    String IP = "192.168.142.128:2181";
    ZooKeeper zooKeeper;

    @Before
    public void before() throws Exception {
        // 计数器对象
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // arg1:服务器的ip和端口
        // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
        // arg3:监视器对象
        zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("连接创建成功!");
                    countDownLatch.countDown();
                }
            }
        });
        // 主线程阻塞等待连接对象的创建成功
        countDownLatch.await();
    }

    @After
    public void after() throws Exception {
        zooKeeper.close();
    }

    @Test
    public void delete1() throws Exception {
        // arg1:删除节点的节点路径
        // arg2:数据版本信息 -1代表删除节点时不考虑版本信息
        zooKeeper.delete("/delete/node1", -1);
    }

    @Test
    public void delete2() throws Exception {
        // 异步使用方式
        zooKeeper.delete("/delete/node2", -1, new AsyncCallback.VoidCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx) {
                // 0代表删除成功
                System.out.println(rc);
                // 节点的路径
                System.out.println(path);
                // 上下文参数对象 I am Context
                System.out.println(ctx);
            }
        }, "I am Context");
        Thread.sleep(10000);
        System.out.println("结束");
    }
}

5.5 查看节点

1
2
3
4
5
6
7
8
9
10
// 同步方式
getData(String path, boolean b, Stat stat)
// 异步方式
getData(String path, boolean b，AsyncCallback.DataCallback callBack，Object ctx)

path - znode路径。
b- 是否使用连接对象中注册的监视器。
stat - 返回znode的元数据。
callBack-异步回调接口
ctx-传递上下文参数

案例：

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
57
58
59
60
61
62
63
64
65
66
67
68
69
/**
 * 查看节点
 *
 * @author wgy
 */
public class ZKGet {

    String IP = "192.168.142.128:2181";
    ZooKeeper zooKeeper;

    @Before
    public void before() throws Exception {
        // 计数器对象
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // arg1:服务器的ip和端口
        // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
        // arg3:监视器对象
        zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("连接创建成功!");
                    countDownLatch.countDown();
                }
            }
        });
        // 主线程阻塞等待连接对象的创建成功
        countDownLatch.await();
    }

    @After
    public void after() throws Exception {
        zooKeeper.close();
    }

    @Test
    public void get1() throws Exception {
        // arg1:节点的路径
        // arg3:读取节点属性的对象
        Stat stat = new Stat();
        byte[] bys = zooKeeper.getData("/get/node1", false, stat);
        // 打印数据
        System.out.println(new String(bys));
        // 版本信息
        System.out.println(stat.getVersion());
    }

    @Test
    public void get2() throws Exception {
        //异步方式
        zooKeeper.getData("/get/node1", false, new AsyncCallback.DataCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
                // 0代表读取成功
                System.out.println(rc);
                // 节点的路径
                System.out.println(path);
                // 上下文参数对象 I am Context
                System.out.println(ctx);
                // 数据
                System.out.println(new String(data));
                // 属性对象
                System.out.println(stat.getVersion());
            }
        }, "I am Context");
        Thread.sleep(10000);
        System.out.println("结束");
    }
}

5.6 查看子节点

1
2
3
4
5
6
7
8
9
// 同步方式
getChildren(String path, boolean b)
// 异步方式
getChildren(String path, boolean b,AsyncCallback.ChildrenCallback callBack,Object ctx)

path - Znode路径。
b- 是否使用连接对象中注册的监视器。
callBack - 异步回调接口。
ctx-传递上下文参数

案例:

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
57
58
59
60
61
62
63
64
65
66
/**
 * 查看子节点
 *
 * @author wgy
 */
public class ZKGetChid {

    String IP = "192.168.142.128:2181";
    ZooKeeper zooKeeper;

    @Before
    public void before() throws Exception {
        // 计数器对象
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // arg1:服务器的ip和端口
        // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
        // arg3:监视器对象
        zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("连接创建成功!");
                    countDownLatch.countDown();
                }
            }
        });
        // 主线程阻塞等待连接对象的创建成功
        countDownLatch.await();
    }

    @After
    public void after() throws Exception {
        zooKeeper.close();
    }

    @Test
    public void get1() throws Exception {
        // arg1:节点的路径
        List<String> list = zooKeeper.getChildren("/get", false);
        for (String str : list) {
            System.out.println(str);
        }
    }

    @Test
    public void get2() throws Exception {
        // 异步用法
        zooKeeper.getChildren("/get", false, new AsyncCallback.ChildrenCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, List<String> children) {
                // 0代表读取成功
                System.out.println(rc);
                // 节点的路径
                System.out.println(path);
                // 上下文参数对象 I am Context
                System.out.println(ctx);
                // 子节点信息
                for (String str : children) {
                    System.out.println(str);
                }
            }
        }, "I am Context");
        Thread.sleep(10000);
        System.out.println("结束");
    }
}

5.7 检查节点是否存在

1
2
3
4
5
6
7
8
9
// 同步方法
exists(String path, boolean b)
// 异步方法
exists(String path, boolean b，AsyncCallback.StatCallback callBack,Object ctx)

path- znode路径。
b- 是否使用连接对象中注册的监视器。
callBack - 异步回调接口。
ctx-传递上下文参数

案例:

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
57
58
59
60
61
62
63
/**
 * 检查节点是否存在
 *
 * @author wgy
 */
public class ZKExists {

    String IP = "192.168.142.128:2181";
    ZooKeeper zooKeeper;

    @Before
    public void before() throws Exception {
        // 计数器对象
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // arg1:服务器的ip和端口
        // arg2:客户端与服务器之间的会话超时时间  以毫秒为单位的
        // arg3:监视器对象
        zooKeeper = new ZooKeeper(IP, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("连接创建成功!");
                    countDownLatch.countDown();
                }
            }
        });
        // 主线程阻塞等待连接对象的创建成功
        countDownLatch.await();
    }

    @After
    public void after() throws Exception {
        zooKeeper.close();
    }

    @Test
    public void exists1() throws Exception {
        // arg1:节点的路径
        Stat stat = zooKeeper.exists("/exists1", false);
        // 节点的版本信息
        System.out.println(stat.getVersion());
    }

    @Test
    public void exists2() throws Exception {
        // 异步方式
        zooKeeper.exists("/exists1", false, new AsyncCallback.StatCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, Stat stat) {
                // 0 代表方式执行成功
                System.out.println(rc);
                // 节点的路径
                System.out.println(path);
                // 上下文参数 I am Context
                System.out.println(ctx);
                // 节点的版本信息
                System.out.println(stat.getVersion());
            }
        }, "I am Context");
        Thread.sleep(10000);
        System.out.println("结束");
    }
}


-------------本文结束感谢您的阅读-------------



本文作者： Wgy
本文标题： zookeeper(一)
本文链接： https://wgy1993.gitee.io/archives/e18db595.html
版权声明： 本作品采用 CC BY-NC-SA 4.0 进行许可。转载请注明出处！
知识共享许可协议
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

本站总访问量:

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

程序员的一天