zookeeper Java API

zookeeper构造方法

1.png

connectString：zk服务器列表，组成：主机+端口
sessionTimeout：会话超时时间（单位毫秒），zookeeper在服务器和客户端之间创建TCP长连接。至于客户端是否有效，通过心跳机制保证，那么一旦超过了我们的参数sessionTimeout，那么这个连接就会失效。
watcher：事件监听器，当监听的事件触发的时候，会发送通知到客户端。
sessionId、sessionPasswd：会话ID和会话密钥，这两个参数保证了会话的唯一性。
canBeReadOnly：就是设置当前会话是否为只读的方式。

public class ZkTest {

    class MyWatcher implements Watcher {

        @Override
        public void process(WatchedEvent we) {
            System.out.println(we.getState());
        }
    }
    
    public static void main(String[] args) throws Exception {
        MyWatcher mw = new ZkTest().new MyWatcher();
        ZooKeeper zk = new ZooKeeper("ip1:2181,ip2:2181,ip3:2181", 
                    5000, mw);
        List<String> childrenList = zk.getChildren("/", mw);
        System.out.println(childrenList);
        System.out.println("获取会话ID：" + zk.getSessionId());
        System.out.println("获取会话超时时间：" + zk.getSessionTimeout());
    }
}

结果：

SyncConnected
[myNode, zookeeper, myNode2]
会话ID：245189035829100546
会话超时时间：5000

创建节点

2.png

path：节点创建的路径
data[] ：初始内容的字节数组
acl ：权限控制策略
createMode：节点类型：
   EPHEMERAL 临时节点
   EPHEMERAL_SEQUENTIAL 临时顺序节点
   PERSISTENT 持久节点
   PERSISTENT_SEQUENTIAL 持久顺序节点
cb：异步回调函数，用于节点创建完毕后回调，需要实现StringCallbacl接口，重写processResult方法
ctx：在回调方法执行时，传递一个上下文信息对象

zk.create("/myNode2", "zhangjl".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

删除节点

3.png

path：数据节点路径
version：数据节点的版本
cb：异步回调函数
crx：传递上下文信息对象

zk.delete("/myNode4", -1);

读取数据

4.png

path：数据节点路径
watcher：注册Watcher，如果子节点发生变化，那么就通知客户端
cb：异步回调函数
ctx：传递上下文信息对象

List<String> list = zk.getChildren("/", new Watcher() {

            @Override
            public void process(WatchedEvent event) {
                if (KeeperState.SyncConnected == event.getState()) {
                    System.out.println("KeeperState.SyncConnected" + KeeperState.SyncConnected);
                }
            }
        });

5.png

path：数据节点路径
watcher：注册Watcher，如果子节点发生变化，那么就通知客户端
stat：传入旧数据节点的节点信息，服务端响应会替换为新的节点信息

        byte[] arr = zk.getData("/myNode3", new Watcher() {

            @Override
            public void process(WatchedEvent event) {
                if (KeeperState.SyncConnected == event.getState()) {
                    System.out.println("KeeperState.SyncConnected" + KeeperState.SyncConnected);
                }
            }
        }, null);
        
        System.out.println(new String(arr));

更新数据

6.png

path：数据节点路径
data[] ：更新数据的字节数组
version：节点的数据版本
cb：异步回调函数
ctx：用于传递上下文的信息对象

zk.setData("/myNode3", "瓜娃子".getBytes(), -1);

检测节点是否存在

7.png

path：数据节点路径
watcher：注册Watcher，如果子节点发生变化，那么就通知客户端
cb：异步回调函数
ctx：传递上下文信息的对象

Stat stat = zk.exists("/myNode3", true);

权限控制

8.png

MyWatcher mw = new ZkTest().new MyWatcher();
        ZooKeeper zk = new ZooKeeper("ip1:2181,ip2:2181,ip3:2181", 
                    5000, mw);
        zk.addAuthInfo("digest", "sunpy:123456789".getBytes());
        zk.create("/myNode4", "4".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
       
        ZooKeeper zk2 = new ZooKeeper("ip1:2181,ip2:2181,ip3:2181", 5000,
                mw);
        zk2.addAuthInfo("digest", "sunpy:123456789".getBytes());
        System.out.println(zk2.getData("/myNode4", false, null));
        
        ZooKeeper zk3 = new ZooKeeper("ip1:2181,ip2:2181,ip3:2181", 5000,
                mw);
        zk3.addAuthInfo("digest", "sunpy:1234567890".getBytes());
        System.out.println(zk3.getData("/myNode4", false, null));

结果：

Exception in thread "main" org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /myNode4
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:113)
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
    at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:1212)
    at cn.spy.zk.test.ZkTest.main(ZkTest.java:39)