ZooKeeper常用命令以及一些JAVA API操作示例_zookeeper rmr

ZooKeeper客户端shell命令

1.进入客户端

zkCli.sh 
1

2.create创建节点

• 创建普通节点

create /test "123"'
1

’

• 创建临时节点,重新进入客户端节点会消失

  create -e /test/test1 “456”

  

• 创建顺序节点

  create -s /test/test1 'a'
  
  create -s /test/test2 'b'
  
  create -s /test/test3 'c'
  1
  2
  3
  4
  5

  

3.ls查看某路径的节点

ls /
1

4.get查看节点详细信息

get /test
1

• czxid-创建节点的事务zxid每次修改ZooKeeper状态都会收到⼀个zxid形式的时间戳，也就是ZooKeeper事务ID。事务ID是ZooKeeper中所有修改总的次序。每个修改都有唯⼀的zxid，如果zxid1⼩于zxid2，那么zxid1在zxid2之前发⽣
• ctime - znode被创建的毫秒数
• mzxid - znode最后更新的事务zxid
• mtime - znode最后修改的毫秒数
• pZxid-znode最后更新的⼦节点zxid
• cversion - znode⼦节点变化号，znode⼦节点修改次数
• dataversion - znode数据变化号
• aclVersion - znode访问控制列表的变化号
• ephemeralOwner- 如果是临时节点，这个是znode拥有者的session id。如果不是临时节点则是0

5.ls2 查看某一路径下的所有节点和信息

ls2 /
1

6.set 设置节点的值

set /test "01"
1

7.delete删除节点（没有字节的情况下）

delete /test/test10000000001
1

8.rmr 递归删除节点

rmr /test
1

9.history查看会话历史命令

history
1

10.help查看命令帮助

help
1

11.监听器的使用

11.1监听节点数据变化

在hadoop001主机上注册监听/test节点数据变化

get /test watch 
1

在hadoop002主机上修改/test节点的数据

set /test "2"
1

观察hadoop001主机收到数据变化的监听

11.2 监听节点的⼦节点变化

在hadoop001主机上注册监听/test节点的⼦节点变化

ls /test watch
1

在hadoop002主机/test节点上 创建⼦节点

 create /test/test01 "2"
1

观察hadoop001主机收到⼦节点变化的监听

Zookeeper的API使⽤

1.引入jar包

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>big_datacode</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>zk_opt</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.10</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.25</version>
    </dependency>


    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>2.13.0</version>
    </dependency>
    <!--https://mvnrepository.com/artifact/org.apache.curator/curator-recipes -->
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>2.13.0</version>
    </dependency>
    <dependency>
        <groupId>com.google.collections</groupId>
        <artifactId>google-collections</artifactId>
        <version>1.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.4.9</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.25</version>
    </dependency>
</dependencies>
<build>
<plugins>
    <!-- java编译插件 -->
    <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.1</version>
        <configuration>
            <source>1.8</source>
            <target>1.8</target>
            <encoding>UTF-8</encoding>
        </configuration>
    </plugin>
</plugins>
</build>
        </project>

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

2.节点操作

2.1创建节点

2.1.1创建持久节点、临时节点

public class zkOpt {
    public static void main(String[] args) throws Exception {
        //定义重试策略
        ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3, Integer.MAX_VALUE);

        //构建zk，连接客户端
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("hadoop001:2181,hadoop002:2181,hadoop003:2181", exponentialBackoffRetry);

        //启动客户端
        curatorFramework.start();

        //创建节点
        curatorFramework.create()
                .creatingParentContainersIfNeeded()
                .withMode(CreateMode.PERSISTENT) //PERSISTENT表示持久节点,EPHEMERAL表示临时节点
                .forPath("/testJAVA","helloWorld".getBytes());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

2.1.2 修改节点数据、删除节点、获取节点数据

        //更改节点值
        curatorFramework.setData().forPath("/testJAVA","haha".getBytes());
        //删除节点
        curatorFramework.delete().forPath("/test/test01");
		//递归删除节点
		  curatorFramework.delete()
                        .deletingChildrenIfNeeded()
                        .forPath("/test");
        //获取节点值
        byte[] bytes = curatorFramework.getData().forPath("/testJAVA");
        System.out.println(new String(bytes));
		//获取节点数据和stat信息
	 Stat statinfo = new Stat();
        byte[] node10 = curatorFramework.getData()
                .storingStatIn(statinfo)//获取stat信息存储到stat对象
                .forPath("/testJAVA");
        System.out.println("=====>该节点信息为：" + new String(node10));
        System.out.println("=====>该节点的数据版本号为：" +
                statinfo.getVersion());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2.1.3 监听机制

        try {
            String path = "treeCacheTest";
            //定义重试策略
            ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3, Integer.MAX_VALUE);

            //构建zk，连接客户端
            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("hadoop001:2181,hadoop002:2181,hadoop003:2181", exponentialBackoffRetry);

            //启动客户端
            curatorFramework.start();
            //如果存在path，删除路径
            Stat stat = curatorFramework.checkExists().forPath("/testJAVA" + path);
            if (stat != null) {
                curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath("/testJAVA" + path);
            }


            //创建节点
            curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/testJAVA"+path,path.getBytes());

            final TreeCache cache = new TreeCache(curatorFramework, "/testJAVA"+path);
            cache.getListenable().addListener((cli,event) -> {
                TreeCacheEvent.Type type = event.getType();
                switch (type){
                    case CONNECTION_RECONNECTED:
                        break;
                    case CONNECTION_SUSPENDED:
                        break;
                    case CONNECTION_LOST:
                        System.out.println("连接丢失");
                        break;
                    case NODE_ADDED:
                        System.out.println("添加节点：" + event.getData().getPath());
                        break;
                    case NODE_UPDATED:
                        System.out.println("更新节点");
                        break;
                    case NODE_REMOVED:
                        System.out.println("删除节点："+ event.getData().getPath());
                        break;
                    default:
                }
            });
            cache.start();
            //创建子节点/testJAVA/1
            curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/testJAVA"+path+"/testJAVA/1",path.getBytes());
            Thread.sleep(1000);

            //创建子节点/testJAVA/1/2
            curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/testJAVA"+path+"/testJAVA/1/2",path.getBytes());
            Thread.sleep(1000);

            //删除子节点/testJAVA/1/2
            curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath("/testJAVA"+path+ "/testJAVA/1/2");
            Thread.sleep(1000);

            //删除子节点/testJAVA/1
            curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath("/testJAVA"+path+"/testJAVA/1");
            //Thread.sleep(1000);

            Thread.sleep(Integer.MAX_VALUE);
            cache.close();

            //关闭客户端
            curatorFramework.close();

        }catch (Exception e) {
            //TODO: handle exception
        }
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

2.1.4 判断节点是否存在

        //判断接待你是否存在
        //连接字符串
        String connectString = "hadoop001:2181,hadoop002:2181,hadoop003:2181";
        //指定重试策略
        ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3, Integer.MAX_VALUE);
        //创建客户端
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString,exponentialBackoffRetry);
        //启动客户端
        client.start();

        Stat exisNodeStat =
                client.checkExists().forPath("/testJAVA");
        if (exisNodeStat == null){
            System.out.println("===>节点不存在");
        }
        if (exisNodeStat.getEphemeralOwner()>0){
            System.out.println("===>临时节点");
        }
        else{
            System.out.println("===>永久节点");
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21