您好,登錄后才能下訂單哦!
maven依賴:
<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>per.ym</groupId>
<artifactId>zk</artifactId>
<version>0.0.1-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.12</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
測試類:
package per.ym.zookeeper;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
public class ZkTest implements Watcher{
private ZooKeeper zk;
private CountDownLatch cdl;
private String path = "/test";
private String rootPath = "/";
private int sessionTimeOut = 15000;
private byte[] data = "data".getBytes();
private byte[] newData = "newData".getBytes();
@Before
public void connect() throws IOException {
zk = new ZooKeeper("192.168.61.131:2184", sessionTimeOut, this);
cdl = new CountDownLatch(1);
}
//同步調用
@Test
public void testSync( ) throws Exception {
//等待與zookeeper服務端連接完成
cdl.await();
//創建一個持久節點/test,并為其賦值為data
zk.create("/test", data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//獲取根節點下的子節點
List<String> children = zk.getChildren("/", this);
System.out.println("根節點下的子節點有: " + children);
//查詢節點/test保存的數據
Stat stat = new Stat();
byte[] tempData = zk.getData(path, this, stat);
System.out.println("test節點數據為: " + new String(tempData));
//設置節點/test的數據為newData
zk.setData(path, newData, -1);
tempData = zk.getData(path, this, stat);
System.out.println("test節點新數據為: " + new String(tempData));
//刪除節點/test
zk.delete(path, -1);
//判斷節點/test是否存在
stat = zk.exists(path, this);
System.out.println(stat);
}
//創建節點后進行回調
private StringCallback cb = new StringCallback() {
@Override
public void proce***esult(int rc, String path, Object ctx, String name) {
switch (Code.get(rc)) {
case CONNECTIONLOSS:
//如果失去連接,我們無法保證是在創建前還是創建后丟失的,因此重試
create();
break;
case OK:
System.out.println("節點/test創建成功");
break;
case NODEEXISTS:
System.out.println("節點/test已經存在");
break;
default:
System.out.println(KeeperException.create(Code.get(rc), path));
}
}
};
//獲取子節點進行回調
private ChildrenCallback ccb = new ChildrenCallback() {
@Override
public void proce***esult(int rc, String path, Object ctx, List<String> children) {
switch (Code.get(rc)) {
case CONNECTIONLOSS:
getChildren();
break;
case OK:
System.out.println("根節點下的子節點有: " + children);
break;
default:
System.out.println(KeeperException.create(Code.get(rc), path));
}
}
};
//獲取數據進行回調
private DataCallback dcb = new DataCallback() {
@Override
public void proce***esult(int rc, String path, Object ctx, byte[] data, Stat stat) {
switch (Code.get(rc)) {
case CONNECTIONLOSS:
getData();
break;
case OK:
System.out.println("test節點數據為: " + new String(data));
break;
default:
System.out.println(KeeperException.create(Code.get(rc), path));
}
}
};
//設置數據進行回調
private StatCallback scb = new StatCallback() {
@Override
public void proce***esult(int rc, String path, Object ctx, Stat stat) {
switch (Code.get(rc)) {
case CONNECTIONLOSS:
setData();
break;
case OK:
//這個ctx就是我們調用zk.setData時傳入的最后一個參數
System.out.println("test節點設置新數 " + new String((byte[])ctx) + "成功");
break;
default:
System.out.println(KeeperException.create(Code.get(rc), path));
}
}
};
//刪除節點進行回調
private VoidCallback vcb = new VoidCallback() {
@Override
public void proce***esult(int rc, String path, Object ctx) {
switch (Code.get(rc)) {
case CONNECTIONLOSS:
delete();
break;
case OK:
System.out.println("刪除節點/test成功");
break;
default:
System.out.println(KeeperException.create(Code.get(rc), path));
}
}
};
//節點是否存在進行回調
private StatCallback scb2 = new StatCallback() {
@Override
public void proce***esult(int rc, String path, Object ctx, Stat stat) {
switch (Code.get(rc)) {
case CONNECTIONLOSS:
exists();
break;
case OK:
System.out.println("節點/test存在");
break;
case NONODE:
System.out.println("節點/test不存在");
break;
default:
break;
}
}
};
//異步調用
@Test
public void testAsync() throws Exception {
cdl.await();
create();
getChildren();
setData();
getData();
exists();
delete();
exists();
}
private void create() {
zk.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, cb, data);
}
private void getChildren() {
zk.getChildren(rootPath, this, ccb, null);
}
private void getData() {
zk.getData(path, this, dcb, null);
}
private void setData() {
zk.setData(path, newData, -1, scb, newData);
}
private void delete() {
zk.delete(path, -1, vcb, null);
}
private void exists() {
zk.exists(path, this, scb2, null);
}
@Override
public void process(WatchedEvent event) {
//如果連接成功,放行
if (event.getType().equals(EventType.None)) {
cdl.countDown();
}
System.out.println(event);
}
@After
public void close() throws InterruptedException {
//測試完成后關閉連接
zk.close();
}
}
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。