二十二、Zookeeper的API代码应用

╰+攻爆jí腚メ 2021-09-02 10:02 293阅读 0赞

上篇文章主要讲述了Zookeeper的命令行操作，本文主要讲解Zookeeper的API代码应用，本文以代码为主，Zookeeper的部分相对来说都是比较简单的。关注专栏[《破茧成蝶——大数据篇》][Link 1]查看相关系列文章~

--------------------

**目录**

一、添加pom.xml文件

二、查看Zookeeper中ZNode

三、创建节点

四、获取节点信息

五、修改节点内容

六、查看结构体

七、删除节点

八、循环注册节点

九、完整代码

--------------------

# 一、添加pom.xml文件 #

在新建的Maven工程里面，添加如下依赖内容：

<dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.10</version>
        </dependency>
    </dependencies>

# 二、查看Zookeeper中ZNode #

/**
         * 查看ZNode中包含的节点
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void ls() throws KeeperException, InterruptedException {
            //默认回调函数
    //        List<String> children = zkCli.getChildren("/", true);
            //自定义回调函数
            List<String> children = zkCli.getChildren("/", e -> {
                System.out.println("自定义回调函数");
            });
    
            System.out.println("=====================================");
            for (String child: children) {
                System.out.println(child);
            }
            System.out.println("=====================================");
    
            Thread.sleep(Long.MAX_VALUE);
        }

# 三、创建节点 #

/**
         * 创建节点
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void create() throws KeeperException, InterruptedException {
            String s = zkCli.create("/test", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println(s);
    
            Thread.sleep(Long.MAX_VALUE);
        }

# 四、获取节点信息 #

/**
         * 获取节点信息
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void get() throws KeeperException, InterruptedException {
            byte[] data = zkCli.getData("/xzw", true, new Stat());
    
            String s = new String(data);
    
            System.out.println(s);
        }

# 五、修改节点内容 #

/**
         * 修改节点内容
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void set() throws KeeperException, InterruptedException {
            Stat stat = zkCli.setData("/xzw", "test".getBytes(), 0);
    
            System.out.println(stat.getDataLength());
        }

# 六、查看结构体 #

/**
         * 查看结构体
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void stat() throws KeeperException, InterruptedException {
            Stat exists = zkCli.exists("/xzw", false);
            if (exists == null) {
                System.out.println("节点不存在");
            } else {
                System.out.println(exists.getDataLength());
            }
        }

# 七、删除节点 #

/**
         * 删除节点
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void delete() throws KeeperException, InterruptedException {
            Stat exists = zkCli.exists("/xzw", false);
            if (exists != null)
                zkCli.delete("/xzw", exists.getVersion());
        }

# 八、循环注册节点 #

/**
         * 循环注册节点：当节点发生变化时，可以不停的通知
         * @throws KeeperException
         * @throws InterruptedException
         */
        public void register() throws KeeperException, InterruptedException {
            byte[] data = zkCli.getData("/xzw", new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    try {
                        register();
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, null);
    
            System.out.println(new String(data));
        }
    
        /**
         * 测试循环注册
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void testRegister() throws KeeperException, InterruptedException {
            register();
            Thread.sleep(Long.MAX_VALUE);
        }

# 九、完整代码 #

package com.xzw.zookeeper;
    
    import org.apache.zookeeper.*;
    import org.apache.zookeeper.data.Stat;
    import org.junit.Before;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.util.List;
    
    /**
     * @author: xzw
     * @create_date: 2020/9/9 9:59
     * @desc: Zookeeper客户端
     * @modifier:
     * @modified_date:
     * @desc:
     */
    public class ZkClient {
    
        //定义Zookeeper客户端实例
        private ZooKeeper zkCli;
        //定义连接串
        private static final String CONNECT_STRING = "master:2181,slave01:2181,slave02:2181";
        //定义session timeout
        private static final int SESSION_TIMEOUT = 2000;
    
        /**
         * 初始化Zookeeper，定义一个默认的回调函数
         * @throws IOException
         */
        @Before
        public void before() throws IOException {
            zkCli = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, e -> {
                System.out.println("默认回调函数");
            });
        }
    
        /**
         * 查看ZNode中包含的节点
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void ls() throws KeeperException, InterruptedException {
            //默认回调函数
    //        List<String> children = zkCli.getChildren("/", true);
            //自定义回调函数
            List<String> children = zkCli.getChildren("/", e -> {
                System.out.println("自定义回调函数");
            });
    
            System.out.println("=====================================");
            for (String child: children) {
                System.out.println(child);
            }
            System.out.println("=====================================");
    
            Thread.sleep(Long.MAX_VALUE);
        }
    
        /**
         * 创建节点
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void create() throws KeeperException, InterruptedException {
            String s = zkCli.create("/test", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.println(s);
    
            Thread.sleep(Long.MAX_VALUE);
        }
    
        /**
         * 获取节点信息
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void get() throws KeeperException, InterruptedException {
            byte[] data = zkCli.getData("/xzw", true, new Stat());
    
            String s = new String(data);
    
            System.out.println(s);
        }
    
        /**
         * 修改节点内容
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void set() throws KeeperException, InterruptedException {
            Stat stat = zkCli.setData("/xzw", "test".getBytes(), 0);
    
            System.out.println(stat.getDataLength());
        }
    
        /**
         * 查看结构体
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void stat() throws KeeperException, InterruptedException {
            Stat exists = zkCli.exists("/xzw", false);
            if (exists == null) {
                System.out.println("节点不存在");
            } else {
                System.out.println(exists.getDataLength());
            }
        }
    
        /**
         * 删除节点
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void delete() throws KeeperException, InterruptedException {
            Stat exists = zkCli.exists("/xzw", false);
            if (exists != null)
                zkCli.delete("/xzw", exists.getVersion());
        }
    
        /**
         * 循环注册节点：当节点发生变化时，可以不停的通知
         * @throws KeeperException
         * @throws InterruptedException
         */
        public void register() throws KeeperException, InterruptedException {
            byte[] data = zkCli.getData("/xzw", new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    try {
                        register();
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, null);
    
            System.out.println(new String(data));
        }
    
        /**
         * 测试循环注册
         * @throws KeeperException
         * @throws InterruptedException
         */
        @Test
        public void testRegister() throws KeeperException, InterruptedException {
            register();
            Thread.sleep(Long.MAX_VALUE);
        }
    }

至此，本文就结束了，你们在此过程中遇到了什么问题，欢迎留言，让我看看你们都遇到了哪些问题~

[Link 1]: https://blog.csdn.net/gdkyxy2013/category_10006538.html