Zookeeper学习笔记3

客户端的操作

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