zookeeper java api介绍

《zookeeper介绍及环境搭建》《zookeeper客户端的使用》两篇文章中,我介绍了zookeeper实验环境的搭建、zookeeper的数据结构和zookeeper的一些操作命令。本篇文章我将对zookeeper java api进行详细的介绍。相关代码

开发环境搭建

zookeeper.jar中包含了zookeeper提供的java api,为了在项目中引入zookeeper.jar,在工程的pom.xml文件中加入下面的依赖:

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

连接zk服务器

构造ZooKeeper类对象的过程就是与ZK服务器建立连接的过程。

ZooKeeper zooKeeper = new ZooKeeper("192.168.1.108:2181", 5000, watcher);

ZooKeeper类的构造函数一共有三个参数:第一个参数是服务器的地址,第二个参数是session超时时间,第三个参数是org.apache.zookeeper.Watcher类型的对象。zookeeper api与服务器建立连接是异步的,上面的调用会马上从ZooKeeper构造函数返回,当与服务器建立好连接之后会调用Watcher中的process方法进行处理。process方法会接受一个WatchedEvent类型的参数,用于表明发生了什么事件。

public void process(WatchedEvent watchedEvent) {
    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) { //判断是否已连接
        if(watchedEvent.getType() == Event.EventType.None && null == watchedEvent.getPath()) {
            // 最初与zk服务器建立好连接
        } else if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
            // 子节点变化事件
        }
        // ...还可以继续监听其它事件类型
    }
    System.out.println(watchedEvent.getState());
}

WatchedEvent包含两方面重要信息:

  1. 与zk服务器连接的状态信息
    可以调用watchedEvent.getState()方法获取与zk服务器连接的状态信息,状态信息取值主要包括SyncConnectedDisconnectedConnectedReadOnlyAuthFailed等等。
  2. 发生的具体事件类型信息
    watchedEvent.getState()方法只是获取与zk服务器连接的状态信息,但在同一个连接状态下,还会发生很多事件的类型。例如在zk中,我们可以watch一个节点的数据内容,当这个节点的数据被改变时,我们可以获取到这个事件。类似的还有子节点列表变化事件等等。
    这就需要我们在SyncConnected同一种连接状态下区分多个事件类型。可以通过watchedEvent.getType()方法获取具体的事件类型。事件类型的取值包括NoneNodeCreatedNodeDeletedNodeDataChangedNodeChildrenChanged

创建节点

下面要介绍的每种api操作都可以分为两种类型——同步和异步。同步操作一般会有返回值,并且会抛出相应的异常。异步操作没有返回值,也不会抛出异常。此外异步方法参数在同步方法参数的基础上,会增加Callback和context两个参数。
如用同步方式创建一个节点的的代码如下:

private void createNodeSync() throws KeeperException, InterruptedException {
    String path = "/poype_node";
    String nodePath = zooKeeper.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println(nodePath);
}

这里的zooKeeper就是通过ZooKeeper构造函数构造的对象,可以调用它的create()方法创建一个节点。同步版的create()方法一共有4个参数,第一个参数是要创建的节点路径。第二个参数是创建节点的数据值,参数类型是字节数组。第三个参数是这个节点的访问权限,我们这里指定该节点可以被任何人访问(关于节点的访问权限,我将在下一篇文章进行详细介绍)。第四个参数是创建节点的类型,在上一篇文章中,我提到过create命令可以有-s和-e两个参数,其中-s是顺序节点,-e是临时节点。这里的CreateMode就是这两个参数的组合,它有下面四种取值:PERSISTENTPERSISTENT_SEQUENTIALEPHEMERALEPHEMERAL_SEQUENTIAL
异步模式创建一个节点的代码如下,注意异步模式方法没有返回值,并且不会抛出任何异常:

private void createNodeAsync() {
    String path = "/poype_node2";
    zooKeeper.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new AsyncCallback.StringCallback() {
        public void processResult(int resultCode, String path, Object ctx, String name) {
            System.out.println(resultCode);
            System.out.println(path);
            System.out.println(ctx);
            System.out.println(name);
        }
    }, "创建");
}

除了同步create方法中的四个参数以外,异步模式的create方法还增加了callback和context两个参数。StringCallback接口中的processResult方法会在节点创建好之后被调用,它有四个参数。第一个是int类型的resultCode,作为创建节点的结果码,当成功创建节点时,resultCode的值为0。第二个参数是创建节点的路径。第三个参数是context,当一个StringCallback类型对象作为多个create方法的参数时,这个参数就很有用了。第四个参数是创建节点的名字,其实与path参数相同。

获取节点的数据值

同步获取一个节点数据值的代码如下:

private void getDataSync() throws KeeperException, InterruptedException {
    Stat stat = new Stat();
    // getData的返回值是该节点的数据值,节点的状态信息会赋值给stat对象
    byte[] data = zooKeeper.getData("/node_1",true, stat);
    System.out.println(new String(data));
    System.out.println(stat);
}

zooKeeper.getData方法的返回值就是节点中存储的数据值,它有三个参数,第一个参数是节点的路径,用于表示要获取哪个节点中的数据。第三个参数stat用于存储节点的状态信息,在调用getData方法前,会先构造一个空的Stat类型对象作为参数传给getData方法,当getData方法调用返回后,节点的状态信息会被填充到stat对象中。
第二个参数是一个bool类型的watch,这个参数比较重要。当watch为true时,表示我们想要监控这个节点的数据变化。当节点的数据发生变化时,我们就可以拿到zk服务器推送给我们的通知。在process方法中会有类似下面的代码:

