zk 原生API 连接

对zk 操作的方式有一下几种：

• 基于java的 shell命令，zkCli.sh
• 原生的java api
• zkClient
• Curator
• spring cloud zookeeper

原生API

代码展示

• ZKConstant 常量

public class ZKConstant {
    public static final String CONNET_STR = "192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181";
    public static final int SESSION_TIMEOUT = 5000;
}

• CreateSession

public class CreateSession_API {
    private static ZooKeeper zk1;
    private static CountDownLatch connectSemaphore = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
//        createSession();
        createSessionWithSID();
    }

    public static void createSession() throws Exception {
        //Zookeeper是API提供的1个类，我们连接zk集群，进行相应的znode操作，都是通过ZooKeeper的实例进行，这个实例就是zk client，和命令行客户端是同样的角色
        //Zookeeper实例的创建需要传递3个参数
        //connectString 代表要连接zk集群服务，通过逗号分隔
        // 注册watcher事件
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR, ZKConstant.SESSION_TIMEOUT, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
//                这个方法只会调用一次，在这个session建立完成调用
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    connectSemaphore.countDown();
                    System.out.println("event:" + watchedEvent);
                    System.out.println("receive session established.");
                }
            }
        });
        System.out.println(zk1.getState());
        connectSemaphore.await();
        System.out.println("zk session established");
    }

    // 重复使用上次session, 利用sessionId和passwd
    public static void createSessionWithSID() throws Exception {
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR, ZKConstant.SESSION_TIMEOUT,
                new MyWatcher());
        connectSemaphore.await();
        long sessionId = zk1.getSessionId();
        byte[] passwd = zk1.getSessionPasswd();

        zk1 = new ZooKeeper(ZKConstant.CONNET_STR, ZKConstant.SESSION_TIMEOUT,
                new MyWatcher(),
                1l, "test".getBytes());

        zk1 = new ZooKeeper(ZKConstant.CONNET_STR, ZKConstant.SESSION_TIMEOUT,
                new MyWatcher(),
                sessionId,
                passwd);
        Thread.sleep(Integer.MAX_VALUE);
    }

    static class MyWatcher implements Watcher {
        @Override
        public void process(WatchedEvent watchedEvent) {
//            只注册一次
            System.out.println("receive watched event:" + watchedEvent);
            if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
                connectSemaphore.countDown();
            }
        }
    }
}

• CreateNode

