- package com.git.zookeeper.passwordmanager.listener;
- import java.util.ArrayList;
- import java.util.List;
- import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Executors;
- import org.apache.curator.RetryPolicy;
- import org.apache.curator.framework.CuratorFramework;
- import org.apache.curator.framework.CuratorFrameworkFactory;
- import org.apache.curator.framework.api.ACLProvider;
- import org.apache.curator.framework.api.CuratorEvent;
- import org.apache.curator.framework.api.CuratorListener;
- import org.apache.curator.framework.recipes.cache.ChildData;
- import org.apache.curator.framework.recipes.cache.NodeCache;
- import org.apache.curator.framework.recipes.cache.NodeCacheListener;
- import org.apache.curator.framework.recipes.cache.PathChildrenCache;
- import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
- import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
- import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
- import org.apache.curator.framework.recipes.cache.TreeCache;
- import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
- import org.apache.curator.framework.recipes.cache.TreeCacheListener;
- import org.apache.curator.retry.ExponentialBackoffRetry;
- import org.apache.curator.retry.RetryNTimes;
- import org.apache.zookeeper.WatchedEvent;
- import org.apache.zookeeper.Watcher;
- import org.apache.zookeeper.ZooDefs;
- import org.apache.zookeeper.ZooDefs.Perms;
- import org.apache.zookeeper.data.ACL;
- import org.apache.zookeeper.data.Id;
- /**
- * curator管理zookeeper的相关方法工具类
- * @author songqinghu
- * 要点:
- * 1.连接的建立 (两种 OK--使用权限方式)
- * 2.节点的管理操作,增删改查--层叠节点的操作(OK --设立命名空间)
- * 3.节点的监听操作,和无限监听事件触发(OK --使用第三种方法)
- * 4.节点的访问控制ACL操作,密码的添加,修改-->连接中设置
- * 5.节点原子性操作
- * 6.节点的分布式锁操作
- * 7.分布式队列
- */
- public class CuratorListenerUtils {
- /**
- * 先测试玩玩
- * @描述:XXXXXXX
- * @param args
- * @return void
- * @exception
- * @createTime:2016年5月17日
- * @author: songqinghu
- * @throws Exception
- */
- public static void main(String[] args) throws Exception {
- CuratorFramework client = clientTwo();
- //setListenterDateNode();
- //setListenterThreeOne(client);
- // setListenterThreeTwo(client);
- setListenterThreeThree(client);
- // getDataNode(client, "/two");
- // setDataNode(client, "/two", "sss");
- Thread.sleep(Long.MAX_VALUE);
- }
- /**
- *
- * @描述:创建一个zookeeper连接---连接方式一: 最简单的连接
- * @return void
- * @exception
- * @createTime:2016年5月17日
- * @author: songqinghu
- */
- private static CuratorFramework clientOne(){
- //zk 地址
- String connectString = "10.125.2.44:2181";
- // 连接时间 和重试次数
- RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
- CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
- client.start();
- return client;
- }
- /**
- *
- * @描述:创建一个zookeeper连接---连接方式二:优选这个
- * @return void
- * @exception
- * @createTime:2016年5月17日
- * @author: songqinghu
- */
- private static CuratorFramework clientTwo(){
- //默认创建的根节点是没有做权限控制的--需要自己手动加权限???----
- ACLProvider aclProvider = new ACLProvider() {
- private List<ACL> acl ;
- @Override
- public List<ACL> getDefaultAcl() {
- if(acl ==null){
- ArrayList<ACL> acl = ZooDefs.Ids.CREATOR_ALL_ACL;
- acl.clear();
- acl.add(new ACL(Perms.ALL, new Id("auth", "admin:admin") ));
- this.acl = acl;
- }
- return acl;
- }
- @Override
- public List<ACL> getAclForPath(String path) {
- return acl;
- }
- };
- String scheme = "digest";
- byte[] auth = "admin:admin".getBytes();
- int connectionTimeoutMs = 5000;
- String connectString = "10.125.2.44:2181";
- String namespace = "testnamespace";
- CuratorFramework client = CuratorFrameworkFactory.builder().aclProvider(aclProvider).
- authorization(scheme, auth).
- connectionTimeoutMs(connectionTimeoutMs).
- connectString(connectString).
- namespace(namespace).
- retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000)).build();
- client.start();
- return client;
- }
- /**
- *
- * @描述:第一种监听器的添加方式: 对指定的节点进行添加操作
- * 仅仅能监控指定的本节点的数据修改,删除 操作 并且只能监听一次 --->不好
- * @return void
- * @exception
- * @createTime:2016年5月18日
- * @author: songqinghu
- * @throws Exception
- */
- private static void setListenterOne(CuratorFramework client) throws Exception{
- // 注册观察者,当节点变动时触发
- byte[] data = client.getData().usingWatcher(new Watcher() {
- @Override
- public void process(WatchedEvent event) {
- System.out.println("获取 two 节点 监听器 : " + event);
- }
- }).forPath("/two");
- System.out.println("two 节点数据: "+ new String(data));
- }
- /**
- *
- * @描述:第二种监听器的添加方式:
- * 也是一次性的监听操作,使用后就无法在继续监听了
- * @return void
- * @exception
- * @createTime:2016年5月18日
- * @author: songqinghu
- * @throws Exception
- */
- private static void setListenterTwo(CuratorFramework client) throws Exception{
- ExecutorService pool = Executors.newCachedThreadPool();
- CuratorListener listener = new CuratorListener() {
- @Override
- public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
- System.out.println("监听器 : "+ event.toString());
- }
- };
- client.getCuratorListenable().addListener(listener,pool);
- client.getData().inBackground().forPath("/two");
- client.getData().inBackground().forPath("/two");
- client.getData().inBackground().forPath("/two");
- client.getData().inBackground().forPath("/two");
- Thread.sleep(Long.MAX_VALUE );
- }
- /**
- *
- * @描述:第三种监听器的添加方式: Cache 的三种实现 实践
- * Path Cache:监视一个路径下1)孩子结点的创建、2)删除,3)以及结点数据的更新。
- * 产生的事件会传递给注册的PathChildrenCacheListener。
- * Node Cache:监视一个结点的创建、更新、删除,并将结点的数据缓存在本地。
- * Tree Cache:Path Cache和Node Cache的“合体”,监视路径下的创建、更新、删除事件,并缓存路径下所有孩子结点的数据。
- * @return void
- * @exception
- * @createTime:2016年5月18日
- * @author: songqinghu
- * @throws Exception
- */
- //1.path Cache 连接 路径 是否获取数据
- //能监听所有的字节点 且是无限监听的模式 但是 指定目录下节点的子节点不再监听
- private static void setListenterThreeOne(CuratorFramework client) throws Exception{
- ExecutorService pool = Executors.newCachedThreadPool();
- PathChildrenCache childrenCache = new PathChildrenCache(client, "/test", true);
- PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
- @Override
- public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
- System.out.println("开始进行事件分析:-----");
- ChildData data = event.getData();
- switch (event.getType()) {
- case CHILD_ADDED:
- System.out.println("CHILD_ADDED : "+ data.getPath() +" 数据:"+ data.getData());
- break;
- case CHILD_REMOVED:
- System.out.println("CHILD_REMOVED : "+ data.getPath() +" 数据:"+ data.getData());
- break;
- case CHILD_UPDATED:
- System.out.println("CHILD_UPDATED : "+ data.getPath() +" 数据:"+ data.getData());
- break;
- default:
- break;
- }
- }
- };
- childrenCache.getListenable().addListener(childrenCacheListener);
- System.out.println("Register zk watcher successfully!");
- childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
- }
- //2.Node Cache 监控本节点的变化情况 连接 目录 是否压缩
- //监听本节点的变化 节点可以进行修改操作 删除节点后会再次创建(空节点)
- private static void setListenterThreeTwo(CuratorFramework client) throws Exception{
- ExecutorService pool = Executors.newCachedThreadPool();
- //设置节点的cache
- final NodeCache nodeCache = new NodeCache(client, "/test", false);
- nodeCache.getListenable().addListener(new NodeCacheListener() {
- @Override
- public void nodeChanged() throws Exception {
- System.out.println("the test node is change and result is :");
- System.out.println("path : "+nodeCache.getCurrentData().getPath());
- System.out.println("data : "+new String(nodeCache.getCurrentData().getData()));
- System.out.println("stat : "+nodeCache.getCurrentData().getStat());
- }
- });
- nodeCache.start();
- }
- //3.Tree Cache
- // 监控 指定节点和节点下的所有的节点的变化--无限监听 可以进行本节点的删除(不在创建)
- private static void setListenterThreeThree(CuratorFramework client) throws Exception{
- ExecutorService pool = Executors.newCachedThreadPool();
- //设置节点的cache
- TreeCache treeCache = new TreeCache(client, "/test");
- //设置监听器和处理过程
- treeCache.getListenable().addListener(new TreeCacheListener() {
- @Override
- public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
- ChildData data = event.getData();
- if(data !=null){
- switch (event.getType()) {
- case NODE_ADDED:
- System.out.println("NODE_ADDED : "+ data.getPath() +" 数据:"+ new String(data.getData()));
- break;
- case NODE_REMOVED:
- System.out.println("NODE_REMOVED : "+ data.getPath() +" 数据:"+ new String(data.getData()));
- break;
- case NODE_UPDATED:
- System.out.println("NODE_UPDATED : "+ data.getPath() +" 数据:"+ new String(data.getData()));
- break;
- default:
- break;
- }
- }else{
- System.out.println( "data is null : "+ event.getType());
- }
- }
- });
- //开始监听
- treeCache.start();
- }
- }
zookeeper入门之Curator的使用之几种监听器的使用
猜你喜欢
转载自www.cnblogs.com/exmyth/p/9013835.html
今日推荐
周排行