当前位置:   article > 正文

zookeeper 常见客户端介绍和使用 zkCli、自带API、 zkClient、Curator_zk客户端

zk客户端

一、Zookeeper的命令行使用

ZooKeeper解压后,在其bin目录下包含着常用的程序,例如 zkServer.sh zkCli.sh
我们使用zkCli.sh 就可以通过命令行使用Zookeeper客户端
连接zookeeper服务器
连接后输入help就可以查看所有命令和使用方式的说明了

#对于本地默认端口 则可以直接 ./zkCli.sh 
# -server 指定服务地址和端口
[root@localhost bin]# ./zkCli.sh -server localhost:15881
  • 1
  • 2
  • 3

创建节点命令
create [-s][-e] path data acl
-s或-e分别指定节点特性,顺序或临时节点,若不指定,则创建持久节点;acl⽤来进⾏权限控制。

# 创建顺序节点
[zk: localhost:15881(CONNECTED) 0] create -s /zk-test dataContent1111
Created /zk-test0000000007 

# 创建临时节点,临时节点在会话结束后由就会被自动删除
[zk: localhost:15881(CONNECTED) 0] create -e /zk-temp data222
Created /zk-temp

# 创建永久节点
[zk: localhost:15881(CONNECTED) 2] create /zk-test-permanent data333
Created /zk-test-permanent
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

读取节点
可以使用ls查看子节点列表,使用 get 命令查看节点的内容

# 使用 ls 命令查看子节点
[zk: localhost:15881(CONNECTED) 4] ls /
[lg-PERSISTENT, zk-premament, zk-temp, zk-test-permanent, zk-test0000000000, zk-test0000000007, zookeeper]

# 使用 get 命令查看节点内容 get -s 则可以附加打印节点状态信息
[zk: localhost:15881(CONNECTED) 6] get /zk-temp
data222

# stat 命令查看节点状态
[zk: localhost:15881(CONNECTED) 0] stat /zk-temp
cZxid = 0x30000000a
ctime = Wed Jul 05 10:48:44 CST 2023
mZxid = 0x30000000a
mtime = Wed Jul 05 10:48:44 CST 2023
pZxid = 0x30000000a
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x100008d52290003
dataLength = 7
numChildren = 0

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

更新节点内容
命令:set path data [version] version表示数据版本,在zookeeper中,节点的数据是有版本概念的,这个参数⽤于指定本次更新操作是基于Znode的哪⼀个数据版本进⾏的,如果版本和最新版本对不上则会更新失败,这样可以防止覆盖最新写入的数据。

set /zk-premament 666
  • 1

删除节点
删除命令 **delete path [version]**** **如果删除的节点包含子节点,那么必须先删除子节点才能删除对应节点。

二、Zookeeper自带API的使用

2.1 引入API

通过Maven引入Zookeeper提供了java客户端API依赖,截至当前时间最新稳定版是 3.7.1

<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.7.1</version>
</dependency>
  • 1
  • 2
  • 3
  • 4
  • 5

2.1 API简单使用

/**
 * zookeeper API 简单使用
 *
 * @author liuyp
 */
public class ZookeeperApiSimpleTest {
    //是否完成连接的建立
    static boolean connected = false;
    static Object lock = new Object();

    //zookeeper实例对象
    static ZooKeeper zooKeeper;

    //定义Watcher的回调 它会收到客户端状态变化的通知,也可以收到节点事件的通知
    static Watcher watcherProcess = (watchedEvent) -> {
        //客户端连接成功状态通知
        if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected && !connected) {
            System.out.println("watcher回调:客户端连接上线");
            synchronized (lock) {
                //连接成功就通知方法返回
                connected = true;
                lock.notifyAll();
            }
        }

        //子节点列表变化通知
        if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
            try {
                //获取最新的子节点,并重新开启watch
                List<String> children = zooKeeper.getChildren(watchedEvent.getPath(), true);
                System.out.println("watcher回调:子节点变化通知 节点:" + watchedEvent.getPath() + " 的最新子节点:" + children);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //节点内容变更事件
        if (watchedEvent.getType() == Watcher.Event.EventType.NodeDataChanged) {
            try {
                byte[] data = zooKeeper.getData(watchedEvent.getPath(), false, null);
                System.out.println("watcher回调:节点数据变化通知 节点:" + watchedEvent.getPath() + " 内容为:" + new String(data));
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //节点删除通知
        if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted) {
            System.out.println("watcher回调:节点被删除通知:" + watchedEvent.getPath());
        }

    };


    /**
     * demo测试入口
     *
     * @param args
     * @throws IOException
     * @throws InterruptedException
     * @throws KeeperException
     */
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        //同步的方式建立会话
        createSession();
        //测试创建节点,先删除上一次创建的
        createZNode();
        //获取节点数据
        getZNodeData();
        //更新节点数据
        updateZNodeData();
        //删除节点
        deleteZNode();
    }


