1. 部署与运行

ZK文档 http://zookeeper.apache.org/doc/r3.4.13/zookeeperStarted.html
ZK下载 https://www.apache.org/dyn/closer.cgi/zookeeper/

2. 客户端脚本

运行 zkCli.sh
连接到Zookeeper集群

2.1 创建

使用create命令创建一个Znode节点，用法如下：

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

其中-s和-e分别指定节点特性：顺序或者是临时节点。默认情况下，即不添加-s和-e参数，创建的是持久节点。如下：

[zk: localhost:2181(CONNECTED) 2] create /zk-book 123
Created /zk-book

2.2 读取

与读取数据有关的命令包括ls和get命令

ls命令

使用ls命令，可以列出Zookeeper指定节点下的所有子节点。

ls path [watch]

执行ls命令如下：

[zk: localhost:2181(CONNECTED) 2] create /zk-book 123
Created /zk-book

get命令

使用get命令，可以获取Zookeeper指定节点的数据内容和属性信息。用法如下：

get path [watch]

执行get命令如下：

[zk: localhost:2181(CONNECTED) 7] get /zk-book
123
cZxid = 0x200000002
ctime = Tue Jan 15 15:11:00 UTC 2019
mZxid = 0x200000002
mtime = Tue Jan 15 15:11:00 UTC 2019
pZxid = 0x200000002
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0

cZxid表示创建该节点的事务ID
mZxid表示更新该节点的事务ID
mtime表示最后一次更新该节点的时间

2.3 更新

使用set命令，可以指定更新节点的数据内容。

set path data [version]

其中data就是更新的数据内容。注意set命令最后还有一个version的参数，在Zookeeper 数据节点中，每个节点的数据都是由版本概念的。这个参数是用来指定本次更新的操作是基于哪个数据版本进行更新的。

执行如下命令：

[zk: localhost:2181(CONNECTED) 8] set /zk-book 456
cZxid = 0x200000002
ctime = Tue Jan 15 15:11:00 UTC 2019
mZxid = 0x200000005
mtime = Tue Jan 15 15:17:19 UTC 2019
pZxid = 0x200000002
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0
[zk: localhost:2181(CONNECTED) 9] get /zk-book
456
cZxid = 0x200000002
ctime = Tue Jan 15 15:11:00 UTC 2019
mZxid = 0x200000005
mtime = Tue Jan 15 15:17:19 UTC 2019
pZxid = 0x200000002
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0

2.4 删除

使用delete命令，可以删除Zookeeper上的指定节点。用法如下：

delete path [version]

此命令中的Version和set命令中的version参数的作用是一致的。

执行如下命令：

[zk: localhost:2181(CONNECTED) 10] delete /zk-book

执行完这个命令之后，就可以把/zk-book这个节点成功删除了。但是这里要注意一点，要想删除某一个指定节点，该节点必须没有子节点存在。

3. Java客户端API的使用

Zookeeper作为一个分布式服务框架，主要用来解决分布式数据一致性问题，它提供了简单的分布式原语，并且提供了多种编程语言API。

3.1 创建会话

客户端可以创建一个实例来连接Zookeeper服务器。Zookeeper的4种构造方法如下：

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly)public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,long sessionId, byte[] sessionPasswd)public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,long sessionId, byte[] sessionPasswd, boolean canBeReadOnly)

public class ZookeeperConstructor implements Watcher{private static CountDownLatch countDownLatch=new CountDownLatch(1);public static void main(String[] args) throws IOException {ZooKeeper zooKeeper=new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT,new ZookeeperConstructor());try {countDownLatch.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("Zookeeper session established");}@Overridepublic void process(WatchedEvent event) {if(Event.KeeperState.SyncConnected==event.getState()){countDownLatch.countDown();}}
}

运行结果：

Zookeeper session established

3.2 创建节点

客户端可以通过API创建一个数据节点，Zookeeper提供了两种创建数据节点的方式。一种是同步方式一种是异步方式。

同步创建数据节点
public String create(final String path, byte data[], List<ACL> acl,CreateMode createMode)
异步创建数据节点
public void create(final String path, byte data[], List<ACL> acl,CreateMode createMode,  StringCallback cb, Object ctx)

