Zookeeper教程
1、Zookeeper CLI
ZooKeeper命令行界面CLI用于与ZooKeeper集合进行交互以进行开发。它有助于调试和解决不同的选项。
要执行ZooKeeper CLI操作,首先打开ZooKeeper服务器bin/zkServer.sh start
,然后打开ZooKeeper客户端
bin/zkCli.sh
。
一旦客户端启动,你可以进行以下操作。
1.1 创建Znodes
用给定的路径创建一个znode,flag参数指定创建的znode是临时的,持久的还是顺序的。
-s
代表顺序节点, -e
代表临时节点,注意其中 -s
和 -e
可以同时使用的,并且临时节点不能再创建子节
点。
默认情况下,所有znode都是持久的。当会话过期或客户端断开连接时,临时节点-e
将被自动删除。
顺序节点保证znode路径将是唯一的。ZooKeeper集合将向znode路径填充10位序列号。例如,znode路径
/myapp
将转换为/myapp0000000001
,下一个序列号将为/myapp0000000002
。如果没有指定flag,则znode被
认为是持久的。
# create 命令用于创建节点并赋值
# -s 代表顺序节点
# -e 代表临时节点
# acl: 访问权限相关,默认是 world,相当于全世界都能访问
create [-s] [-e] path data [acl]
create /test1 test1-content
要创建顺序节点,请添加-s参数,如下所示。
create -s /test2 test2-content
会产生两个顺序的文件,/test20000000020
和/test20000000021
。
要创建临时节点,请添加-e参数,如下所示。
create -e /test3 test3-content
记住当客户端断开连接时,临时节点将被删除。你可以通过退出ZooKeeper CLI,然后重新打开CLI来尝试。
-s和-e同时使用添加临时顺序节点,如下所示。
create -s -e /test4 test4-content
创建的节点既是有序,又是临时节点。
1.2 获取数据
它返回znode的关联数据和指定znode的元数据。你将获得信息,例如上次修改数据的时间,修改的位置以及数据
的相关信息。此CLI还用于分配监视器以显示数据相关的通知。
# get 命令用于获取节点数据和状态信息
# watch: 对节点进行事件监听
# 当指定的znode或znode的子数据更改时,监视器会显示通知,你只能在 get 命令中设置watch
get path [watch]
get /test1
要访问顺序节点,必须输入znode的完整路径。
get /test20000000020
watch参数的使用
以下实例查看同时开启两个终端。
终端一:
get /test3 watch
在终端二对此节点进行修改:
set /test3 test3-content-update1
终端一自动显示 NodeDataChanged 事件:
1.3 设置数据
设置指定znode的数据。完成此设置操作后,你可以使用 get 命令检查数据。
# set 命令用于修改节点存储的数据
# version: 可选项,版本号(可用作乐观锁)
set path data [version]
get /test1
set /test1 test1-content-update1 0
dataVersion
会自增。
get /test1
如果指定的版本号错误:
set /test1 test1-content-update2 10
只有正确的版本号才能设置成功。
1.4 创建子项/子节点
创建子节点类似于创建新的znode。唯一的区别是,子znode的路径也将具有父路径。
# parent/path/subnode/path为多个路径
create parent/path/subnode/path data
create /test1/child1 firstchildren
create /test1/child2 secondchildren
1.5 列出子项
此命令用于列出和显示znode的子项。
# ls 命令用于查看某个路径下目录列表
ls path
示例
ls /test1
ls2
命令用于查看某个路径下目录列表,它比 ls 命令列出更多的详细信息。
ls2 /test1
1.6 检查状态
状态描述指定的znode的元数据。它包含时间戳,版本号,ACL,数据长度和子znode等细项。
# stat 命令用于查看节点状态信息
# watch: 对节点进行事件监听
stat path [watch]
示例
stat /test1
1.7 移除Znode
移除指定的znode并递归其所有子节点。只有在这样的znode可用的情况下才会发生。
rmr path
rmr /test1
删除delete /path
命令类似于 remove
命令,但是delete
只适用于没有子节点的znode。
# delete 命令用于删除某节点
# version: 可选项,版本号(同 set 命令)
delete /path [version]
delete /test3
2、Zookeeper API
ZooKeeper有一个绑定Java和C的官方API。Zookeeper社区为大多数语言(.NET,python等)提供非官方API。
使用ZooKeeper API,应用程序可以连接,交互,操作数据,协调,最后断开与ZooKeeper集合的连接。
ZooKeeper API具有丰富的功能,以简单和安全的方式获得ZooKeeper集合的所有功能。ZooKeeper API提供同步
和异步方法。
ZooKeeper集合和ZooKeeper API在各个方面都完全相辅相成,对开发人员有很大的帮助。让我们在本章讨论Java
绑定。
2.1 ZooKeeper API的基础知识
与ZooKeeper集合进行交互的应用程序称为ZooKeeper客户端或简称客户端。
Znode是ZooKeeper集合的核心组件,ZooKeeper API提供了一小组方法使用ZooKeeper集合来操纵znode的所有
细节。
客户端应该遵循以步骤,与ZooKeeper集合进行清晰和干净的交互。
-
连接到ZooKeeper集合。ZooKeeper集合为客户端分配会话ID。
-
定期向服务器发送心跳。否则,ZooKeeper集合将过期会话ID,客户端需要重新连接。
-
只要会话ID处于活动状态,就可以获取/设置znode。
-
所有任务完成后,断开与ZooKeeper集合的连接。如果客户端长时间不活动,则ZooKeeper集合将自动断开
客户端。
2.2 Java绑定
让我们来了解本章中最重要的一组ZooKeeper API。ZooKeeper API的核心部分是ZooKeeper类。它提供了在其构
造函数中连接ZooKeeper集合的选项,并具有以下方法:
-
connect : 连接到ZooKeeper集合
-
create : 创建znode
-
exists : 检查znode是否存在及其信息
-
getData : 从特定的znode获取数据
-
setData : 在特定的znode中设置数据
-
getChildren : 获取特定znode中的所有子节点
-
delete : 删除特定的znode及其所有子项
-
close : 关闭连接
使用到的依赖:
<?xml version="1.0" encoding="UTF-8"?>
<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>com.example</groupId>
<artifactId>spring-boot-zookeeper-rumen</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.8</version>
</dependency>
</dependencies>
</project>
2.3 连接到ZooKeeper集合
ZooKeeper类通过其构造函数提供connect功能。构造函数的签名如下 :
ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)
-
connectionString - ZooKeeper集合主机。
-
sessionTimeout - 会话超时(以毫秒为单位)。
-
watcher - 实现监视器界面的对象,ZooKeeper集合通过监视器对象返回连接状态。
让我们创建一个新的帮助类 ZooKeeperConnection ,并添加一个方法 connect 。 connect 方法创建一个
ZooKeeper对象,连接到ZooKeeper集合,然后返回对象。
这里 CountDownLatch 用于停止(等待)主进程,直到客户端与ZooKeeper集合连接。
ZooKeeper集合通过监视器回调来回复连接状态。一旦客户端与ZooKeeper集合连接,监视器回调就会被调用,
并且监视器回调函数调用CountDownLatch的countDown方法来释放锁,在主进程中await。
package com.example;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
/**
* Zookeeper连接工具类
*/
public class ZooKeeperConnection {
private ZooKeeper zoo;
final CountDownLatch connectedSignal = new CountDownLatch(1);
public ZooKeeper connect(String host) throws IOException, InterruptedException {
zoo = new ZooKeeper(host, 5000, new Watcher() {
public void process(WatchedEvent we) {
if (we.getState() == KeeperState.SyncConnected) {
connectedSignal.countDown();
}
}
});
connectedSignal.await();
return zoo;
}
public void close() throws InterruptedException {
zoo.close();
}
public static void main(String[] args) throws IOException, InterruptedException {
ZooKeeperConnection zooKeeperConnection = new ZooKeeperConnection();
ZooKeeper zooKeeper = zooKeeperConnection.connect("192.168.99.199:2181");
// CONNECTED
System.out.println(zooKeeper.getState());
}
}
2.4 创建Znode
ZooKeeper类提供了在ZooKeeper集合中创建一个新的znode的create方法。 create 方法的签名如下:
create(String path, byte[] data, List<ACL> acl, CreateMode createMode)
-
path - Znode路径。例如,/myapp1,/myapp2,/myapp1/mydata1,
myapp2/mydata1/myanothersubdata
-
data - 要存储在指定znode路径中的数据
-
acl - 要创建的节点的访问控制列表。ZooKeeper API提供了一个静态接口 ZooDefs.Ids 来获取一些基本的acl
列表。例如,ZooDefs.Ids.OPEN_ACL_UNSAFE返回打开znode的acl列表。
-
createMode - 节点的类型,即临时,顺序或两者。这是一个枚举。
让我们创建一个新的Java应用程序来检查ZooKeeper API的 create 功能。创建文件 ZKCreate.java 。在main方
法中,创建一个类型为 ZooKeeperConnection 的对象,并调用 connect 方法连接到ZooKeeper集合。
connect方法将返回ZooKeeper对象 zk 。现在,请使用自定义path和data调用 zk 对象的 create 方法。
package com.example;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
/**
* 创建Znode
*/
public class ZKCreate {
private static ZooKeeper zk;
private static ZooKeeperConnection conn;
public static void create(String path, byte[] data) throws
KeeperException, InterruptedException {
zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
public static void main(String[] args) {
String path = "/MyFirstZnode";
byte[] data = "My first zookeeper app".getBytes();
try {
conn = new ZooKeeperConnection();
zk = conn.connect("192.168.99.199:2181");
create(path, data);
conn.close();
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
}
一旦编译和执行应用程序,将在ZooKeeper集合中创建具有指定数据的znode。
你可以使用ZooKeeper CLI zkCli.sh
进行检查。
$ cd /home/zhangshixing/kafka/zookeeper-3.4.9/bin
$ ./zkCli.sh
>>> get /MyFirstZnode
2.5 Exists - 检查Znode的存在
ZooKeeper类提供了 exists 方法来检查znode的存在。如果指定的znode存在,则返回一个znode的元数据。
exists方法的签名如下:
exists(String path, boolean watcher)
-
path- Znode路径
-
watcher - 布尔值,用于指定是否监视指定的znode
让我们创建一个新的Java应用程序来检查ZooKeeper API的exists功能。
package com.example;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
public class ZKExists {
private static ZooKeeper zk;
private static ZooKeeperConnection conn;
public static Stat zNodeExists(String path) throws
KeeperException, InterruptedException {
return zk.exists(path, true);
}
public static void main(String[] args) throws InterruptedException, KeeperException {
String path = "/MyFirstZnode";
try {
conn = new ZooKeeperConnection();
zk = conn.connect("192.168.99.199:2181");
Stat stat = zNodeExists(path);
if (stat != null) {
System.out.println("Node exists and the node version is " +
stat.getVersion());
} else {
System.out.println("Node does not exists");
}
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
}
一旦编译和执行应用程序,你将获得以下输出。
Node exists and the node version is 0
2.6 getData方法
ZooKeeper类提供 getData 方法来获取附加在指定znode中的数据及其状态。 getData 方法的签名如下:
getData(String path, Watcher watcher, Stat stat)
-
path - Znode路径。
-
watcher - 监视器类型的回调函数。当指定的znode的数据改变时,ZooKeeper集合将通过监视器回调进行通
知。这是一次性通知。
-
stat - 返回znode的元数据。
让我们创建一个新的Java应用程序来了解ZooKeeper API的 getData 功能。
package com.example;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import java.util.concurrent.CountDownLatch;
public class ZKGetData {
private static ZooKeeper zk;
private static ZooKeeperConnection conn;
public static Stat zNodeExists(String path) throws
KeeperException, InterruptedException {
return zk.exists(path, true);
}
public static void main(String[] args) throws InterruptedException, KeeperException {
String path = "/MyFirstZnode";
final CountDownLatch connectedSignal = new CountDownLatch(1);
try {
conn = new ZooKeeperConnection();
zk = conn.connect("192.168.99.199:2181");
Stat stat = zNodeExists(path);
if (stat != null) {
byte[] b = zk.getData(path, new Watcher() {
public void process(WatchedEvent we) {
if (we.getType() == Event.EventType.None) {
switch (we.getState()) {
case Expired:
connectedSignal.countDown();
break;
}
} else {
String path = "/MyFirstZnode";
try {
byte[] bn = zk.getData(path,
false, null);
String data = new String(bn,
"UTF-8");
System.out.println(data);
connectedSignal.countDown();
} catch (Exception ex) {
System.out.println(ex.getMessage());
}
}
}
}, null);
String data = new String(b, "UTF-8");
System.out.println(data);
connectedSignal.await();
} else {
System.out.println("Node does not exists");
}
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
}
一旦编译和执行应用程序,你将获得以下输出:
My first zookeeper app
2.7 setData方法
ZooKeeper类提供 setData 方法来修改指定znode中附加的数据。 setData 方法的签名如下:
setData(String path, byte[] data, int version)
-
path- Znode路径
-
data - 要存储在指定znode路径中的数据。
-
version- znode的当前版本。每当数据更改时,ZooKeeper会更新znode的版本号。
现在让我们创建一个新的Java应用程序来了解ZooKeeper API的 setData 功能。
package com.example;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
public class ZKSetData {
private static ZooKeeper zk;
private static ZooKeeperConnection conn;
public static void update(String path, byte[] data) throws
KeeperException, InterruptedException {
zk.setData(path, data, zk.exists(path, true).getVersion());
}
public static void main(String[] args) throws InterruptedException, KeeperException {
String path = "/MyFirstZnode";
byte[] data = "Success".getBytes();
try {
conn = new ZooKeeperConnection();
zk = conn.connect("192.168.99.199:2181");
update(path, data);
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
}
2.8 getChildren方法
ZooKeeper类提供 getChildren 方法来获取特定znode的所有子节点。 getChildren 方法的签名如下:
getChildren(String path, Watcher watcher)
-
path - Znode路径。
-
watcher - 监视器类型的回调函数。当指定的znode被删除或znode下的子节点被创建/删除时,ZooKeeper
集合将进行通知。这是一次性通知。
package com.example;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import java.util.List;
public class ZKGetChildren {
private static ZooKeeper zk;
private static ZooKeeperConnection conn;
public static Stat znode_exists(String path) throws
KeeperException, InterruptedException {
return zk.exists(path, true);
}
public static void main(String[] args) throws InterruptedException, KeeperException {
String path = "/MyFirstZnode";
try {
conn = new ZooKeeperConnection();
zk = conn.connect("192.168.99.199:2181");
Stat stat = znode_exists(path);
if (stat != null) {
List<String> children = zk.getChildren(path, false);
for (int i = 0; i < children.size(); i++)
System.out.println(children.get(i));
} else {
System.out.println("Node does not exists");
}
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
}
child1
child2
2.9 删除Znode
ZooKeeper类提供了 delete 方法来删除指定的znode。 delete 方法的签名如下:
delete(String path, int version)
-
path - Znode路径。
-
version - znode的当前版本。
让我们创建一个新的Java应用程序来了解ZooKeeper API的 delete 功能。
package com.example;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
public class ZKDelete {
private static ZooKeeper zk;
private static ZooKeeperConnection conn;
public static void delete(String path) throws KeeperException, InterruptedException {
zk.delete(path, zk.exists(path, true).getVersion());
}
public static void main(String[] args) throws InterruptedException, KeeperException {
String path = "/MyFirstZnode";
try {
conn = new ZooKeeperConnection();
zk = conn.connect("192.168.99.199:2181");
delete(path);
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
}