zookeeper入门之Curator的使用之几种监听器的使用

其他 2018-05-09 14:16:11 阅读次数: 2

[java]  view plain  copy
 
    1. package com.git.zookeeper.passwordmanager.listener;  
    2.   
    3. import java.util.ArrayList;  
    4. import java.util.List;  
    5. import java.util.concurrent.ExecutorService;  
    6. import java.util.concurrent.Executors;  
    7.   
    8. import org.apache.curator.RetryPolicy;  
    9. import org.apache.curator.framework.CuratorFramework;  
    10. import org.apache.curator.framework.CuratorFrameworkFactory;  
    11. import org.apache.curator.framework.api.ACLProvider;  
    12. import org.apache.curator.framework.api.CuratorEvent;  
    13. import org.apache.curator.framework.api.CuratorListener;  
    14. import org.apache.curator.framework.recipes.cache.ChildData;  
    15. import org.apache.curator.framework.recipes.cache.NodeCache;  
    16. import org.apache.curator.framework.recipes.cache.NodeCacheListener;  
    17. import org.apache.curator.framework.recipes.cache.PathChildrenCache;  
    18. import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;  
    19. import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;  
    20. import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;  
    21. import org.apache.curator.framework.recipes.cache.TreeCache;  
    22. import org.apache.curator.framework.recipes.cache.TreeCacheEvent;  
    23. import org.apache.curator.framework.recipes.cache.TreeCacheListener;  
    24. import org.apache.curator.retry.ExponentialBackoffRetry;  
    25. import org.apache.curator.retry.RetryNTimes;  
    26. import org.apache.zookeeper.WatchedEvent;  
    27. import org.apache.zookeeper.Watcher;  
    28. import org.apache.zookeeper.ZooDefs;  
    29. import org.apache.zookeeper.ZooDefs.Perms;  
    30. import org.apache.zookeeper.data.ACL;  
    31. import org.apache.zookeeper.data.Id;  
    32.   
    33. /** 
    34.  * curator管理zookeeper的相关方法工具类 
    35.  * @author songqinghu 
    36.  * 要点: 
    37.  * 1.连接的建立  (两种 OK--使用权限方式) 
    38.  * 2.节点的管理操作,增删改查--层叠节点的操作(OK --设立命名空间) 
    39.  * 3.节点的监听操作,和无限监听事件触发(OK --使用第三种方法) 
    40.  * 4.节点的访问控制ACL操作,密码的添加,修改-->连接中设置 
    41.  * 5.节点原子性操作 
    42.  * 6.节点的分布式锁操作 
    43.  * 7.分布式队列 
    44.  */  
    45. public class CuratorListenerUtils {  
    46.   
    47.     /** 
    48.      * 先测试玩玩   
    49.      * @描述：XXXXXXX 
    50.      * @param args 
    51.      * @return void 
    52.      * @exception 
    53.      * @createTime：2016年5月17日 
    54.      * @author: songqinghu 
    55.      * @throws Exception  
    56.      */   
    57.     public static void main(String[] args) throws Exception {  
    58.         CuratorFramework client = clientTwo();  
    59.         //setListenterDateNode();  
    60.         //setListenterThreeOne(client);  
    61.        // setListenterThreeTwo(client);  
    62.         setListenterThreeThree(client);  
    63.        // getDataNode(client, "/two");  
    64.        // setDataNode(client, "/two", "sss");  
    65.         Thread.sleep(Long.MAX_VALUE);  
    66.     }  
    67.   
    68.     /** 
    69.      *  
    70.      * @描述：创建一个zookeeper连接---连接方式一: 最简单的连接 
    71.      * @return void 
    72.      * @exception 
    73.      * @createTime：2016年5月17日 
    74.      * @author: songqinghu 
    75.      */  
    76.     private static CuratorFramework clientOne(){  
    77.         //zk 地址  
    78.         String connectString = "10.125.2.44:2181";  
    79.         // 连接时间 和重试次数  
    80.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);  
    81.         CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);  
    82.         client.start();  
    83.         return client;  
    84.     }  
    85.       
    86.     /** 
    87.      *  
    88.      * @描述：创建一个zookeeper连接---连接方式二:优选这个 
    89.      * @return void 
    90.      * @exception 
    91.      * @createTime：2016年5月17日 
    92.      * @author: songqinghu 
    93.      */  
    94.     private static CuratorFramework clientTwo(){  
    95.   
    96.         //默认创建的根节点是没有做权限控制的--需要自己手动加权限???----  
    97.         ACLProvider aclProvider = new ACLProvider() {  
    98.             private List<ACL> acl ;  
    99.             @Override  
    100.             public List<ACL> getDefaultAcl() {  
    101.                 if(acl ==null){  
    102.                     ArrayList<ACL> acl = ZooDefs.Ids.CREATOR_ALL_ACL;  
    103.                     acl.clear();  
    104.                     acl.add(new ACL(Perms.ALL, new Id("auth", "admin:admin") ));  
    105.                     this.acl = acl;  
    106.                 }  
    107.                 return acl;  
    108.             }  
    109.             @Override  
    110.             public List<ACL> getAclForPath(String path) {  
    111.                 return acl;  
    112.             }  
    113.         };  
    114.         String scheme = "digest";  
    115.         byte[] auth = "admin:admin".getBytes();  
    116.         int connectionTimeoutMs = 5000;  
    117.         String connectString = "10.125.2.44:2181";  
    118.         String namespace = "testnamespace";  
    119.         CuratorFramework client = CuratorFrameworkFactory.builder().aclProvider(aclProvider).  
    120.         authorization(scheme, auth).  
    121.         connectionTimeoutMs(connectionTimeoutMs).  
    122.         connectString(connectString).  
    123.         namespace(namespace).  
    124.         retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000)).build();  
    125.         client.start();  
    126.         return client;  
    127.     }  
    128.       
    129.       
    130.     /** 
    131.      *  
    132.      * @描述：第一种监听器的添加方式: 对指定的节点进行添加操作 
    133.      * 仅仅能监控指定的本节点的数据修改,删除 操作 并且只能监听一次 --->不好 
    134.      * @return void 
    135.      * @exception 
    136.      * @createTime：2016年5月18日 
    137.      * @author: songqinghu 
    138.      * @throws Exception  
    139.      */  
    140.     private static void setListenterOne(CuratorFramework client) throws Exception{  
    141.      // 注册观察者，当节点变动时触发    
    142.         byte[] data = client.getData().usingWatcher(new Watcher() {  
    143.             @Override  
    144.             public void process(WatchedEvent event) {  
    145.                 System.out.println("获取 two 节点 监听器 : " + event);  
    146.             }  
    147.         }).forPath("/two");  
    148.         System.out.println("two 节点数据: "+ new String(data));  
    149.     }  
    150.     /** 
    151.      *  
    152.      * @描述：第二种监听器的添加方式:  
    153.      * 也是一次性的监听操作,使用后就无法在继续监听了 
    154.      * @return void 
    155.      * @exception 
    156.      * @createTime：2016年5月18日 
    157.      * @author: songqinghu 
    158.      * @throws Exception  
    159.      */  
    160.     private static void setListenterTwo(CuratorFramework client) throws Exception{  
    161.   
    162.         ExecutorService pool = Executors.newCachedThreadPool();  
    163.           
    164.          CuratorListener listener = new CuratorListener() {  
    165.             @Override  
    166.             public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {  
    167.                 System.out.println("监听器  : "+ event.toString());  
    168.             }  
    169.         };  
    170.         client.getCuratorListenable().addListener(listener,pool);  
    171.         client.getData().inBackground().forPath("/two");  
    172.         client.getData().inBackground().forPath("/two");  
    173.         client.getData().inBackground().forPath("/two");  
    174.         client.getData().inBackground().forPath("/two");  
    175.         Thread.sleep(Long.MAX_VALUE );  
    176.     }  
    177.     /** 
    178.      *  
    179.      * @描述：第三种监听器的添加方式: Cache 的三种实现 实践 
    180.      *   Path Cache：监视一个路径下1）孩子结点的创建、2）删除，3）以及结点数据的更新。 
    181.      *                  产生的事件会传递给注册的PathChildrenCacheListener。 
    182.      *  Node Cache：监视一个结点的创建、更新、删除，并将结点的数据缓存在本地。 
    183.      *  Tree Cache：Path Cache和Node Cache的“合体”，监视路径下的创建、更新、删除事件，并缓存路径下所有孩子结点的数据。 
    184.      * @return void 
    185.      * @exception 
    186.      * @createTime：2016年5月18日 
    187.      * @author: songqinghu 
    188.      * @throws Exception  
    189.      */  
    190.     //1.path Cache  连接  路径  是否获取数据  
    191.     //能监听所有的字节点 且是无限监听的模式 但是 指定目录下节点的子节点不再监听  
    192.     private static void setListenterThreeOne(CuratorFramework client) throws Exception{  
    193.         ExecutorService pool = Executors.newCachedThreadPool();  
    194.         PathChildrenCache childrenCache = new PathChildrenCache(client, "/test", true);  
    195.         PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {  
    196.             @Override  
    197.             public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {  
    198.                 System.out.println("开始进行事件分析:-----");  
    199.                 ChildData data = event.getData();  
    200.                 switch (event.getType()) {  
    201.                 case CHILD_ADDED:  
    202.                     System.out.println("CHILD_ADDED : "+ data.getPath() +"  数据:"+ data.getData());  
    203.                     break;  
    204.                 case CHILD_REMOVED:  
    205.                     System.out.println("CHILD_REMOVED : "+ data.getPath() +"  数据:"+ data.getData());  
    206.                     break;  
    207.                 case CHILD_UPDATED:  
    208.                     System.out.println("CHILD_UPDATED : "+ data.getPath() +"  数据:"+ data.getData());  
    209.                     break;  
    210.                 default:  
    211.                     break;  
    212.                 }  
    213.             }  
    214.         };  
    215.         childrenCache.getListenable().addListener(childrenCacheListener);  
    216.         System.out.println("Register zk watcher successfully!");  
    217.         childrenCache.start(StartMode.POST_INITIALIZED_EVENT);  
    218.     }  
    219.       
    220.     //2.Node Cache  监控本节点的变化情况   连接 目录 是否压缩  
    221.     //监听本节点的变化  节点可以进行修改操作  删除节点后会再次创建(空节点)  
    222.     private static void setListenterThreeTwo(CuratorFramework client) throws Exception{  
    223.         ExecutorService pool = Executors.newCachedThreadPool();  
    224.         //设置节点的cache  
    225.        final  NodeCache nodeCache = new NodeCache(client, "/test", false);  
    226.         nodeCache.getListenable().addListener(new NodeCacheListener() {  
    227.             @Override  
    228.             public void nodeChanged() throws Exception {  
    229.                 System.out.println("the test node is change and result is :");  
    230.                 System.out.println("path : "+nodeCache.getCurrentData().getPath());  
    231.                 System.out.println("data : "+new String(nodeCache.getCurrentData().getData()));  
    232.                 System.out.println("stat : "+nodeCache.getCurrentData().getStat());  
    233.             }  
    234.         });  
    235.         nodeCache.start();  
    236.     }  
    237.     //3.Tree Cache    
    238.     // 监控 指定节点和节点下的所有的节点的变化--无限监听  可以进行本节点的删除(不在创建)  
    239.     private static void setListenterThreeThree(CuratorFramework client) throws Exception{  
    240.         ExecutorService pool = Executors.newCachedThreadPool();  
    241.         //设置节点的cache  
    242.         TreeCache treeCache = new TreeCache(client, "/test");  
    243.         //设置监听器和处理过程  
    244.         treeCache.getListenable().addListener(new TreeCacheListener() {  
    245.             @Override  
    246.             public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {  
    247.                 ChildData data = event.getData();  
    248.                 if(data !=null){  
    249.                     switch (event.getType()) {  
    250.                     case NODE_ADDED:  
    251.                         System.out.println("NODE_ADDED : "+ data.getPath() +"  数据:"+ new String(data.getData()));  
    252.                         break;  
    253.                     case NODE_REMOVED:  
    254.                         System.out.println("NODE_REMOVED : "+ data.getPath() +"  数据:"+ new String(data.getData()));  
    255.                         break;  
    256.                     case NODE_UPDATED:  
    257.                         System.out.println("NODE_UPDATED : "+ data.getPath() +"  数据:"+ new String(data.getData()));  
    258.                         break;  
    259.                           
    260.                     default:  
    261.                         break;  
    262.                     }  
    263.                 }else{  
    264.                     System.out.println( "data is null : "+ event.getType());  
    265.                 }  
    266.             }  
    267.         });  
    268.         //开始监听  
    269.         treeCache.start();  
    270.           
    271.     }  
    272.    
    273. }  

猜你喜欢

转载自www.cnblogs.com/exmyth/p/9013835.html
今日推荐
周排行
成为C++高手之宏与枚举
在CAD二次开发中使用进度条
Js插件ECharts，HighCharts学习网址整理
Celery提交任务出错(on windows.)
cephfs内核客户端性能追踪
thinkphp中PHPExcel用法
EntityFramework动态组合多排序字段
汇编语言（八）实验9 根据材料编程
安装ubuntu后必须做的事情（对我而言）
JS函数式编程
每日归档
更多
2024-10-22(0)
2024-10-21(0)
2024-10-20(0)
2024-10-19(0)
2024-10-18(0)
2024-10-17(0)
2024-10-16(0)
2024-10-15(0)
2024-10-14(0)
2024-10-13(0)

代码天地 © 2018 codetd.com

境外服务器   最新电视剧2022