本文主要是介绍Curator实现zookeeper的节点监听,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
Curator实现zookeeper的节点监听
Curtor框架中一共有三个实现监听的方式
一种是NodeCache监听指定节点
一种是pathChildrenCache监听子节点
一种是TreeCache可以监控所有节点 相当于以上两种的合集
引入依赖
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.3.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-client</artifactId><version>4.3.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.3.0</version></dependency>
创建一个测试类 连接好客户端
public class CuratorTest {private CuratorFramework curatorFramework;@Beforepublic void testCreate() {ExponentialBackoffRetry retry = new ExponentialBackoffRetry(2, 10);curatorFramework = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").connectionTimeoutMs(60 * 1000).sessionTimeoutMs(15 * 10000).retryPolicy(retry).build();curatorFramework.start();}@Afterpublic void close() {if (curatorFramework != null) {curatorFramework.close();}}
}
Watch监听之NodeCache
监听一个指定节点
@Testpublic void testUpdate() throws Exception {//监听一个节点NodeCache nodeCache = new NodeCache(curatorFramework,"/dongwuyuan");//注册监听nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化了!!!!");}});//开启监听 参数 如果设置为true 则开启监听时加载缓存数据nodeCache.start(true);while (true){}}
set /dongwuyuan "laohu"
Watch监听之PathChildrenCache
监听子节点的变化
@Testpublic void testPathChildrenCache() throws Exception {// 参数 客户端,路径 ,缓存数据,是否压缩,线程池PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework,"/dongwuyuan",true);//绑定监听器pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {System.out.println("子节点变化了");System.out.println(pathChildrenCacheEvent);//监听子节点的变更,并且拿到变更后的数据PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();//判断类型是否是updateif(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){//拿到数据byte[] data = pathChildrenCacheEvent.getData().getData();System.out.println(data);}}});//开启监听pathChildrenCache.start();while (true){}}
watch监听之TreeCache
/*** TreeCache:监听节点自己和所有子节点们*/@Testpublic void testTreeCache(){//1.创建监听器TreeCache treeCache = new TreeCache(curatorFramework, "/dongwuyuan");//2.注册监听treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {System.out.println("节点变化了");System.out.println(event);}});//开启监听try {treeCache.start();while (true){}} catch (Exception e) {e.printStackTrace();}}
[zk: localhost:2181(CONNECTED) 13] delete /dongwuyuan/node1
[zk: localhost:2181(CONNECTED) 15] set /dongwuyuan/node2 "shizi"
完整代码
public class CuratorTest {private CuratorFramework curatorFramework;@Beforepublic void testCreate() {ExponentialBackoffRetry retry = new ExponentialBackoffRetry(2, 10);curatorFramework = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").connectionTimeoutMs(60 * 1000).sessionTimeoutMs(15 * 10000).retryPolicy(retry).build();curatorFramework.start();}@Testpublic void testUpdate() throws Exception {//监听一个节点NodeCache nodeCache = new NodeCache(curatorFramework,"/dongwuyuan");//注册监听nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化了!!!!");}});//开启监听 参数 如果设置为true 则开启监听时加载缓存数据nodeCache.start(true);while (true){}}@Testpublic void testPathChildrenCache() throws Exception {// 参数 客户端,路径 ,缓存数据,是否压缩,线程池PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework,"/dongwuyuan",true);//绑定监听器pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {System.out.println("子节点变化了");System.out.println(pathChildrenCacheEvent);//监听子节点的变更,并且拿到变更后的数据PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();//判断类型是否是updateif(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){//拿到数据byte[] data = pathChildrenCacheEvent.getData().getData();System.out.println(data);}}});//开启监听pathChildrenCache.start();while (true){}}/*** TreeCache:监听节点自己和所有子节点们*/@Testpublic void testTreeCache(){//1.创建监听器TreeCache treeCache = new TreeCache(curatorFramework, "/dongwuyuan");//2.注册监听treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {System.out.println("节点变化了");System.out.println(event);}});//开启监听try {treeCache.start();while (true){}} catch (Exception e) {e.printStackTrace();}}@Afterpublic void close() {if (curatorFramework != null) {curatorFramework.close();}}
}
这篇关于Curator实现zookeeper的节点监听的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!