使用同步API创建一个节点

//ZooKeeper API创建节点，使用同步(sync)接口。
public class ZooKeeperCreate implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);public static void main(String[] args) throws Exception{ZooKeeper zookeeper = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT, //new ZooKeeperCreate());connectedSemaphore.await();String path1 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);System.out.println("Success create znode: " + path1);String path2 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println("Success create znode: " + path2);}public void process(WatchedEvent event) {if (KeeperState.SyncConnected == event.getState()) {connectedSemaphore.countDown();}}
}

运行结果：

Success create znode: /zk-test-ephemeral-
Success create znode: /zk-test-ephemeral-0000000003

使用异步API创建一个节点

// ZooKeeper API创建节点，使用异步(async)接口。
public class ZooKeeperCreateASync implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);public static void main(String[] args) throws Exception{ZooKeeper zookeeper = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT, //new ZooKeeperCreateASync());connectedSemaphore.await();zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new IStringCallback(), "I am context.");zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new IStringCallback(), "I am context.");zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, new IStringCallback(), "I am context.");Thread.sleep( Integer.MAX_VALUE );}public void process(WatchedEvent event) {if (Event.KeeperState.SyncConnected == event.getState()) {if (Event.EventType.None == event.getType() && null == event.getPath()) {connectedSemaphore.countDown();}}}
}
class IStringCallback implements AsyncCallback.StringCallback{public void processResult(int rc, String path, Object ctx, String name) {System.out.println("Create path result: [" + rc + ", " + path + ", "+ ctx + ", real path name: " + name);}}

运行结果：