public void process(WatchedEvent watchedEvent) {
    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) { //与zk服务器处于连接状态
        if(watchedEvent.getType() == Event.EventType.None && null == watchedEvent.getPath()) {
            createNodeAsync();
        } else if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
            // 节点的子节点列表发生变化
        } else if(watchedEvent.getType() == Event.EventType.NodeDataChanged) {
            // 节点的数据内容发生变化
        }
    }
}

当节点的数据内容发生变化时,我们就会接收到NodeDataChanged这个事件。值得注意的是,zooKeeper设置的监听只生效一次,如果在接收到NodeDataChanged事件后还想继续对该节点的数据内容改变进行监听,需要在事件处理逻辑中重新调用getData方法并将watch参数设置为true。
异步获取一个节点数据值的代码如下:

private void getDataAsync() {
        zooKeeper.getData("/node_1", true, new AsyncCallback.DataCallback() {
            public void processResult(int resultCode, String path, Object ctx, byte[] data, Stat stat) {
                System.out.println(resultCode);
                System.out.println(path);
                System.out.println(ctx);
                System.out.println(new String(data));
                System.out.println(stat);
            }
        }, "异步获取节点的数据值");
    }

异步getData方法中的最后一个参数是context,我这里将其设置为"异步获取节点的数据值"。

获取节点的子节点列表

同步获取一个节点的子节点列表:

private void getChildrenSync() throws KeeperException, InterruptedException {
    List<String> childrenNode = zooKeeper.getChildren("/",true);
    for(String child : childrenNode) {
        System.out.println(child);
    }
}

方法getChildren的第二个参数同样为watch,当节点的子节点列表发生变化时,zk服务器会向我们推送类型为NodeChildrenChanged的事件。
异步获取一个节点的子节点列表:

private void getChildrenAsync() {
    zooKeeper.getChildren("/", true, new AsyncCallback.Children2Callback() {
        public void processResult(int resultCode, String path, Object ctx, List<String> children, Stat stat) {
            System.out.println(resultCode);  
            System.out.println(path);        
            System.out.println(ctx);
            for(String child : children) {
                System.out.println(child);
            }
            System.out.println(stat);
        }
    }, "获取/下面的子节点");
}

查看一个节点是否存在

同步查看一个节点是否存在:

private void existSync() throws KeeperException, InterruptedException {
    Stat stat = zooKeeper.exists("/poype_node2", true);
    System.out.println(stat);
}

exists方法的watch参数比较特别,如果将其指定为true,那么代表你对该节点的创建事件、节点删除事件和该节点的数据内容改变事件都感兴趣,所以会同时响应三种事件类型。请看process方法中对事件的处理:

public void process(WatchedEvent watchedEvent) {
    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
        if(watchedEvent.getType() == Event.EventType.None && null == watchedEvent.getPath()) {
            existAsync();
        } else if(watchedEvent.getType() == Event.EventType.NodeCreated) { 
            System.out.println("监控到了该节点被创建");
            existAsync();
        } else if(watchedEvent.getType() == Event.EventType.NodeDataChanged) {
            System.out.println("监控到了该节点的数据内容发生变化");
            existAsync();
        } else if(watchedEvent.getType() == Event.EventType.NodeDeleted) {
            System.out.println("监控到了该节点被删除");
            existAsync();
        }
    }
}

异步查看一个节点是否存在:

private void existAsync() {
    zooKeeper.exists("/poype_node2", true, new AsyncCallback.StatCallback() {
        public void processResult(int resultCode, String path, Object ctx, Stat stat) {
            System.out.println(resultCode);
            System.out.println(path);
            System.out.println(ctx);
            System.out.println(stat);
        }
    }, "异步查看一个节点是否存在");
}

修改节点的数据内容

同步修改节点的数据内容:

private void setDataSync() throws KeeperException, InterruptedException {
    Stat stat = zooKeeper.setData("/poype_node2", "poype5211314".getBytes(), 1);
    System.out.println(stat);
}

setData方法有三个参数,前两个参数分别是节点的路径和要修改的数据值,最后一个参数是version字段。在上一篇文章我提到过,在节点的状态信息中包含dataVersion字段,是该节点的数据内容版本号。在调用setData方法修改节点数据内容时,只有当version参数的值与节点状态信息中的dataVersion值相等时,数据修改才能成功,否则会抛出BadVersion异常。这是为了防止丢失数据的更新,在ZooKeeper提供的API中,所有的写操作(例如后面要提到的delete)都有version参数。
异步修改节点的数据内容:

private void setDataAsync() {
    zooKeeper.setData("/poype_node2", "poype5211314".getBytes(), 3, new AsyncCallback.StatCallback() {
        public void processResult(int resultCode, String path, Object ctx, Stat stat) {
            System.out.println(resultCode);
            System.out.println(path);
            System.out.println(ctx);
            System.out.println(stat.getVersion()); // 获取数据节点版本号
        }
    }, "异步设置一个节点的数据");
}

删除一个节点

同步方式删除一个节点:

private void deleteSync() throws KeeperException, InterruptedException {
    zooKeeper.delete("/node_1", 12);
}

delete方法的第二个参数也是version,含义与setData方法中的version参数类似。
异步方式删除一个节点:

private void deleteAsync() {
    zooKeeper.delete("/poype_node", 3, new AsyncCallback.VoidCallback() {
        public void processResult(int resultCode, String path, Object ctx) {
            System.out.println(resultCode);
            System.out.println(path);
            System.out.println(ctx);
        }
    }, "异步删除一个节点");
}

小结

本文介绍了ZooKeeper Java API的基本使用方式,在下一篇文章中,我将详细介绍ZooKeeper中节点权限的概念和使用方法。

相关文章

相关标签/搜索