春节假期过去了,收收心我们来开始继续学习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
*/
持续更新中.......(未完待续)
网友评论