客户端的操作
zkCli.sh
ls /
[zookeeper]
get /
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1
create /test 123
Created /test
set /test 456
cZxid = 0x2
ctime = Wed Mar 28 11:20:40 CST 2018
mZxid = 0x3
mtime = Wed Mar 28 11:21:22 CST 2018
pZxid = 0x2
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0
delete /test
Java API
创建会话
public class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
public void process(WatchedEvent watchedEvent) {
System.out.println("receive wathedEvent:" + watchedEvent);
if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
countDownLatch.countDown();
}
}
public static void main(String[] args) throws IOException, InterruptedException {
ZooKeeper zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java());
System.out.println("zk status:" + zooKeeper.getState());
countDownLatch.await();
}
}
输出：
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
使用sessionId，sessionPasswd
ZooKeeper zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java());
System.out.println("zk status:" + zooKeeper.getState());
countDownLatch.await();
long sessionId = zooKeeper.getSessionId();
byte[] sessionPasswd = zooKeeper.getSessionPasswd();
zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java(), sessionId, sessionPasswd);
zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java(), sessionId, "123456".getBytes());
Thread.sleep(Integer.MAX_VALUE);
结果：
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
receive wathedEvent:WatchedEvent state:Expired type:None path:null
receive wathedEvent:WatchedEvent state:Disconnected type:None path:null
receive wathedEvent:WatchedEvent state:Disconnected type:None path:null
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
Expired 是使用错误sessionPasswd的结果
至于结果中的不断地SyncConnected，Disconnected尚不清楚。
创建节点
同步执行：创建临时节点，临时顺序节点
//sync
String s = zooKeeper.create("/test1", "value1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("success create path:" + s);
s = zooKeeper.create("/test1", "value1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("success create path:" + s);
结果：
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
success create path:/test1
success create path:/test10000000001
临时，顺序在zkCli用-e， -s指定，临时节点会在会话结束时删除，顺序节点会自动添加序号。
同步执行：
public class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
public void process(WatchedEvent watchedEvent) {
System.out.println("receive wathedEvent:" + watchedEvent);
if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
countDownLatch.countDown();
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
ZooKeeper zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java());
System.out.println("zk status:" + zooKeeper.getState());
countDownLatch.await();
IStringCallBack iStringCallBack = new ZkDemon4java().new IStringCallBack();
//async
zooKeeper.create("/test1", "value1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, iStringCallBack, "i am ctx");
zooKeeper.create("/test1", "value1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, iStringCallBack, "i am ctx");
zooKeeper.create("/test1", "value1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, iStringCallBack, "i am ctx");
Thread.sleep(Integer.MAX_VALUE);
}
class IStringCallBack implements AsyncCallback.StringCallback {
public void processResult(int i, String s, Object o, String s1) {
System.out.println("Create Path result resultCode:" + i + ",path:" + s + ",ctx:" + o.toString() + ",real path name:" + s1);
}
}
}
结果：
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
Create Path result resultCode:0,path:/test1,ctx:i am ctx,real path name:/test1
Create Path result resultCode:-110,path:/test1,ctx:i am ctx,real path name:null
Create Path result resultCode:0,path:/test1,ctx:i am ctx,real path name:/test10000000009
删除节点
zooKeeper.delete("/test", -1);
其中version=-1表示不论节点的dataVersion是什么都删除，如果指定一个大于-1的数，则只当version不一样时不能删除。
读取数据
getChildren（同步）
public class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static ZooKeeper zooKeeper;
public void process(WatchedEvent watchedEvent) {
System.out.println("receive wathedEvent:" + watchedEvent);
if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
if (watchedEvent.getType() == Event.EventType.None && null == watchedEvent.getPath()) {
countDownLatch.countDown();
} else if (Event.EventType.NodeChildrenChanged == watchedEvent.getType()) {
try {
//if you need get advice again you must set watch:true
System.out.println("ReGet child:" + zooKeeper.getChildren(watchedEvent.getPath(), false));
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java());
System.out.println("zk status:" + zooKeeper.getState());
countDownLatch.await();
String path = "/zk_book";
zooKeeper.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//get children List and register Watch
System.out.println("get child result:" + zooKeeper.getChildren(path, new ZkDemon4java()));
zooKeeper.create(path + "/c1", "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
zooKeeper.create(path + "/c2", "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
Thread.sleep(Integer.MAX_VALUE);
}
}
System.out.println("ReGet child:" + zooKeeper.getChildren(watchedEvent.getPath(), false));一次通知
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
get child result:[]
receive wathedEvent:WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/zk_book
ReGet child:[c1]
System.out.println("ReGet child:" + zooKeeper.getChildren(watchedEvent.getPath(), true));多次通知
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
get child result:[]
receive wathedEvent:WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/zk_book
ReGet child:[c1]
receive wathedEvent:WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/zk_book
ReGet child:[c1, c2]
getChildren（异步）
class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static ZooKeeper zooKeeper;
public void process(WatchedEvent watchedEvent) {
System.out.println("receive wathedEvent:" + watchedEvent);
if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
if (watchedEvent.getType() == Event.EventType.None && null == watchedEvent.getPath()) {
countDownLatch.countDown();
} else if (Event.EventType.NodeChildrenChanged == watchedEvent.getType()) {
try {
//if you need get advice again you must set watch:true
System.out.println("ReGet child:" + zooKeeper.getChildren(watchedEvent.getPath(), true));
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java());
System.out.println("zk status:" + zooKeeper.getState());
countDownLatch.await();
String path = "/zk_book";
zooKeeper.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//get children List and register Watch
zooKeeper.getChildren(path, true, new IChildren2Callback(), "i am ctx");
zooKeeper.create(path + "/c1", "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
zooKeeper.create(path + "/c2", "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
Thread.sleep(Integer.MAX_VALUE);
}
}
class IChildren2Callback implements AsyncCallback.Children2Callback {
public void processResult(int i, String s, Object o, List<String> list, Stat stat) {
System.out.println("get children node result: result code:" + i + ",param path:" + s + ",ctx:" + o.toString() + ",children List:" + list + ",stat:" + stat);
}
}
结果：
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
get children node result: result code:0,param path:/zk_book,ctx:i am ctx,children List:[],stat:211,211,1522231252330,1522231252330,0,0,0,0,3,0,211
receive wathedEvent:WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/zk_book
ReGet child:[c1]
receive wathedEvent:WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/zk_book
ReGet child:[c1, c2]
getData(同步)
public class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static ZooKeeper zooKeeper;
public void process(WatchedEvent watchedEvent) {
System.out.println("receive wathedEvent:" + watchedEvent);
try {
if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
if (watchedEvent.getType() == Event.EventType.None && null == watchedEvent.getPath()) {
countDownLatch.countDown();
} else if (Event.EventType.NodeDataChanged == watchedEvent.getType()) {
System.out.println("ReGet Data:" + zooKeeper.getData(watchedEvent.getPath(), true, new Stat()));
}
}
} catch (Exception e) {
e.toString();
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
zooKeeper = new ZooKeeper("192.168.2.192:2181", 5000, new ZkDemon4java());
System.out.println("zk status:" + zooKeeper.getState());
countDownLatch.await();
String path = "/zk_book";
System.out.println("get data result:" + zooKeeper.getData(path, true, new Stat()));
zooKeeper.setData(path, "123".getBytes(), -1);
zooKeeper.setData(path, "123".getBytes(), -1);
Thread.sleep(Integer.MAX_VALUE);
}
}
结果：
zk status:CONNECTING
receive wathedEvent:WatchedEvent state:SyncConnected type:None path:null
get data result:[B@73f792cf
receive wathedEvent:WatchedEvent state:SyncConnected type:NodeDataChanged path:/zk_book
ReGet Data:[B@20d91839
receive wathedEvent:WatchedEvent state:SyncConnected type:NodeDataChanged path:/zk_book
ReGet Data:[B@2e1ee252
getData(异步)
略
下面的api如果异步比较简单就都省略
setData
package com.xh.zk.javaapi;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
/**
* Created by root on 3/28/18.
*/
public class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static ZooKeeper zk;
public void process(WatchedEvent watchedEvent) {
if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
if (Event.EventType.None == watchedEvent.getType() && watchedEvent.getPath() == null) {
countDownLatch.countDown();
}
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
String connection = "192.168.2.192:2181";
String path = "/zk_book";
zk = new ZooKeeper(connection, 5000, new ZkDemon4java());
countDownLatch.await();
zk.create(path, "hello".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
byte[] get_data = zk.getData(path, null, null);
Stat set_stat = zk.setData(path, "456".getBytes(), -1);
System.out.println(set_stat.getCzxid() + "," + set_stat.getMzxid() + "," + set_stat.getVersion());
Stat set_stat2 = zk.setData(path, "qqq".getBytes(), set_stat.getVersion());
System.out.println(set_stat2.getCzxid() + "," + set_stat2.getMzxid() + "," + set_stat2.getVersion());
Stat set_stat3 = zk.setData(path, "aaa".getBytes(), set_stat.getVersion());
System.out.println(set_stat3.getCzxid() + "," + set_stat3.getMzxid() + "," + set_stat3.getVersion());
Thread.sleep(Integer.MAX_VALUE);
}
}
结果：
13,14,1
13,15,2
Exception in thread "main" org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for /zk_book
Version传入-1表示非原子操作，每次修改数据，版本号就递增，不一致就报错
exists
package com.xh.zk.javaapi;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
/**
* Created by root on 3/28/18.
*/
public class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static ZooKeeper zk;
static String path = "/zk_book";
public void process(WatchedEvent watchedEvent) {
try {
if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
if (Event.EventType.None == watchedEvent.getType() && watchedEvent.getPath() == null) {
countDownLatch.countDown();
} else if (Event.EventType.NodeCreated == watchedEvent.getType()) {
System.out.println(watchedEvent.getPath() + " NodeCreated");
zk.exists(path, true);
} else if (Event.EventType.NodeDataChanged == watchedEvent.getType()) {
System.out.println(watchedEvent.getPath() + " NodeDataChanged");
zk.exists(path, true);
} else if (Event.EventType.NodeDeleted == watchedEvent.getType()) {
System.out.println(watchedEvent.getPath() + " NodeDeleted");
zk.exists(path, true);
}
}
} catch (Exception e) {
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
String connection = "192.168.2.192:2181";
zk = new ZooKeeper(connection, 5000, new ZkDemon4java());
countDownLatch.await();
zk.exists(path, true);
zk.create(path, "hello".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zk.create(path + "/abc", "abc".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zk.setData(path, "world".getBytes(), -1);
zk.delete(path + "/abc", -1);
zk.delete(path, -1);
Thread.sleep(Integer.MAX_VALUE);
}
}
结果：
/zk_book NodeCreated
/zk_book NodeDataChanged
/zk_book NodeDeleted
无论节点是否存在，都可用exists监听，对该节点增删改都会通知，但是不会通知其子节点的操作
addAuthInfo
创建一个需要权限访问的节点
public class ZkDemon4java implements Watcher {
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static ZooKeeper zk_auth;
static String path = "/zk_book";
public void process(WatchedEvent watchedEvent) {
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
String connection = "192.168.2.192:2181";
zk_auth = new ZooKeeper(connection, 5000, new ZkDemon4java());
zk_auth.addAuthInfo("digest", "foo:true".getBytes());
zk_auth.create(path, "hello".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL);
Thread.sleep(Integer.MAX_VALUE);
}
}
创建几个客户端访问之前的节点，分别是 ：没有权限信息，错误权限信息，正确权限信息
/**
* no auth info
*/
zk_1 = new ZooKeeper(connection, 5000, new Zk1());
zk_1.getData(path, false, null);
Thread.sleep(Integer.MAX_VALUE);
结果：
Exception in thread "main" org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /zk_book
/**
* wrong auth info
*/
zk_2 = new ZooKeeper(connection, 5000, new Zk2());
zk_2.addAuthInfo("digest", "foo:true121".getBytes());
zk_2.getData(path, false, null);
Thread.sleep(Integer.MAX_VALUE);
结果：
Exception in thread "main" org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /zk_book
/**
* right auth info
*/
zk_3 = new ZooKeeper(connection, 5000, new Zk3());
zk_3.addAuthInfo("digest", "foo:true".getBytes());
byte[] zk_3Data = zk_3.getData(path, false, null);
System.out.println(zk_3Data);
Thread.sleep(Integer.MAX_VALUE);
结果：
[B@37bba400