Create path result: [0, /zk-test-ephemeral-, I am context., real path name: /zk-test-ephemeral-
Create path result: [-110, /zk-test-ephemeral-, I am context., real path name: null
Create path result: [0, /zk-test-ephemeral-, I am context., real path name: /zk-test-ephemeral-0000000005

3.3 删除节点

客户端可以通过API创建一个数据节点，Zookeeper提供了两种删除数据节点的方式。一种是同步方式一种是异步方式。

Zookeeper deleteAPI 参数接口

// ZooKeeper API 删除节点，使用同步(sync)接口。
public class DeleteSync implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);private static ZooKeeper zk;public static void main(String[] args) throws Exception {String path = "/zk-book";zk = new ZooKeeper("domain1.book.zookeeper:2181", 5000, //new DeleteSync());connectedSemaphore.await();zk.create( path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL );zk.delete( path, -1 );Thread.sleep( Integer.MAX_VALUE );}@Overridepublic void process(WatchedEvent event) {if (KeeperState.SyncConnected == event.getState()) {if (EventType.None == event.getType() && null == event.getPath()) {connectedSemaphore.countDown();}}}
}

运行结果：

删除成功

3.4 读取数据

getChildren

客户端可以通过API获取一个节点的所有子节点，有如下8个接口可以供使用：

public List<String> getChildren(final String path, Watcher watcher)public List<String> getChildren(String path, boolean watch)public void getChildren(final String path, Watcher watcher,ChildrenCallback cb, Object ctx)public void getChildren(String path, boolean watch, ChildrenCallback cb, Object ctx)public List<String> getChildren(final String path, Watcher watcher,Stat stat)等等

// ZooKeeper API 获取子节点列表，使用同步(sync)接口。
public class ZooKeeperGetChildSync implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);private static ZooKeeper zk = null;public static void main(String[] args) throws Exception{String path = "/zk-book";zk = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT, //new ZooKeeperGetChildSync());connectedSemaphore.await();zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);zk.create(path+"/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);List<String> childrenList = zk.getChildren(path, true);System.out.println(childrenList);zk.create(path+"/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);Thread.sleep( Integer.MAX_VALUE );}public void process(WatchedEvent event) {if (KeeperState.SyncConnected == event.getState()) {if (EventType.None == event.getType() && null == event.getPath()) {//说明事件通知是连接建立事件通知connectedSemaphore.countDown();} else if (event.getType() == EventType.NodeChildrenChanged) {try {System.out.println("ReGet Child:"+zk.getChildren(event.getPath(),true));} catch (Exception e) {}}}}

运行结果：

[c1]
ReGet Child:[c1, c2]

首先我们来看注册Watcher。如果Zookeeper客户端在获取指定节点列表之后，还需要订阅这个子列表的变化通知，那就可以注册一个Watcher来实现。当有子节点被添加或者是删除的时候，服务器就会向客户端发送一个NodeChildrenChanged的通知。需要注意的是，在服务器发送给客户端的通知事件中，是不包含最新的节点列表的，客户端必须重新进行获取。

getData

客户端可以通过getData API获取一个节点的内容。getData有如下4个接口：

public byte[] getData(final String path, Watcher watcher, Stat stat)public byte[] getData(String path, boolean watch, Stat stat)public void getData(final String path, Watcher watcher,
DataCallback cb, Object ctx)public void getData(String path, boolean watch, DataCallback cb, Object ctx)

Zookeeper getData API 方法参数说明：

// ZooKeeper API 获取节点数据内容，使用同步(sync)接口。
public class GetDataSync implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);private static ZooKeeper zk = null;private static Stat stat = new Stat();public static void main(String[] args) throws Exception {String path = "/zk-book";zk = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT, //new GetDataSync());connectedSemaphore.await();zk.create( path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL );System.out.println(new String(zk.getData( path, true, stat )));System.out.println(stat.getCzxid()+","+stat.getMzxid()+","+stat.getVersion());zk.setData( path, "123".getBytes(), -1 );Thread.sleep( Integer.MAX_VALUE );}public void process(WatchedEvent event) {if (KeeperState.SyncConnected == event.getState()) {if (EventType.None == event.getType() && null == event.getPath()) {connectedSemaphore.countDown();} else if (event.getType() == EventType.NodeDataChanged) {try {System.out.println(new String(zk.getData( event.getPath(), true, stat )));System.out.println(stat.getCzxid()+","+stat.getMzxid()+","+stat.getVersion());} catch (Exception e) {}}}}
}

由于我们之前在这个节点上注册了一个Watcher,所以一旦该节点的数据发生变化，Zookeeper服务端就会向客户端发出一个NodeDataChanged“数据变更”的通知。于是客户端在收到这个通知之后，再次通过调用getData接口来返回最新的数据内容。

另外，在调用getData接口的同时，我们传入了一个新的Stat变量，在Zookeeper客户端的实现中，会从服务端的响应中获取到数据节点的最新节点的状态信息，来替换这个客户端的旧状态。

运行结果:

123
8589934622,8589934622,0
123
8589934622,8589934623,1

3.5 更新数据

Zookeeper客户端可以通过Zookeeper的API来更新一个节点的数据内容，有如下两个接口：

public Stat setData(final String path, byte data[], int version)public void setData(final String path, byte data[], int version, StatCallback cb, Object ctx)

这里列出的两个API分别是同步和异步更新的两个API接口

更新的接口就较为简单了。我们重点来看一下方法中的version参数。version参数用于制定数据节点的数据版本，表明本次更新是针对数据版本进行更新的。

具体来说，假设一个客户端进行更新操作，它会携带上次获取到的version值进行更新。而如果在这段时间内，Zookeeper服务器上该节点的数据恰好已经被其他客户端匹配了，那么其数据版本一定也发生了变化，因此与客户端携带的version无法匹配。于是其更新操作失败————因此Zookeeper利用此特性可以有效地避免一些分布式更新当中的并发问题。

// ZooKeeper API 更新节点数据内容，使用同步(sync)接口。
public class SetDataSync implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);private static ZooKeeper zk;public static void main(String[] args) throws Exception {String path = "/zk-book";zk = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT, //new SetDataSync());connectedSemaphore.await();zk.create( path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL );zk.getData( path, true, null );Stat stat = zk.setData( path, "456".getBytes(), -1 );System.out.println(stat.getCzxid()+","+stat.getMzxid()+","+stat.getVersion());Stat stat2 = zk.setData( path, "456".getBytes(), stat.getVersion() );System.out.println(stat2.getCzxid()+","+stat2.getMzxid()+","+stat2.getVersion());try {zk.setData( path, "456".getBytes(), stat.getVersion() );} catch ( KeeperException e ) {System.out.println("Error: " + e.code() + "," + e.getMessage());}Thread.sleep( Integer.MAX_VALUE );}@Overridepublic void process(WatchedEvent event) {if (KeeperState.SyncConnected == event.getState()) {if (EventType.None == event.getType() && null == event.getPath()) {connectedSemaphore.countDown();}}}
}

运行结果：

8589934627,8589934628,1
8589934627,8589934629,2
Error: BADVERSION,KeeperErrorCode = BadVersion for /zk-book

在第一次更新操作中，使用的版本是“-1”,并且更新成功。在Zookeeper中数据版本都是从0开始计数的，如果客户端传入的版本参数是-1就是告诉Zookeeper服务器，客户端需要基于数据最新版本进行更新操作。

第一次更新操作之后返回给客户端一个数据节点的节点状态信息：Stat。从这个数据结构，我们就可以获取该节点最新版本信息。

3.6 检查数据是否存在

Zookeeper客户端可以通过以下4种API来判断数据节点是否存在：

public Stat exists(final String path, Watcher watcher)public Stat exists(String path, boolean watch)public void exists(final String path, Watcher watcher,StatCallback cb, Object ctx)public void exists(String path, boolean watch, StatCallback cb, Object ctx)

这里列出来的4种API分别使用同步和异步模式检查数据节点是否存在。API的参数说明如下：

// ZooKeeper API 判断节点是否存在，使用同步(sync)接口。
public class ExistSync implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);private static ZooKeeper zk;public static void main(String[] args) throws Exception {String path = "/zk-book";zk = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT, //new ExistSync());connectedSemaphore.await();zk.exists( path, true );zk.create( path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT );zk.setData( path, "123".getBytes(), -1 );zk.create( path+"/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT );zk.delete( path+"/c1", -1 );zk.delete( path, -1 );Thread.sleep( Integer.MAX_VALUE );}@Overridepublic void process(WatchedEvent event) {try {if (KeeperState.SyncConnected == event.getState()) {if (EventType.None == event.getType() && null == event.getPath()) {connectedSemaphore.countDown();} else if (EventType.NodeCreated == event.getType()) {System.out.println("Node(" + event.getPath() + ")Created");zk.exists( event.getPath(), true );} else if (EventType.NodeDeleted == event.getType()) {System.out.println("Node(" + event.getPath() + ")Deleted");zk.exists( event.getPath(), true );} else if (EventType.NodeDataChanged == event.getType()) {System.out.println("Node(" + event.getPath() + ")DataChanged");zk.exists( event.getPath(), true );}}} catch (Exception e) {}}
}