public class CreateNode_API {
    private static ZooKeeper zk1;
    private static CountDownLatch connectSemaphore = new CountDownLatch(1); // 同步计数器
    public static void main(String[] args) throws Exception {
//        createNodeASync();
        createNodeSync();
    }
    public static void createNodeSync() throws Exception {
        ZooKeeper zookeeper = new ZooKeeper(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectSemaphore.await();
        // 创建临时节点
        String path1 = zookeeper.create("/zk-test-ephemeral-",
                "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL);
        System.out.println("Success create znode: " + path1);

        String path2 = zookeeper.create("/zk-test-ephemeral-",
                "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("Success create znode: " + path2);
    }
    public static void createNodeASync() throws Exception {
        ZooKeeper zk1 = new ZooKeeper(ZKConstant.CONNET_STR, ZKConstant.SESSION_TIMEOUT,
                new MyWatcher());
        connectSemaphore.await();
        // 创建临时节点
        zk1.create("/zk-test-eph-", "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                new IStringCallback(), "I am context1.");
        zk1.create("/zk-test-eph-", "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                new IStringCallback(), "i am context2");
        //  创建临时有序节点
        zk1.create("/zk-test-eph-", "".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                new IStringCallback(), "i am context3");
        Thread.sleep(Integer.MAX_VALUE);
    }
    static class MyWatcher implements Watcher {
        @Override
        public void process(WatchedEvent watchedEvent) {
//            建立连接成功回调
            if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                connectSemaphore.countDown();
            }
        }
    }
    // 创建节点成功回调
    static class IStringCallback implements AsyncCallback.StringCallback {
        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            System.out.println("create path result: [" + rc + "," + path + "," + ctx + ", real path name:" + name);
        }
    }
}

• DeleteNode

public class DeleteNode_API {
   private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
   private static ZooKeeper zk;
   public static void main(String[] args) throws Exception {
       deleteNodeSync();
   }
   public static void deleteNodeSync() throws Exception {
       String path = "/zk_book";
       zk = new ZooKeeper(ZKConstant.CONNET_STR, ZKConstant.SESSION_TIMEOUT,
               new DeleteWatcher());
       connectedSemaphore.await();
       zk.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
       zk.delete(path, -1);
       Thread.sleep(Integer.MAX_VALUE);
   }
   static class DeleteWatcher implements Watcher {
       @Override
       public void process(WatchedEvent watchedEvent) {
           if (Event.KeeperState.SyncConnected == watchedEvent.getState() && watchedEvent.getPath() == null) {
               connectedSemaphore.countDown();
           }
       }
   }
}

• ExistsNode

public class ExistsNode_API {
    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(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectedSemaphore.await();
//         对path 路劲进行监听
        zk.exists(path, true);
        zk.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.setData(path, "123".getBytes(), -1);
        zk.create(path+"/c1", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.setData(path + "/c1", "000".getBytes(), -1);
        zk.delete(path + "/c1", -1);
        zk.delete(path, -1);
        Thread.sleep(Integer.MAX_VALUE);
    }
    static class MyWatcher implements Watcher {
        public void process(WatchedEvent watchedEvent) {
            try {
                if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
                    if (Event.EventType.None == watchedEvent.getType() && null == watchedEvent.getPath()) {
                        connectedSemaphore.countDown();
                    } else if (Event.EventType.NodeCreated == watchedEvent.getType()) {
                        System.out.println("node (" + watchedEvent.getPath() + ") created ");
                        zk.exists(watchedEvent.getPath(), true);
                    } else if (Event.EventType.NodeDeleted == watchedEvent.getType()) {
                        System.out.println("node (" + watchedEvent.getPath() + ") deleted ");
                        zk.exists(watchedEvent.getPath(), true);
                    } else if (Event.EventType.NodeDataChanged == watchedEvent.getType()) {
                        System.out.println("node (" + watchedEvent.getPath() + ") dataChanged");
                        zk.exists(watchedEvent.getPath(), true);
                    }
                }
            } catch (Exception e) {
            }
        }
    }
}

• GetData

public class GetData_API {
    private static ZooKeeper zk1;
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    private static Stat stat = new Stat();
    static String path = "/zk-book";
    public static void main(String[] args) throws Exception {
        sync_setData();
    }
    public static void sync_setData() throws Exception {
        String path = "/zk-book";
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectedSemaphore.await();
        Stat stat = zk1.setData(path, "haha".getBytes(), -1);
        System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "," +
                stat.getVersion());
        Stat stat2 = zk1.setData(path, "haha".getBytes(), -1);
        System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "," +
                stat.getVersion());
        try {
            // 指定version， 需要正确的version才可以通过
            zk1.setData(path, "456".getBytes(), stat.getVersion());
        } catch (KeeperException e) {
            System.out.println("Error: " + e.code() + "," + e.getMessage());
        }
        Thread.sleep(Integer.MAX_VALUE);
    }

    public static void async_setData() throws Exception {
        String path = "/zk-book";
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectedSemaphore.await();
        zk1.setData(path, "456".getBytes(), -1, new AsyncCallback.StatCallback() {
            public void processResult(int i, String s, Object o, Stat stat) {
                if (i == 0) {
                    System.out.println("SUCCESS");
                }
            }
        }, null);
        Thread.sleep(Integer.MAX_VALUE);
    }
    public static void sync_getChildren() throws Exception {
        String path = "/zk-book";
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectedSemaphore.await();
//        zk1.delete(path+"/c1", 0);
        zk1.delete(path, 0);
        zk1.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        zk1.create(path + "/c1", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        List<String> childrenList = zk1.getChildren(path, true);
        System.out.println(childrenList);
        zk1.create(path + "/c2", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL);
        Thread.sleep(Integer.MAX_VALUE);
    }

    public static void async_getChildren() throws Exception {
        String path = "/zk-book";
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectedSemaphore.await();
        zk1.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        zk1.create(path + "/c1", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        // 只会响应一次
        zk1.getChildren(path, true, new AsyncCallback.Children2Callback() {
            public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
                System.out.println("Get Children znode result: [response code: " + rc + ", param path: " + path
                        + ", ctx: " + ctx + ", children list: " + children + ", stat: " + stat);
            }
        }, null);
        zk1.create(path + "/c2", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL);
        Thread.sleep(Integer.MAX_VALUE);
    }
    public static void sync_getData() throws Exception {
        String path = "/zk-book";
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectedSemaphore.await();
        zk1.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println(new String(zk1.getData(path, true, stat)));
        System.out.println(stat.getCzxid() + "," + stat.getMzxid() + "," + stat.getVersion());
        zk1.setData(path, "456".getBytes(), -1);
        Thread.sleep(Integer.MAX_VALUE);
    }
    public static void async_getData() throws Exception {
        zk1 = new ZooKeeper(ZKConstant.CONNET_STR,
                ZKConstant.SESSION_TIMEOUT, //
                new MyWatcher());
        connectedSemaphore.await();
        zk1.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        zk1.getData(path, true, new IDataCallback(), null);
        zk1.setData(path, "456".getBytes(), -1);
        Thread.sleep(Integer.MAX_VALUE);
    }
    static class MyWatcher implements Watcher {
        public void process(WatchedEvent watchedEvent) {
            if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
                if (Event.EventType.None == watchedEvent.getType() && null == watchedEvent.getPath()) {
                    connectedSemaphore.countDown();
                } else if (watchedEvent.getType() == Event.EventType.NodeDataChanged) {
                    try {
                        zk1.getData(watchedEvent.getPath(), true, new IDataCallback(), null);
                    } catch (Exception e) {
                    }
                }
            }
        }
    }
    static class IDataCallback implements AsyncCallback.DataCallback {
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            System.out.println(rc + ", " + path + ", " + new String(data));
            System.out.println("--" + stat.getCzxid() + "," +
                    stat.getMzxid() + "," +
                    stat.getVersion());
        }
    }
}

PS: 若你觉得可以、还行、过得去、甚至不太差的话，可以“关注”或者“点赞”一下，就此谢过!