Hadoop-30 ZooKeeper集群 JavaAPI 客户端 POM Java操作ZK 监听节点 监听数据变化 创建节点 删除节点

章节内容

上节我们完成了如下的内容：

• ZK创建节点：永久、顺序、临时
• ZK读取节点：列出、查看、更新
• ZK删除节点

背景介绍

这里是三台公网云服务器，每台 2C4G，搭建一个Hadoop的学习环境，供我学习。
之前已经在 VM 虚拟机上搭建过一次，但是没留下笔记，这次趁着前几天薅羊毛的3台机器，赶紧尝试在公网上搭建体验一下。

• 2C4G 编号 h121
• 2C4G 编号 h122
• 2C2G 编号 h123

新建Java工程

这里就跳过了，新建一个Maven工程。

POM文件

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.8.4</version>
</dependency>
<dependency>	
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.11</version>
</dependency>
1
2
3
4
5
6
7
8
9
10

创建回话

public class Test01 {

    public static void main(String[] args) {
        ZkClient zkClient = new ZkClient("h121.wzk.icu:2181");
        System.out.println("ZooKeeper session created.");
    }

}
1
2
3
4
5
6
7
8

运行程序，结果如下：

创建节点

public class Test01 {

    public static void main(String[] args) {
        ZkClient zkClient = new ZkClient("h121.wzk.icu:2181");
        System.out.println("ZooKeeper session created.");
        // true 则可以递归创建目录
        zkClient.createPersistent("/wzk-java/temp", true);
        System.out.println("ZooKeeper craete ZNode");

    }

}
1
2
3
4
5
6
7
8
9
10
11
12

运行程序，结果如下：

删除节点

public class Test01 {

    public static void main(String[] args) {
        ZkClient zkClient = new ZkClient("h121.wzk.icu:2181");
        System.out.println("ZooKeeper session created.");
        // true 则可以递归创建目录
        zkClient.createPersistent("/wzk-java/temp", true);
        System.out.println("ZooKeeper create ZNode");
        // 删除数据
        zkClient.deleteRecursive("/wzk-java/temp");
        System.out.println("ZooKeeper delete recursive");
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

运行程序，结果如下：

监听节点

public class Test02 {

    public static void main(String[] args) throws Exception {
        ZkClient zkClient = new ZkClient("h121.wzk.icu:2181");
        // 监听器 不会对当前目录进行监控 只会监听子目录变化！！！
        zkClient.subscribeChildChanges("/wzk-data", new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                System.out.println("parentPath: " + parentPath + ", " + currentChilds);
            }
        });

        // 添加数据
        zkClient.createPersistent("/wzk-data/test-data", true);
        Thread.sleep(1000);

        // 删除数据
        zkClient.deleteRecursive("/wzk-data/test-data");
        Thread.sleep(1000);
    }

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

运行程序，结果如下：

监听数据

public class Test03 {

    public static void main(String[] args) throws Exception {
        ZkClient zkClient = new ZkClient("h121.wzk.icu:2181");
        // 序列化
        zkClient.setZkSerializer(new SerializableSerializer());

        // 判断节点是否存在
        final boolean exists = zkClient.exists("/wzk-data/test-data");
        if (!exists) {
            // 不存在则创建出来
            zkClient.createPersistent("/wzk-data/test-data", true);
            System.out.println("ZooKeeper create ZNode");
        }

        zkClient.subscribeDataChanges("/wzk-data/test-data", new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                System.out.println("数据改变: " + dataPath + ", " + data);
            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println("数据删除: " + dataPath);
            }
        });

        // 更新数据 出发监听器
        final Object o = zkClient.readData("/wzk-data/test-data");
        zkClient.writeData("/wzk-data/test-data", "更新了数据");
        Thread.sleep(1000);

        zkClient.deleteRecursive("/wzk-data/test-data");
        Thread.sleep(1000);
    }

}

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

运行程序，结果如下：