运行结果：

Node(/zk-book)Created
Node(/zk-book)DataChanged
Node(/zk-book)Deleted

3.7 权限控制

为了避免存储在Zookeeper服务器上面的数据被其他进程干扰或者认为操作修改，需要对Zookeeper上的数据访问进行权限控制（Access Control）。Zookeeper提供了ACL的权限控制机制，简单的说，就是通过设置Zookeeper服务器上数据节点的ACL来控制对数据节点的数据访问权限：如果一个客户端符合该ACL控制，那么就可以访问，否则将无法操作。

针对这样的控制机制，Zookeeper提供了多种权限控制模式（Schema）。分别是world、auth、digest、ip和super。

开发人员如果要用Zookeeper的权限控制功能，需要在完成Zookeeper会话创建后给会话添加上相应的权限信息（AuthInfo）。

addAuthInfo(String name, byte[] auth)

//删除节点的权限控制
public class AuthSample_Delete {final static String PATH  = "/zk-book-auth_test";final static String PATH2 = "/zk-book-auth_test/child";public static void main(String[] args) throws Exception {ZooKeeper zookeeper1 = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT,null);zookeeper1.addAuthInfo("digest", "foo:true".getBytes());zookeeper1.create( PATH, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT );zookeeper1.create( PATH2, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL );try {ZooKeeper zookeeper2 = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT,null);zookeeper2.delete( PATH2, -1 );} catch ( Exception e ) {System.out.println( "删除节点失败: " + e.getMessage() );}ZooKeeper zookeeper3 = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT,null);zookeeper3.addAuthInfo("digest", "foo:true".getBytes());zookeeper3.delete( PATH2, -1 );System.out.println( "成功删除节点：" + PATH2 );ZooKeeper zookeeper4 = new ZooKeeper(Constant.ZK_CONNECT_STRING,Constant.ZK_SESSION_TIMEOUT,null);zookeeper4.delete( PATH, -1 );System.out.println( "成功删除节点：" + PATH );}
}

