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

àì夳堔傛蜴生んèń 2021-09-11 06:26 417阅读 0赞

**\[java\]**  [view plain][]   [copy][view plain]

[![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

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. \}  
274.

[view plain]: http://blog.csdn.net/sqh201030412/article/details/51446434#
[CODE]: https://code.csdn.net/assets/CODE_ico.png
[CODE_CODE]: https://code.csdn.net/snippets/1688239
[ico_fork.svg]: https://code.csdn.net/assets/ico_fork.svg
[ico_fork.svg 1]: https://code.csdn.net/snippets/1688239/fork