春节假期过去了,收收心我们来开始继续学习Zookeeper的其他相关知识.今天要说的是 Zookeeper Java客户端API的使用
在切入正题之前,延着上节内容我们来说几个客户端的脚本(客户端连接,创建,读取,更新,删除)
- 客户端连接
$ sh zkCli.sh
image.png
看到末尾是
WatchedEvent state:SyncConnected type:None path:null
[zk: localhost:2181(CONNECTED) 0]
就表示连接成功了
上面这种方式就是直接连,连接本地的Zookeeper.
那么要连接指定地址
$ sh zkCli.sh -server ip:port
- 创建
creat
创建节点,其中 -s 或 -e 分别指定特性: 顺序或者临时节点,默认不添加-s , -e. 创建的是持久节点
create [-s] [-e] path data acl
[zk: 192.168.131.134:2181(CONNECTED) 0] create /zk-test 123
Created /zk-test
其中参数的含义,
/zk-test 为节点的path, 123 为数据data, 最后一个参数是acl策略,缺省情况下,不做任何权限的控制,详细的说明后面继续
- 读取
ls
使用ls命令, 可以查看指定节点下的所有第一级的所有子节点
ls path [watch]
path 表示的是指定数据节点的节点路径
image.png
ls /
第一次部署集群,默认在根节点 "/" 下面又一个叫/zookeeper的保留节点
image.png
get
get path [watch]
image.png
其中
czxid. 节点创建时的zxid.
mzxid. 节点最新一次更新发生时的zxid.
ctime. 节点创建时的时间戳.
mtime. 节点最新一次更新发生时的时间戳.
dataVersion. 节点数据的更新次数.
cversion. 其子节点的更新次数.
aclVersion. 节点ACL(授权信息)的更新次数.
ephemeralOwner. 如果该节点为ephemeral节点, ephemeralOwner值表示与该节点绑定的session id. 如果该节点不是ephemeral节点, ephemeralOwner值为0. 至于什么是ephemeral节点, 请看后面的讲述.
dataLength. 节点数据的字节数.
numChildren. 子节点个数.
更新
set
set path data [version]
image.png
执行完上面的命令后,会发现dataversion的值由原来的 0 变成 1.关于ZNode版本的知识后面我再深入浅出
- 删除
delete
delete path [version]
image.png
image.png
由以上两个示例应该很容易理解为什么删除失败的原因了吧,
刚刚简单的介绍了下命令行的增删改查,
这里不多说了.下面我们进入主要的内容也就是标题上的 Java客户端API的使用
主要要讲的有下面几点
- 创建会话
- 创建节点
- 删除节点
- 读取数据
- 更新数据
- 检测节点是否存在
- 权限控制
zookeeper常用的四个字命令
ZooKeeper 支持某些特定的四字命令字母与其的交互。它们大多是查询命令，用来获取 ZooKeeper 服务的当前状态及相关信息。用户在客户端可以通过 telnet 或 nc 向 ZooKeeper 提交相应的命令
- 可以通过命令：echo stat|nc 127.0.0.1 2181 来查看哪个节点被选择作为follower或者leader
- 使用echo ruok|nc 127.0.0.1 2181 测试是否启动了该Server，若回复imok表示已经启动。
- echo dump| nc 127.0.0.1 2181 ,列出未经处理的会话和临时节点。
- echo kill | nc 127.0.0.1 2181 ,关掉server
- echo conf | nc 127.0.0.1 2181 ,输出相关服务配置的详细信息。
- echo cons | nc 127.0.0.1 2181 ,列出所有连接到服务器的客户端的完全的连接 / 会话的详细信息。
- echo envi |nc 127.0.0.1 2181 ,输出关于服务环境的详细信息（区别于 conf 命令）。
- echo reqs | nc 127.0.0.1 2181 ,列出未经处理的请求。
- echo wchs | nc 127.0.0.1 2181 ,列出服务器 watch 的详细信息。
- echo wchc | nc 127.0.0.1 2181 ,通过 session 列出服务器 watch 的详细信息，它的输出是一个与 watch 相关的会话的列表。
- echo wchp | nc 127.0.0.1 2181 ,通过路径列出服务器 watch 的详细信息。它输出一个与 session 相关的路径。
创建会话
创建一个客户端Zookeeper(org.apache.zookeeper.ZooKeeper)示例,来连接服务器
image.png
参数的详细解释一定要看官方的说明,为了方便我把几处说明复制到了一起,看起了方便很多
/**
* @param connectString
* comma separated host:port pairs, each corresponding to a zk
* server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
* the optional chroot suffix is used the example would look
* like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
* where the client would be rooted at "/app/a" and all paths
* would be relative to this root - ie getting/setting/etc...
* "/foo/bar" would result in operations being run on
* "/app/a/foo/bar" (from the server perspective).
* @param sessionTimeout
* session timeout in milliseconds
* @param watcher
* a watcher object which will be notified of state changes, may
* also be notified for node events
* @param canBeReadOnly
* (added in 3.4) whether the created client is allowed to go to
* read-only mode in case of partitioning. Read-only mode
* basically means that if the client can't find any majority
* servers but there's partitioned server it could reach, it
* connects to one in read-only mode, i.e. read requests are
* allowed while write requests are not. It continues seeking for
* majority in the background.
* @param sessionId
* specific session id to use if reconnecting
* @param sessionPasswd
* password for this session
*/
ZookeeperDemo.java
创建一个最基本的Zookeeper的会话实例
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
public class ZookeeperDemo {
private static Logger logger = LoggerFactory.getLogger(ZookeeperDemo.class);
//同步工具类
private static CountDownLatch latch = new CountDownLatch(1);
public static void main(String[] args) {
try {
//实例Zookeeper
ZooKeeper zooKeeper = new ZooKeeper("192.168.131.134:2181", 3000,new MyWatcher(latch));
logger.info("zooKeeper: [{}]" , zooKeeper);
latch.await();
Thread.sleep(2000);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
MyWatcher.java
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import java.util.concurrent.CountDownLatch;
public class MyWatcher implements Watcher {
private CountDownLatch latch;
public MyWatcher(CountDownLatch latch) {
this.latch=latch;
}
@Override
public void process(WatchedEvent event) {
if (Event.KeeperState.SyncConnected == event.getState()){
System.out.println(event);
latch.countDown();
}
}
}
在源码中有这么一段代码是KeepState的描述,从字面就可以理解其含义
public static KeeperState fromInt(int intValue) {
switch(intValue) {
case -1: return KeeperState.Unknown;
case 0: return KeeperState.Disconnected;
case 1: return KeeperState.NoSyncConnected;
case 3: return KeeperState.SyncConnected;
case 4: return KeeperState.AuthFailed;
case 5: return KeeperState.ConnectedReadOnly;
case 6: return KeeperState.SaslAuthenticated;
case -112: return KeeperState.Expired;
default:
throw new RuntimeException("Invalid integer value for conversion to KeeperState");
}
}
我们在构造函数的注释中发现Zookeeper的构造方法允许传入sessionId和sessionPasswd.其目的是为了复用会话,以维持之前会话的有效性.下面的例子就是复用sessionId和sessionPasswd来创建一个Zookeeper对象实例
public static void main(String[] args) {
try {
ZooKeeper zooKeeper1 = new ZooKeeper("192.168.131.134:2181", 3000,new MyWatcher(latch));
logger.info("zooKeeper 1:[{}]" , zooKeeper1);
latch.await();
long sessionId = zooKeeper1.getSessionId();
byte[] passwd = zooKeeper1.getSessionPasswd();
logger.info("zooKeeper=> sessionId:[{}], passwd:[{}]" , sessionId, new String(passwd));
ZooKeeper zooKeeper2 = new ZooKeeper("192.168.131.134:2181", 3000,new MyWatcher2(latch1),sessionId,passwd);
logger.info("zooKeeper 2:[{}]" , zooKeeper2);
Thread.sleep(2000);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
image.png
创建节点
两个节点创建分别以同步和异步的方式来创建节点
image.png
/**
* @param path
* the path for the node
* @param data
* the initial data for the node
* @param acl
* the acl for the node
* @param createMode
* specifying whether the node to be created is ephemeral
* and/or sequential
* @param cb
* 回调函数
* @param ctx
* 用于传递一个对象的上下文,在回调方法执行的时候使用(通常放一个上下文(Context)信息)
*/
ZookeeperDemo.java
同步接口创建节点
public class ZookeeperDemo {
private static Logger logger = LoggerFactory.getLogger(ZookeeperDemo.class);
private static CountDownLatch latch = new CountDownLatch(1);
public static void main(String[] args) {
try {
ZooKeeper zooKeeper = new ZooKeeper("192.168.131.134:2181", 3000,new MyWatcher(latch));
latch.await();
//同步创建
String path1 = zooKeeper.create("/zk-test-e-", //path
"".getBytes(), //data
Ids.OPEN_ACL_UNSAFE, //acl
CreateMode.EPHEMERAL); //createMode
logger.info("path 1: [{}]" , path1);
String path2 = zooKeeper.create("/zk-test-e-child","".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
logger.info("path 2: [{}]" , path2);
Thread.sleep(2000);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
}
image.png
ZookeeperDemo.java
异步接口创建节点
public class ZookeeperDemo {
private static Logger logger = LoggerFactory.getLogger(ZookeeperDemo.class);
private static CountDownLatch latch = new CountDownLatch(1);
public static void main(String[] args) {
try {
ZooKeeper zooKeeper = new ZooKeeper("192.168.131.134:2181", 3000,new MyWatcher(latch));
latch.await();
//异步创建
zooKeeper.create("/zk-test-e-", //path
"".getBytes(), //data
Ids.OPEN_ACL_UNSAFE, //acl
CreateMode.EPHEMERAL, //createMode
new MyStringCallback(), //callback
"I am Context"); //Context
zooKeeper.create("/zk-test-e-child", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new MyStringCallback(), "I am Context");
Thread.sleep(2000);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
import org.apache.zookeeper.AsyncCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 异步回调函数实现类
*/
public class MyStringCallback implements AsyncCallback.StringCallback {
private static Logger logger = LoggerFactory.getLogger(MyStringCallback.class);
@Override
public void processResult(int rc, String path, Object ctx, String name) {
logger.info("Create path result:[{}] path:[{}] ctx:[{}] name:[{}]", rc, path, ctx, name);
}
}
image.png
在CreateMode 的源码中的这段信息
- 持久(PERSISTENT)
- 持久顺序(PERSISTENT_SEQUENTIAL)
- 临时(EPHEMERAL)
- 临时顺序(EPHEMERAL_SEQUENTIAL )
/**
* The znode will not be automatically deleted upon client's disconnect.
*/
PERSISTENT (0, false, false),
/**
* The znode will not be automatically deleted upon client's disconnect,
* and its name will be appended with a monotonically increasing number.
*/
PERSISTENT_SEQUENTIAL (2, false, true),
/**
* The znode will be deleted upon the client's disconnect.
*/
EPHEMERAL (1, true, false),
/**
* The znode will be deleted upon the client's disconnect, and its name
* will be appended with a monotonically increasing number.
*/
EPHEMERAL_SEQUENTIAL (3, true, true);
创建节点的回调函数包含了七种不同的回调接口.所有我们要在不同的异步接口中实现不同的接口
- StatCallback
- DataCallback
- ACLCallback
- ChildrenCallback
- Children2Callback
- StringCallback
- VoidCallback
回调函数方法参数说明
processResult(int rc, String path, Object ctx, String name)
| 参数 | 说明 |
|---|---|
| rc | Result Code, 服务器响应码 [0] :调用成功 [-4] 客户端和服务端连接断开 [-110] 指定节点已存在 [-112] 会话已经过期 |
| path | 接口调用时传入API的数据节点的节点路径参数值 |
| ctx | 接口调用时传入API的ctx参数值 |
| name | 实际在服务端创建的节点名 |
删除节点
删除节点Zookeeper API 提供了2个接口,如图
image.png
参数的含义
/**
* @param path
* the path of the node to be deleted.
* @param version
* the expected node version.
* @param cb
* 回调函数
* @param ctx
* 用于传递一个对象的上下文,在回调方法执行的时候使用(通常放一个上下文(Context)信息)
*/
先来查看下我们现有服务器上的节点,删除标红的节点
image.png
public class ZookeeperDeleteNode {
private static CountDownLatch latch = new CountDownLatch(1);
private static CountDownLatch latch1 = new CountDownLatch(1);
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
ZooKeeper zooKeeper = new ZooKeeper("192.168.131.134:2181", 3000,new MyWatcher(latch));
latch.await();
zooKeeper.delete("/zk-test-e-child",
0,
new MyVoidCallback(latch1),
"I am Context");
latch1.await();
}
}
public class MyVoidCallback implements AsyncCallback.VoidCallback {
private static Logger logger = LoggerFactory.getLogger(MyVoidCallback.class);
private CountDownLatch latch;
public MyVoidCallback(CountDownLatch latch){
this.latch = latch;
}
@Override
public void processResult(int rc, String path, Object ctx) {
logger.info("delete path result:[{}] path:[{}] ctx:[{}] ", rc, path, ctx);
latch.countDown();
}
}
删除成功
image.png
删除失败
已经删除过了.
image.png
读取数据
读取节点Zookeeper API 提供了8个接口,如图
image.png
/**
* @param path
* @param watcher explicit watcher
* @param watch whether need to watch this node
* @param cb a handler for the callback
* @param ctx context to be provided to the callback
* @param stat stat of the znode designated by path
*/
持续更新中.......(未完待续)