运行结果：

删除节点失败: KeeperErrorCode = NoAuth for /zk-book-auth_test/child成功删除节点：/zk-book-auth_test/child成功删除节点：/zk-book-auth_test/child

当客户端对一个数据添加权限信息之后，对于删除操作，其作用范围是其子节点，也就是说当我们对一个数据节点添加权限范围后，依然可以自由的删除这个节点，但是对于这个节点的子节点，就必须使用相应的权限信息才能够删除它。

Zookeeper分布式一致性原理（五）：Zookeeper-Java-API相关推荐

《从Paxos到zookeeper分布式一致性原理与实践》笔记
<从Paxos到zookeeper分布式一致性原理与实践>笔记文章目录 <从Paxos到zookeeper分布式一致性原理与实践>笔记一.概念二.一致性协调 2.1 2P ...
《从Paxos到zookeeper分布式一致性原理与实践》
<从Paxos到zookeeper分布式一致性原理与实践> 一.概念 ACID: Automaticy.consistency.isolation. Durability CAP: con ...
[201502][从 Paxos 到 ZooKeeper][分布式一致性原理与实践][倪超][著]
[201502][从 Paxos 到 ZooKeeper][分布式一致性原理与实践][倪超][著] http://zookeeper.apache.org 第 1 章分布式架构 1.1 从集中式到分 ...
Zookeeper分布式一致性原理（四）：Zookeeper简介
zookeeper是一个典型的分布式数据一致性的解决方案,分布式应用程序可以基于它实现数据发布/订阅.负载均衡.命名服务.分布式协调/通知.集群管理.master选举.分布式锁和分布式队列等.Zook ...
《从Paxos到Zookeeper 分布式一致性原理与实践》
第1章分布式架构 1．1 从集中式到分布式 1．1．1 集中式的特点集中式的特点:部署结构简单(因为基于底层性能卓越的大型主机,不需考虑对服务多个节点的部署,也就不用考虑多个节点之间分布式协调问题 ...
Zookeeper分布式一致性原理（八）：Zookeeper典型应用场景
1. 简介 Zookeeper是一个高可用的分布式数据管理和协调框架,并且能够很好的保证分布式环境中数据的一致性.在越来越多的分布式系统(Hadoop.HBase.Kafka)中,Zookeeper都 ...
《从Paxos到ZooKeeper 分布式一致性原理与实践》读书笔记
一.分布式架构 1.分布式特点分布性对等性.分布式系统中的所有计算机节点都是对等的并发性.多个节点并发的操作一些共享的资源缺乏全局时钟.节点之间通过消息传递进行通信和协调,因为缺乏全局时钟,很 ...
Zookeeper分布式一致性原理（六）：Zookeeper开源客户端zkClient
zkClient 是GitHub上的一个开源Zookeeper客户端项目,是由Datameer的工程师Stefan和Groschupf和PeteVoss一起开发的.zkClient在Zookeeper ...
Zookeeper分布式一致性原理（二）：一致性协议
为了解决分布式一致性问题,在长期的研究过程中,提出了一大批经典的一致性协议和算法,其中最著名的就是2PC和3PC以及Paxos算法了. 1. 2PC和3PC 在分布式系统中,每个节点都明确知道自己事务 ...

Zookeeper分布式一致性原理（五）：Zookeeper-Java-API

1. 部署与运行

2. 客户端脚本

2.1 创建

2.2 读取

ls命令

get命令

2.3 更新

2.4 删除

3. Java客户端API的使用

3.1 创建会话

3.2 创建节点

3.3 删除节点

3.4 读取数据

getChildren

getData

3.5 更新数据

3.6 检查数据是否存在

3.7 权限控制

Zookeeper分布式一致性原理（五）：Zookeeper-Java-API相关推荐

最新文章

热门文章