ZooKeeper官网提供了Java和C的API。 本文使用Java API来实现ZooKeeper的基本操作。
前言
下图中的Replicated Database是包含完整数据树（entire data tree）的内存数据库。ZooKeeper的节点都是存放在内存中，所以读写速度很快。更新日志被记录到了磁盘中，以便用于恢复数据。在更新内存中节点数据之前，会先序列化到磁盘中，然后才会加载到内存中。
ZooKeeper中的每个节点存储的数据要被原子性的操作。也就是说读操作将获取与节点相关的所有数据，写操作也将替换掉节点的所有数据。另外，每一个节点都拥有自己的ACL(访问控制列表)，这个列表规定了用户的权限，即限定了特定用户对目标节点可以执行的操作。
目录
一、API引用
二、API简介
三、API详解
1. 连接zooKeeper服务器
2. 创建znode（zk节点）
3. 删除znode节点
4. 更新znode节点数据
5. 判断znode是否存在
6. 获取znode节点的子节点列表
7. 获取znode节点数据
四、代码示例
一、API引用
<!-- zookeeper -->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>${zookeeper.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
二、API简介
ZooKeeper API共包含五个包：
- org.apache.zookeeper
- org.apache.zookeeper.data
- org.apache.zookeeper.server
- org.apache.zookeeper.server.quorum
- org.apache.zookeeper.server.upgrade
其中org.apache.zookeeper，包含ZooKeeper类，他是我们编程时最常用的类文件。这个类是ZooKeeper客户端的主要类文件。如果要使用ZooKeeper服务，应用程序首先必须创建一个ZooKeeper实例，这时就需要使用此类。一旦客户端和ZooKeeper服务建立起了连接，ZooKeeper系统将会给次连接会话分配一个ID值，并且客户端将会周期性的向服务器端发送心跳来维持会话连接。只要连接有效，客户端就可以使用ZooKeeperAPI来做相应处理了。
三、API详解
1. 连接zooKeeper服务器
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
参数说明：
| connectString | zooKeeper server列表, 以逗号隔开. zooKeeper对象初始化后, 将从server列表中选择一个server, 并尝试与其建立连接. 如果连接建立失败, 则会从列表的剩余项中选择一个server, 并再次尝试建立连接。 |
| sessionTimeout | 指定连接的超时时间。 |
| watcher | 一个 watcher 实例。 |
注意，创建ZooKeeper对象时，只要对象完成初始化便立刻返回。建立连接是以异步的形式进行的，当连接成功建立后，会回调watcher的process方法。如果想要同步建立与server的连接，需要自己进一步封装。一般我们创建完Zookeeper之后，直接等着其连接好了之后，在watcher的process方法里面让主函数结束等待。
2. 创建znode（zk节点）
public String create(final String path, byte data[], List<ACL> acl, CreateMode createMode)
参数说明：
| path | znode的路径，必须是绝对路径。 |
| data | 与znode关联的数据。 |
| acl | 指定权限信息, 如果不想指定权限, 可以传入Ids.OPEN_ACL_UNSAFE。 |
| createMode | 指定znode类型. CreateMode是一个枚举类, 从中选择一个成员传入即可。 |
createMode的可选的成员类型：
| PERSISTENT | 创建后只要不删就永久存在. |
| EPHEMERAL | 会话结束年结点自动被删除，EPHEMERAL结点不允许有子节点。 |
| SEQUENTIAL | 节点名末尾会自动追加一个10位数的单调递增的序号，同一个节点的所有子节点序号是单调递增的。 |
| PERSISTENT_SEQUENTIAL | 结合PERSISTENT和SEQUENTIAL。 |
| EPHEMERAL_SEQUENTIAL | 结合EPHEMERAL和SEQUENTIAL。 |
znode中关联的数据不能超过1M。ZooKeeper的使命是分布式协作，而不是数据存储。
3. 删除znode节点
public void delete(final String path, int version)
参数说明：
| path | znode的路径，必须是绝对路径。 |
| version | 指定要删除的数据的版本, 如果version和真实的版本不同, 删除操作将失败。指定version为-1则忽略版本检查。 |
4. 更新znode节点数据
public Stat setData(final String path, byte data[], int version)
参数说明：
| path | znode的路径，必须是绝对路径。 |
| data | 要设置的到znode的数据，注意数据是直接清空原来的数据，然后写新的数据，且数据的大小不能超过1MB。 |
| version | 指定要更新的数据的版本, 如果version和真实的版本不同, 更新操作将失败。 指定version为-1则忽略版本检查。 |
5. 判断znode是否存在
public Stat exists(final String path, Watcher watcher)
参数说明：
| path | znode的路径，必须是绝对路径。 |
| watcher | 一个 watcher 实例。 |
如果该node存在，则返回该node的状态信息，否则返回null。
public Stat exists(String path, boolean watch)
其余和上面都一样，就是boolean如果为true，表示要用watcher，这时的watcher是初始化zookeeper时候指定的watcher。
6. 获取znode节点的子节点列表
public List<String> getChildren(final String path, Watcher watcher)
参数说明：
| path | znode的路径，必须是绝对路径。 |
| watcher | 一个 watcher 实例。 |
如果有child存在，就返回child，如果没有的话，就会返回null。
public List<String> getChildren(String path, boolean watch)
其余和上面都一样，就是boolean如果为true，表示要用watcher，这时的watcher是初始化zookeeper时候指定的watcher。
7. 获取znode节点数据
public byte[] getData(final String path, Watcher watcher, Stat stat)
参数说明：
| path | znode的路径，必须是绝对路径。 |
| watcher | 用于指定是否监听path node的删除事件，以及数据更新事件，注意，不监听path node的创建事件，因为如果path node不存在，该方法将抛出KeeperException.NoNodeException异常。 |
| stat | 是个传出参数，getData方法会将path node的状态信息设置到该参数中。是个传出参数，getData方法会将path node的状态信息设置到该参数中。 |
public byte[] getData(String path, boolean watch, Stat stat)
其余和上面都一样，就是boolean如果为true，表示要用watcher，这时的watcher是初始化zookeeper时候指定的watcher。
四、代码示例
public class ZooKeeperService {
/**
* Server列表, 以逗号分割, 例如: 192.168.3.4:2181,192.168.3.5:2181,192.168.3.6:2181
*/
private final String hosts;
/**
* 连接的超时时间, 毫秒
*/
private int timeoutMills;
private CountDownLatch connectedSemaphore = new CountDownLatch(1);
private ZooKeeper zk;
/**
* watcher实例
*/
private Watcher connWatcher = new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println("conn watched event >>> " + event.toString());
// 连接建立, 回调process接口时, 其event.getState()为KeeperState.SyncConnected
if (event.getState() == Event.KeeperState.SyncConnected) {
// 放开闸门, wait在connect方法上的线程将被唤醒
connectedSemaphore.countDown();
}
}
};
private Watcher existsWatcher = new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println("exists watched event >>> " + event.toString());
System.out.println("----------------------------------------------");
System.out.println("path : " + event.getPath() + ", type : " + event.getType() + ", state : " + event.getState());
try {
System.out.println("数据: " + watchData(event.getPath()));
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
System.out.println("----------------------------------------------");
}
};
public ZooKeeperService(String hosts, int timeoutMills) throws IOException, InterruptedException {
this.hosts = hosts;
this.timeoutMills = timeoutMills;
// 连接ZooKeeper服务器
connect();
}
/**
* 连接ZooKeeper Server
* <p>
* 创建ZooKeeper对象时, 只要对象完成初始化便立刻返回.
* 建立连接是以异步的形式进行的, 当连接成功建立后, 会回调Watcher的process方法.
* 如果想要同步建立与server的连接, 需要自己进一步封装.
*/
public void connect() throws IOException, InterruptedException {
zk = new ZooKeeper(hosts, timeoutMills, connWatcher);
// 等待连接完成
connectedSemaphore.await();
}
/**
* 关闭连接
*/
public void disconnect() throws InterruptedException {
if (zk != null) {
zk.close();
}
}
/**
* 创建节点
*
* @param path 节点路径
* @param data 初始数据
* @return 节点的路径
*/
public boolean createPath(String path, String data) throws KeeperException, InterruptedException {
return zk.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT) != null;
}
/**
* 读取指定节点数据
*
* @param path 节点路径
* @return 节点数据
*/
public String readData(String path) throws KeeperException, InterruptedException {
return new String(zk.getData(path, false, null));
}
/**
* 监听指定节点数据
*
* @param path 节点路径
* @return 节点数据
*/
public String watchData(String path) throws KeeperException, InterruptedException {
Stat stat = zk.exists(path, existsWatcher);
return new String(zk.getData(path, existsWatcher, stat));
}
/**
* 更新指定节点数据
*
* @param path 节点路径
* @param data 数据
* @return 是否更新成功
*/
public boolean writeData(String path, String data) throws KeeperException, InterruptedException {
return zk.setData(path, data.getBytes(), -1) != null;
}
/**
* 删除节点
*
* @param path 节点路径
*/
public void deletePath(String path) throws KeeperException, InterruptedException {
zk.delete(path, -1);
}
public static void main(String[] args) throws Exception {
String hosts = "192.168.3.4:2181,192.168.3.5:2181,192.168.3.6:2181";
String path = "/header";
int timeoutMills = 3000;
ZooKeeperService service = new ZooKeeperService(hosts, timeoutMills);
if (service.createPath(path, "I'm the initial data.")) {
System.out.println("数据: " + service.readData(path));
service.writeData(path, "I'm the updated data.");
System.out.println("数据: " + service.readData(path));
service.deletePath(path);
}
service.disconnect();
}
public static void mainWatch(String[] args) throws Exception {
String hosts = "192.168.3.4:2181,192.168.3.5:2181,192.168.3.6:2181";
String path = "/header";
int timeoutMills = 3000;
ZooKeeperService service = new ZooKeeperService(hosts, timeoutMills);
System.out.println(service.watchData(path));
System.in.read();
}
}