大数据生态之zookeeper（API）

Posted 2021-01-30

在使用maven项目编写zookeeper程序时导入相关的jar包：

        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.10</version>
        </dependency>

---

1.创建集群连接

  //集群节点节点，之间使用逗号隔开
    static String url="hadoop01:2181,hadoop02:2181,hadoop03:2181";
    //集群编程入口
    static ZooKeeper zk;
    //超时时间
    static long time=5000;
    //使用静态代码块，初始化ZooKeeper对象
    static {
        try {
            zk=new ZooKeeper(url,(int)time,null);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

2.创建节点

    public void createZnode(ZooKeeper zk){
        //必须使用绝对路径
        String path="/test/test04";
        //在创建创建节点时，必须指定节点内容
        byte data[]="I am a good boy!".getBytes();
        try {
            /**
             * zk.create的四个参数表示：
             * arg1:节点路径
             * arg2:节点存放的内容
             * arg3:权限设置：默认
             * arg4:节点类型（永久无顺序、永久有顺序、临时无顺序、临时有顺序）
             */
            String info = zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE
                    , CreateMode.EPHEMERAL);
            System.out.println(info==null?"fail":"success");
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

3.查看节点信息

    public void getInfo(ZooKeeper zk,String path){
        try {
            /**

             * zk.getData(path, null, null);
             *  参数1:节点的绝对路径  string
             *  参数2:节点的监听
             *  参数3：节点的状态对象
             */
            byte[] data = zk.getData(path, null, null);
            System.out.println("节点："+path+"的信息是："+(new String(data)));
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

4.修改节点数据

 public void updateData(ZooKeeper zk,String path){
        try {
            /**
             * zk.setData(path, data, version);
             * path：修改内容的节点
             * data：更新的数据
             * version：修改的版本：如果不知道，写-1，表示最新版本 
             */
            zk.setData(path,"haha".getBytes(),-1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

5.删除节点

 public void deleteZnode(ZooKeeper zk,String path){
        try {
            /**
             * zk.delete(path, version);
             * path：路径
             * version：删除的版本，如果不知道，写-1，表示删除最新版本  int
             */
            zk.delete(path,-1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

6.获取znode节点的权限信息

public void getACL(ZooKeeper zk,String path){
        try {
            /**
             * zk.getACL(path,stat)
             * path:路径
             * stat:状态信息
             */
            List<ACL> acls = zk.getACL(path, null);
            for(ACL acl:acls){
                System.out.println(acl.getPerms());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

7.获取子节点

public void getChildren(ZooKeeper zk, String path) {
        try {
            //判断节点是否存在
            if (zk.exists(path, null) == null) {
                List<String> childrens = zk.getChildren(path, null);
                for(String children:childrens){
                    //注意这里面收到的都是当前节点下的子节点的相对路径
                    System.out.println(children);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

8.znode的测试监听

public class testZKWatcher {
    static String url="hadoop01:2181,hadoop02:2181,hadoop03:2181";
    static ZooKeeper zk;
    static long time=5000;
    static {
        try {
            zk=new ZooKeeper(url,(int) time,null);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    //设置监听
    public static void liisten(String path,ZooKeeper zk){
        try {
            zk.exists(path, new Watcher() {
                @Override
                //WatchedEvent event监听事件对象
                public void process(WatchedEvent event) {
                    //节点触发的类型（新增、修改、创建...）
                    Event.EventType type = event.getType();
                    //获取触发的节点的路径
                    event.getPath();
                    System.out.println("触发事件的路径"+path+",触发的事件的类型"+type);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        String path="/test/test01";
        liisten(path,zk);
        try {
            //触发监听
            zk.create("jj","zzy".getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

| 这里需要注意的是，监听事件是一次性的，如果需要一直监听，那么需要设置循环监听。