【zookeeper】java API 例子

之前体验了命令行客户端，这次看一下java API操作zk。

server还是按照之前的配置，一个server1，server2和server3的伪集群。

maven：

这里使用maven管理zk的jar包，大致需要zk的jar和日志的jar，pom.xml：

<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">
  <modelVersion>4.0.0</modelVersion>

  <groupId>yao</groupId>
  <artifactId>yao</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>yao</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>

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

    <dependency>
      <groupId>log4j</groupId>
      <artifactId>log4j</artifactId>
      <version>1.2.16</version>
    </dependency>

    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-api</artifactId>
      <version>1.6.1</version>
    </dependency>
  </dependencies>
</project>

1.连接zk：

package main;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;

public class ZooKeeperClient implements Watcher{

  private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
  public static void main(String[] args) {
    try {
      ZooKeeper zk = new ZooKeeper("127.0.0.1:2183", 5000, new ZooKeeperClient());
      System.out.println("Created zk");
      connectedSemaphore.await();
    } catch (IOException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public void process(WatchedEvent event) {
    System.out.println("Event recieved: " + event);
    if(event.getState() == KeeperState.SyncConnected){
      System.out.println("Connected");
      connectedSemaphore.countDown();
    }
  }

}

client类包含main方法和一个回调函数。main方法创建了一个zk的实例，构造方法的最后一个参数是回调接口。这个构造函数是一个异步函数，调用的时候直接返回，此时一般是没有创建好zk实例的，也就是没有连接成功的，所以main方法里new完了不是直接返回，使用一个latch来阻塞一下，latch的意思是只要构造函数的里数字不为0，那么await的地方就阻塞。那么解除阻塞的地方自然是在回调函数里了。一旦建立了连接，那么zk就会调用process回调函数。运行结果：

2.创建节点：

package main;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class ZooKeeperClient2 implements Watcher{

  private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
  public static void main(String[] args) {
    try {
      ZooKeeper zk = new ZooKeeper("127.0.0.1:2183", 5000, new ZooKeeperClient2());
      System.out.println("Created zk");
      connectedSemaphore.await();
      
      String path = zk.create("/zk-node2", "java-cli".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      System.out.println("Created [" + path + "] successfully");
    } catch (IOException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
      e.printStackTrace();
    } catch (KeeperException e) {
      e.printStackTrace();
    }
  }

  public void process(WatchedEvent event) {
    System.out.println("Event recieved: " + event);
    if(event.getState() == KeeperState.SyncConnected){
      System.out.println("Connected");
      connectedSemaphore.countDown();
    }
  }
}

主要使用了create函数，第二个参数是节点的数据，zk是不管序列化的，所以这里是byte数组参数，string的话java里面直接getBytes即可，其他的数据需要使用序列化工具。第三个参数是acl，与安全有关，第四个是节点类型，持久还是瞬时。

运行以后，可以使用cli客户端查看节点是都创建。

3.更新，并且监听。

package main;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class ZooKeeperClient3 implements Watcher {

  private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
  private static ZooKeeper zk = null;
  private static Stat stat = new Stat();

  public static void main(String[] args) throws Exception {
    String path = "/zk-book";
    zk = new ZooKeeper("127.0.0.1:2181", 5000, new ZooKeeperClient3());
    connectedSemaphore.await();
    zk.getData(path, true, stat);
    Thread.sleep(Integer.MAX_VALUE);
  }

  public void process(WatchedEvent event) {
    if (KeeperState.SyncConnected == event.getState()) {
      if (EventType.None == event.getType() && null == event.getPath()) {
        connectedSemaphore.countDown();
        System.out.println("connected");
      } else if (event.getType() == EventType.NodeDataChanged) {
        System.out.println("enter change callback func: ");
        try {
          System.out.println("str: " + new String(zk.getData("/zk-book", true, stat)));
        } catch (KeeperException e) {
          e.printStackTrace();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
}

这里多了一个stat对象实例，这个是zk的jar自带类型，用于存放zk的返回结果。

zk的watch机制需要注意的是watch只能watch一次，比方说这次加了一个监听器，然后事件发生，监听器发一个通知，再发生一个事件，这个监听器就不管用了，所以需要在事件发生时的回调函数里新建一个监听器，否则就只能监听一次。

这就是process函数里面必须调用一次zk的getData函数的原因，可以通过getData函数在、来添加一次监听。其中的第二个参数就是监听器，true表示用默认的监听器，也就是当前类。第三个参数是存放数据的stat。同样的在main函数里调用一次getData来添加第一次的监听器。这是持续监听的关键。

最后是关于回调函数里的if，有两个层次，第一个是state，每一个state下面是会发生不同的event，所以需要两级的if语句。具体每一state发生什么event，见：

3.1状态：KeeperState.Disconnected(0)

此时客户端处于断开连接状态，和ZK集群都没有建立连接。

3.1.1事件：EventType.None(-1)

触发条件：一般是在与服务器断开连接的时候，客户端会收到这个事件。

3.2状态：KeeperState. SyncConnected(3)

3.2.1事件：EventType.None(-1)

触发条件：客户端与服务器成功建立会话之后，会收到这个通知。

3.2.2事件：EventType. NodeCreated (1)

触发条件：所关注的节点被创建。

3.2.3事件：EventType. NodeDeleted (2)

触发条件：所关注的节点被删除。

3.2.4事件：EventType. NodeDataChanged (3)

触发条件：所关注的节点的内容有更新。注意，这个地方说的内容是指数据的版本号dataVersion。因此，即使使用相同的数据内容来更新，还是会收到这个事件通知的。无论如何，调用了更新接口，就一定会更新dataVersion的。

3.2.5事件：EventType. NodeChildrenChanged (4)

触发条件：所关注的节点的子节点有变化。这里说的变化是指子节点的个数和组成，具体到子节点内容的变化是不会通知的。

3.3状态 KeeperState. AuthFailed(4)

3.3.1事件：EventType.None(-1)

3.4状态 KeeperState. Expired(-112)

3.4.1事件：EventType.None(-1)

以上内容引自：​​http://nileader.blog.51cto.com/1381108/954670​​

运行以后：

使用cli客户端修改数据：

可以看到确实可以持续监听数据节点。