zookeeper分布式集群中Curator的TreeCache监测节点数据更新添加删除
zookeeper分布式集群中Curator的TreeCache监测节点数据更新添加删除import org.apache.curator.RetryPolicy;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFacto...
·
zookeeper分布式集群中Curator的TreeCache监测节点数据更新添加删除
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import java.util.concurrent.TimeUnit;
public class Main {
private final String TAG = "调试# ";
public static void main(String[] args) {
//初始化log4j,zookeeper否则报错。
//org.apache.log4j.BasicConfigurator.configure();
try {
Main m = new Main();
} catch (Exception e) {
e.printStackTrace();
}
}
private String getAddress() {
String ip = "127.0.0.1";
return ip + ":2181," + ip + ":2182," + ip + ":2183";
}
public Main() throws Exception {
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString(getAddress())
.sessionTimeoutMs(10 * 1000)
.connectionTimeoutMs(20 * 1000)
.retryPolicy(retryPolicy)
.build();
client.start();
String path = "/test_data/my_path";
//检测是否存在该路径。
Stat stat = client.checkExists().forPath(path);
//如果不存在这个路径,stat为null,创建新的节点路径。
if (stat == null) {
String s = client.create()
.creatingParentsIfNeeded()
.withMode(CreateMode.PERSISTENT)
.forPath(path);
System.out.println(TAG + "已创建:" + s);
} else {
System.out.println(TAG + "已存在:" + path);
}
TreeCache treeCache = new TreeCache(client, path);
TreeCacheListener treeCacheListener = new TreeCacheListener() {
@Override
public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
System.out.println("=============");
System.out.println(TAG + "路径:" + event.getData().getPath());
System.out.println(TAG + "数据:" + new String(event.getData().getData(), "UTF-8"));
System.out.println(TAG + "ZooKeeper:" + client.getZookeeperClient().getZooKeeper().toString());
}
};
treeCache.getListenable().addListener(treeCacheListener);
treeCache.start();
client.setData().forPath(path, "hello,world!".getBytes());
TimeUnit.SECONDS.sleep(5);
client.setData().forPath(path, "zhang phil".getBytes());
TimeUnit.SECONDS.sleep(5);
client.delete().deletingChildrenIfNeeded().forPath(path);
TimeUnit.SECONDS.sleep(5);
treeCache.close();
client.close();
System.out.println("结束");
}
}
输出:
调试# 已创建:/test_data/my_path
=============
调试# 路径:/test_data/my_path
调试# 数据:172.17.238.45
调试# ZooKeeper:State:CONNECTED Timeout:10000 sessionid:0x300001d3a5a0014 local:/127.0.0.1:53720 remoteserver:127.0.0.1/127.0.0.1:2183 lastZxid:38654705956 xid:11 sent:11 recv:10 queuedpkts:0 pendingresp:1 queuedevents:0
=============
=============
调试# 路径:/test_data/my_path
调试# 数据:hello,world!
调试# ZooKeeper:State:CONNECTED Timeout:10000 sessionid:0x300001d3a5a0014 local:/127.0.0.1:53720 remoteserver:127.0.0.1/127.0.0.1:2183 lastZxid:38654705957 xid:12 sent:12 recv:13 queuedpkts:0 pendingresp:0 queuedevents:0
=============
调试# 路径:/test_data/my_path
调试# 数据:zhang phil
调试# ZooKeeper:State:CONNECTED Timeout:10000 sessionid:0x300001d3a5a0014 local:/127.0.0.1:53720 remoteserver:127.0.0.1/127.0.0.1:2183 lastZxid:38654705958 xid:14 sent:15 recv:17 queuedpkts:0 pendingresp:0 queuedevents:0
=============
调试# 路径:/test_data/my_path
调试# 数据:zhang phil
调试# ZooKeeper:State:CONNECTED Timeout:10000 sessionid:0x300001d3a5a0014 local:/127.0.0.1:53720 remoteserver:127.0.0.1/127.0.0.1:2183 lastZxid:38654705959 xid:15 sent:17 recv:20 queuedpkts:0 pendingresp:0 queuedevents:0
结束
更多推荐
已为社区贡献3条内容
所有评论(0)