    /**
     * 一、创建会话
     * 创建Zookeeper会话初始化Zookeeper对象
     * 这里改成同步执行,连接上了方法才返回
     */
    public synchronized static void createSession() throws IOException, InterruptedException {
        //可以配置多个地址客户端会随机连接例如 192.168.188.130:15881,192.168.188.130:15882
        String connectString = "192.168.188.130:15881";
        //会话超时时间 单位是毫秒
        int sessionTimeout = 5000;
        //执行结果立即返回,后台异步建立连接。watcherProcess
        zooKeeper = new ZooKeeper(connectString, sessionTimeout, watcherProcess);
        if (connected) {
            return;
        }
        //如果没执行完,就让出锁进入等待状态,等待出结果后被唤醒
        synchronized (lock) {
            lock.wait();
        }
    }


    /**
     * 二、创建znode
     */
    public static void createZNode() throws KeeperException, InterruptedException {
        //创建一个测试的公共节点,后续都在这个节点下面测试,并且给他加一个watch
        String testParentNodePath = "/zookeeperApi";
        if (zooKeeper.exists(testParentNodePath,false)==null){
            zooKeeper.create(testParentNodePath, "父节点".getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        //添加监听 exist&getData
        zooKeeper.addWatch(testParentNodePath, AddWatchMode.PERSISTENT_RECURSIVE);
        zooKeeper.getChildren(testParentNodePath, true);

        /**
         * path:节点创建路径
         * data[] :字节数组格式保存到节点的数据
         * acl:节点ACL权限设置
         * createMode:创建的节点类型。PERSISTENT:持久节点 EPHEMERAL临时节点 ,还有临时顺序节点,持久顺序节点
         */
        String zNodePersistent = zooKeeper.create(
                testParentNodePath + "/persistent",
                "持久节点内容".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        String zNodeEphemeralSequential = zooKeeper.create(
                testParentNodePath + "/ephemeralSequential",
                "临时顺序节点内容".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        String zNodeEphemeral = zooKeeper.create(
                testParentNodePath + "/persistentEphemeral",
                "临时节点内容".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

    }


    /**
     * 三、获取节点数据
     */
    public static void getZNodeData() throws KeeperException, InterruptedException {
        String testParentNodePath = "/zookeeperApi";
        byte[] data = zooKeeper.getData(testParentNodePath, false, null);
        System.out.println("节点:" + testParentNodePath + " 内容为:" + new String(data));
    }

    /**
     * 三、更新节点数据
     */
    public static void updateZNodeData() throws KeeperException, InterruptedException {
        String testParentNodePath = "/zookeeperApi";
        zooKeeper.setData(testParentNodePath, ("新数据" + Math.random()).getBytes(), -1);
    }


    /**
     * 四、删除znode
     */
    public static void deleteZNode() throws KeeperException, InterruptedException {
        String testParentNodePath = "/zookeeperApi";
        zooKeeper.delete(testParentNodePath + "/persistent", -1);
    }


}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145
  • 146
  • 147
  • 148
  • 149
  • 150
  • 151
  • 152
  • 153
  • 154
  • 155
  • 156
  • 157
  • 158
  • 159
  • 160
  • 161
  • 162
  • 163
  • 164
  • 165
  • 166
  • 167

三、Zookeeper三方客户端zkClient的使用

项目地址:https://github.com/sgroschupf/zkclient/issues
zkClient是git上的一个开源的zookeeper的java客户端项目,是对zookeeper原生API的封装,使得其更易用了。
优势:1. session重连 2.watch重主策 3.递归删除/添加节点
注意:项目最新更新日期是2018年,上生产使用前需要考虑漏洞问题。

3.1 引入依赖

<!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
<dependency>
  <groupId>com.101tec</groupId>
  <artifactId>zkclient</artifactId>
  <version>0.11</version>
</dependency>

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

3.2 简单的使用案例

public class ZkClientTest {

    static CountDownLatch countDownLatch = new CountDownLatch(1);

    public static void main(String[] args) throws InterruptedException {
        String testzkClientPath = "/zkClientAPI";

        //建立连接,这里是同步的方式
        String connectString = "192.168.188.130:15881";
        ZkClient zkClient = new ZkClient(connectString);

        //创建节点,zkClient支持递归创建,没有父节点会自动创建对应的父节点
        zkClient.createPersistent(testzkClientPath + "/persistent", true);
        zkClient.createPersistent(testzkClientPath + "/persistent_readyDelete", true);

        //删除节点 zkClient支持自动删除节点下的子节点
        zkClient.delete(testzkClientPath + "/persistent_readyDelete", -1);

        //获取子节点
        List<String> children = zkClient.getChildren(testzkClientPath);
        System.out.println("读取节点:" + testzkClientPath + " 子节点:" + children);

        //监听事件注册
        //注册子节点变更事件
        zkClient.subscribeChildChanges(testzkClientPath, (path, childNodeList) -> {
            System.out.println("节点子节点监听事件通知:节点:" + path + " 最新子节点:" + childNodeList);
        });
        //注册节点数据变更事件
        zkClient.subscribeDataChanges(testzkClientPath, new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("节点数据监听事件通知:节点:" + s + " 最新数据:" + o);
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                System.out.println("节点数据监听事件通知:节点:" + s + " 已删除");
            }
        });

        //写入节点数据
        zkClient.writeData(testzkClientPath, System.currentTimeMillis() + "写入数据");

        //获取节点数据
        Object readDataResult = zkClient.readData(testzkClientPath);
        System.out.println("读取节点数据:" + testzkClientPath + " : " + readDataResult);

        //删除节点
        zkClient.deleteRecursive(testzkClientPath);

        //阻塞最后的结束程序
        countDownLatch.await();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54

四、Curator 客户端框架

项目地址:https://github.com/apache/curator
最开始由 netflix 在github上开源,2013年成为apache顶级项目,至今仍在更新
和ZkClient一样,Curator解决了很多细节的底层工作,包括连接重连、watch自动重新注册
节点不存在异常等,并且提供了基于fluent编程风格的支持

4.1 引入依赖

<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-framework -->
<dependency>
  <groupId>org.apache.curator</groupId>
  <artifactId>curator-framework</artifactId>
  <version>5.5.0</version>
</dependency>

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

4.2 简单使用案例

/**
 * Curator 是Netflix公司开源的一套ZooKeeper客户端框架
 * 和ZkClient一样,Curator解决了很多细节的底层工作,包括连接重连、watch自动重新注册
 * 节点不存在异常等,并且提供了基于fluent编程风格的支持
 * @author liuyp
 */
public class CuratorTest {

    public static void main(String[] args) throws Exception {
        //连接信息,多个连接使用逗号分隔
        String connectString = "192.168.188.130:15881";

        /**
         * 一、发起连接
         *
         * RetryPolicy重连策略 默认提供三种重连策略
         * 1、ExponentialBackoffRetry(基于backoff的重连策略)重新尝试一定次数,并增加重试之间的睡眠时间
         * 2、RetryNTimes(重连N次策略)
         * 3、RetryForever(永远重试策略)
         *
         * 创建连接 CuratorFramework
         * 1、通过CuratorFrameworkFactory.newClient 底层是CuratorFrameworkFactory.build
         * 2、直接通过 CuratorFrameworkFactory.build
         *
         * 启动连接 CuratorFramework.start()
         */
        int baseSleepTimeMs=1000; //重试之间等待的初始时间
        int maxRetries=5;//最大重试次数
        int maxSleepMs=5000;//每次重试的最大睡眠时间 如果算出来的sleepMs超过这个时间,则采用maxSleepMs
        //重试间隔时间: baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1)));
        RetryPolicy retryPolicy=new ExponentialBackoffRetry(baseSleepTimeMs,maxRetries,maxSleepMs);

        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(10000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .namespace("curatorAPI") //加上这个以后,所有路径都是以这个路径为根路径
                .build();

        client.start();

        System.out.println("**********客户端已启动**********");

        /**
         * 二、创建节点
         * 1、默认创建内容为空的永久节点
         * 2、设置节点内容和原生一样,使用字节数组
         * 3、可以使用 creatingParentsIfNeeded 方法自动创建父节点,避免需要递归判断父节点是否存在
         */
        client.create()
                .creatingParentContainersIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath("/tempNode/create","临时节点".getBytes(StandardCharsets.UTF_8));

        /**
         * 三、测试增加监听
         * 1、监听类型 PERSISTENT_RECURSIVE 会循环监听注册节点和其子节点的数据变化和是否存在
         */
        CuratorWatcher curatorWatcher=(watchevent)->{
            System.out.println("[监听通知:]"+"节点:"+watchevent.getPath()+" "+watchevent.getType());
        };
        client.watchers().add().withMode(AddWatchMode.PERSISTENT_RECURSIVE).usingWatcher(curatorWatcher).forPath("/tempNode");
        client.create().forPath("/tempNode/watcher");

        /**
         * 三、读取&修改节点数据 并获取状态数据
         */
        Stat stat=new Stat();
        byte[] bytes = client.getData().storingStatIn(stat).forPath("/tempNode/create");
        System.out.println("读取节点数据:"+new String(bytes,StandardCharsets.UTF_8));
        System.out.println("读取节点状态:"+stat.toString());
        client.setData().forPath("/tempNode/create","节点/tempNode/create的新数据".getBytes(StandardCharsets.UTF_8));

        /**
         * 四、删除节点
         */
        client.delete().forPath("/tempNode/watcher");
        client.delete().forPath("/tempNode/create");
        client.delete().forPath("/tempNode");


    }

}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/weixin_40725706/article/detail/606578
推荐阅读
相关标签
  

闽ICP备14008679号