watch机制
Zookeeper watch是一种监听通知机制，可以随时监听一些数据的变化，从而实现数据的及时性。
Zookeeper所有的读操作getData(), getChildren()和 exists()都可以设置监听(watch)。【写操作则是不能设置监视点的。】
Watch的三个关键点：
- 一次有效：当设置监视的数据发生改变时，该监视事件会被发送到客户端，并且该监听将会停止，除非重启注册监听；
- 顺序保证：网络延迟或者其他因素可能导致不同的客户端在不同的时刻感知某一监视事件，但是不同的客户端所看到的一切具有一致的顺序；
- 可以监听数据和子节点：getData()和 exists()可以设置监听数据变化；getChildren 可以设置监听子节点变化；
常用API
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException
public String create(String path, byte[] data, List<ACL> acl, CreateMode createMode) throws KeeperException, InterruptedException
public List<String> getChildren(String path, boolean watch) throws KeeperException, InterruptedException
public byte[] getData(String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException
public void getData(String path, boolean watch, AsyncCallback.DataCallback cb, Object ctx)
public Stat setData(String path, byte[] data, int version) throws KeeperException, InterruptedException
public Stat exists(String path, boolean watch) throws KeeperException, InterruptedException
public void delete(String path, int version) throws InterruptedException, KeeperExceptionJAVA调用
初始化
try {
ZooKeeper zooKeeper = new ZooKeeper("172.23.34.13:2181", 15000, event -> {
if (event.getType() == Watcher.Event.EventType.None && event.getState() == Watcher.Event.KeeperState.SyncConnected) {
System.out.println("Connectted successful.");
}
});
} catch (IOException e) {
e.printStackTrace();
}
创建节点: create
@Test
public void create() throws KeeperException, InterruptedException {
//参数：1，节点路径； 2，要存储的数据； 3，节点的权限； 4，节点的类型
String nodePath = zooKeeper.create("/java/2183", "This is Java Node 2183.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println(nodePath);
}
获取子节点： ls
public void getChildren() throws KeeperException, InterruptedException {
List<String> children = zooKeeper.getChildren("/", true);
for (String child : children) {
System.out.println("child: "+child);
}
}
同步获取节点内容： get
@Test
public void getData() throws KeeperException, InterruptedException {
String path = "/java";
byte[] bytes = zooKeeper.getData(path, event -> {
if (event.getType() == Watcher.Event.EventType.NodeDataChanged && path.equals(event.getPath())) {
System.out.println("Date changed.");
}
}, null);
System.out.printf("The data of %s is : %s \n",path, new String(bytes));
}
异步获取节点内容： get
@Test
public void getDataAsync() {
String path = "/java";
zooKeeper.getData(path, false, new AsyncCallback.DataCallback() {
@Override
public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
System.out.printf("The data of %s is : %s \n",path, new String(bytes));
}
},"1000");
//休眠20秒，查看响应结果
try {
Thread.sleep(20000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
指定版本号更新数据：set
@Test
public void setData() throws KeeperException, InterruptedException {
Stat stat = zooKeeper.setData("/java", "This is from java.".getBytes(), -1);
//更新节点后，version会自动+1。故，返回值为0
System.out.println(stat.getAversion());
}
多线程下更新数据：set
@Test
public void setDataThread() throws KeeperException, InterruptedException {
String path = "/java";
Stat stat = new Stat();
//1,先获取节点的当前版本
zooKeeper.getData(path,false,stat);
//2，在当前版本的基础上修改节点内容
zooKeeper.setData(path, "This is from java.".getBytes(), stat.getVersion());
}
判断节点是否存在
@Test
public void exists() throws KeeperException, InterruptedException {
Stat stat = zooKeeper.exists("/java", false);
if (stat == null) {
System.out.println("Not Exists.");
}else {
System.out.println("Exists.");
}
}
删除节点
@Test
public void delete() throws KeeperException, InterruptedException {
//version = -1 : 匹配所有的版本
zooKeeper.delete("/java/2182", -1);
}
到此这篇关于Java调用Zookeeper的实现步骤的文章就介绍到这了,更多相关Java调用Zookeeper内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！
