之前体验了命令行客户端,这次看一下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客户端修改数据:
可以看到确实可以持续监听数据节点。