zookeeper的api使用
将zookeeper服务全部起来使用API来来替代命令行中的增删改查操作比如:
创建节点向节点中写入描述信息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
private ZooKeeper zk = null;
@Before
public void init() throws Exception {
zk = new ZooKeeper("192.168.0.203:2181", 2000, new Watcher() {
/** 监听事件发生时的回调方法*/
@Override
public void process(WatchedEvent event) {
if (event.getType() == EventType.None)
return;
System.out.println(event.getType());
System.out.println(event.getPath());
try {
zk.getData("/test1", true, null);
zk.getChildren("/test1", true);
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
});
}
/**
* 向zookeeper服务集群中注册数据，添加znode
*/
@Test
public void testCreateZnode() throws UnsupportedEncodingException, KeeperException, InterruptedException {
// 该节点如果没创建了下次在创建就会报错
//zk.create("/test1", "diyigeceshi".getBytes("utf-8"), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
/** 第一个参数:在已有的节点中创建文件节点
第二个参数:写入对该节点的描述信息
第三个参数:是否开发写入操作
第四个参数:指定该节点的生命周期,sequential的顺序维护是在一个父节点的范围之内*/
zk.create("/test1/a", "zijiedianA".getBytes("utf-8"), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
zk.create("/test1/b", "zijiedianB".getBytes("utf-8"), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
// 换一个父节点，序号的递增顺序重新开始
//zk.create("/test2/a", "如诗".getBytes("utf-8"), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
zk.close();
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
* 从zkeeper中删除znode
*
* @throws Exception
*/
@Test
public void testDeleteZnode() throws Exception {
// 参数1：要删除的节点的路径 参数2：要删除的节点的版本，-1匹配所有版本
zk.delete("/test1", -1);
//false是否监听
Stat exists = zk.exists("/sh18qi", false);
System.out.println(exists);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/**
* 更新节点
* @throws Exception
*/
@Test
public void testUpdateZnode() throws Exception {
byte[] data = zk.getData("/test1", false, null);
System.out.println(new String(data, "utf-8"));
zk.setData("/test1", "我有一头小毛驴我从来也不骑".getBytes("utf-8"), -1);
//false 是否监听 获取哪个版本 如果为null获取最新的
data = zk.getData("/test1", false, null);
System.out.println(new String(data, "utf-8"));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
* 获取子节点信息
* @throws Exception
*/
@Test
public void testGetChildren() throws Exception {
List<String> children = zk.getChildren("/test1", false);
for (String child : children) {
byte[] data = zk.getData("/test1/"+child, false, null);
System.out.println("子节点名称:"+child+"子节点内容: "+new String(data, "utf-8"));
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
/**
* zk的监听机制：
* 1、事先定义好监听的回调函数
* 2、在对znode进行各种访问操作时可以注册监听
* 3、监听的znode上发生相应事件时，客户端zk会接收到zookeeper集群的事件通知
* 4、客户端zk根据事件调用我们事先定义好的回调函数
*/
public class ZKDemo {
private ZooKeeper zk = null;
@Before
public void init() throws Exception {
zk = new ZooKeeper("192.168.0.203:2181", 2000, new Watcher() {
/** 监听事件发生时的回调方法*/
@Override
public void process(WatchedEvent event) {
if (event.getType() == EventType.None)
return;
System.out.println(event.getType());
System.out.println(event.getPath());
try {
zk.getData("/test1", true, null);
zk.getChildren("/test1", true);
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
});
}
@Test
public void testWatch() throws KeeperException, InterruptedException {
// 在获取znode数据时注册了监听
// 监听器是一次性，只要监听到一次事件，就失效了
// getData监听的事件是数据的更改
byte[] data = zk.getData("/sh18qi2", true, null);
// 在做查询子节点操作时注册监听
// 监听的事件就是监听节点下的子节点变化事件
//监听该节点下的子节点变化 并不监听子节点数据变化
List<String> children = zk.getChildren("/sh18qi2", true);
Thread.sleep(Long.MAX_VALUE);
}
}
页:
[1]