八：Zookeeper开源客户端Curator的api测试

 1 <dependency>
 2       <groupId>org.apache.curator</groupId>
 3       <artifactId>curator-framework</artifactId>
 4       <version>2.8.0</version>
 5     </dependency>
 6     <dependency>
 7       <groupId>org.apache.curator</groupId>
 8       <artifactId>curator-recipes</artifactId>
 9       <version>2.8.0</version>
10     </dependency>

  1 package com.yeepay.sxf.testrurator;
  2 
  3 import java.util.List;
  4 import java.util.concurrent.ExecutorService;
  5 import java.util.concurrent.Executors;
  6 
  7 import org.apache.curator.RetryPolicy;
  8 import org.apache.curator.framework.CuratorFramework;
  9 import org.apache.curator.framework.CuratorFrameworkFactory;
 10 import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
 11 import org.apache.curator.framework.api.BackgroundCallback;
 12 import org.apache.curator.framework.api.CreateBuilder;
 13 import org.apache.curator.framework.api.CuratorEvent;
 14 import org.apache.curator.framework.api.CuratorEventType;
 15 import org.apache.curator.framework.api.DeleteBuilder;
 16 import org.apache.curator.framework.api.ExistsBuilder;
 17 import org.apache.curator.framework.api.GetChildrenBuilder;
 18 import org.apache.curator.framework.api.GetDataBuilder;
 19 import org.apache.curator.framework.api.SetDataBuilder;
 20 import org.apache.curator.framework.recipes.cache.NodeCache;
 21 import org.apache.curator.framework.recipes.cache.NodeCacheListener;
 22 import org.apache.curator.framework.recipes.cache.PathChildrenCache;
 23 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
 24 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type;
 25 import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
 26 import org.apache.curator.retry.ExponentialBackoffRetry;
 27 import org.apache.curator.retry.RetryNTimes;
 28 import org.apache.curator.retry.RetryUntilElapsed;
 29 import org.apache.zookeeper.CreateMode;
 30 import org.apache.zookeeper.data.Stat;
 31 
 32 /**
 33  * 测试Rurator客户端测试
 34  * @author sxf
 35  *
 36  */
 37 public class TestRurator {
 38     
 39     //创建一个线程池，让异步调用的的回调方法，申请线程池中的资源
 40     private static ExecutorService es=Executors.newFixedThreadPool(5);
 41     public static void main(String[] args) throws Exception {
 42                 //重试策略
 43                 RetryPolicy retryPolicy =new ExponentialBackoffRetry(1000, 3);
 44                 
 45                 //客户端构建器
 46                 Builder builder=CuratorFrameworkFactory.builder();
 47                 
 48                 builder.connectString("10.151.30.75:2181");//客户端链接地址：端口号
 49                 builder.sessionTimeoutMs(5000);//会话的超时时间
 50                 builder.connectionTimeoutMs(5000);//链接的超时时间
 51                 builder.retryPolicy(retryPolicy);//指定重试策略
 52                 
 53                 //构建客户端
 54                 CuratorFramework client=builder.build();
 55                 client.start();
 56                 
 57                 //测试创建节点
 58                 //testCreateNode(client);
 59                 
 60                 //测试删除节点
 61                 //testDeleteNode(client);
 62                 
 63                 //获取子节点列表
 64                 //getChildList(client);
 65                 
 66                 //获取节点中存取的值
 67                 //getNodeValue(client);
 68                 
 69                 //修改节点中存储数据的值
 70                 //updateNodeValue(client);
 71                 
 72                 //判断一个节点是否存在
 73                 //isExistsNode(client);
 74                 
 75                 //测试异步调用使用方法
 76                 //testSynCallBack(client);
 77                 
 78                 //测试节点数据变动和节点被创建的事件足册
 79                 //nodeValueChangeListern(client);
 80                 
 81                 //给某个节点注册监听其子节点的活动
 82                 nodeChildListernDothing(client);
 83                 Thread.sleep(Integer.MAX_VALUE);
 84     }
 85     
 86     
 87     /**
 88      * 建立客户端链接的第一种方法
 89      * @throws InterruptedException
 90      */
 91     public void testCreateSession() throws InterruptedException{
 92         //重试策略可以实现该接口。通过自定义规则，进行重试
 93         //重试链接的策略的接口
 94         //(最开始时间间隔，最多重试次数) 随着次数递增，时间间隔还会增加
 95         RetryPolicy retryPolicy =new ExponentialBackoffRetry(1000, 3);
 96         //(最多重试次数，每次重试间隔的时间长度)
 97         RetryPolicy retryPolicy2 =new RetryNTimes(5, 1000);
 98         //(重试时间的总长度，每次重试的时间间隔) 这个基本可重试10次
 99         RetryPolicy retryPolicy3 =new RetryUntilElapsed(10000, 1000);
100         
101         
102         
103         //(ip地址端口,会话超时时间，建立连接超时时间,重试链接的策略)
104         CuratorFramework client=CuratorFrameworkFactory.newClient("10.151.30.75:2181", 5000,5000,retryPolicy);
105         //启动链接
106         client.start();
107         
108         Thread.sleep(Integer.MAX_VALUE);
109         
110     }
111     
112     /**
113      * 建立客户端链接的第二种方法
114      */
115     public static void testCreateSession2(){
116         //重试策略
117         RetryPolicy retryPolicy =new ExponentialBackoffRetry(1000, 3);
118         
119         //客户端构建器
120         Builder builder=CuratorFrameworkFactory.builder();
121         
122         builder.connectString("10.151.30.75:2181");//客户端链接地址：端口号
123         builder.sessionTimeoutMs(5000);//会话的超时时间
124         builder.connectionTimeoutMs(5000);//链接的超时时间
125         builder.retryPolicy(retryPolicy);//指定重试策略
126         builder.authorization(null);//给客户端添加权限AuthInfo 相关的知识在原生api里有讲过
127     
128 //        AuthInfo a=new AuthInfo("ip", "10.151.30.75:2181".getBytes());
129 //        AuthInfo b=new AuthInfo("digest", "shangxiaofei:shangxiaofei".getBytes());
130         
131         //构建客户端
132         CuratorFramework client=builder.build();
133         
134         //建立链接
135         client.start();
136         
137     }
138     
139     /**
140      * 测试创建节点(调用顺序)
141      * 两种创建方式都可以
142      * @throws Exception 
143      */
144     public static void testCreateNode(CuratorFramework client) throws Exception{
145         //创建节点构建器
146 //        String path=client.create()
147 //                                   .creatingParentsIfNeeded()//防止创建节点的父亲节点不存在，导致创建节点失败。如果父节点不存在，先创建父节点，再创建子节点
148 //                                   .withMode(CreateMode.PERSISTENT)//指定节点的类型(持久节点，临时节点，持久节并顺序节点，临时并顺序节点)
149 //                                   .forPath("/node_134", "xiaoshuai".getBytes());//节点路径，节点中存储的数据
150 //    
151         CreateBuilder builder=client.create();
152         builder.creatingParentsIfNeeded();
153         builder.withMode(CreateMode.PERSISTENT);
154         builder.withACL(null);//添加这个节点的权限。相关内容，在原生zookeeper客户端有讲解。上上篇博客
155         /**
156          * //基于ip的权限，意味着这个ip的客户端对此节点有读取权限
157          *     ACL ipacl=new ACL(Perms.READ, new Id("ip", "10.151.30.75"));
158          *     //基于digest的权限，意味着只有这个用户名和密码的客户端才能读取和写的权限
159          *     ACL digetacl=new ACL(Perms.READ|Perms.WRITE,new Id("digest",DigestAuthenticationProvider.generateDigest("shangxiaofei:shangxiaofei")));
160          * 
161          *     List<ACL> myaclAcls=new ArrayList<ACL>();
162          *     myaclAcls.add(ipacl);
163          *     myaclAcls.add(digetacl);
164          */
165         String pathString=builder.forPath("/node_135/node_135_2", "zhengjiuzhonghua".getBytes());
166     
167     
168         
169         System.out.println("TestRurator.testCreateNode()"+pathString);
170     }
171     
172     /**
173      * 测试删除节点(调用顺序不能乱)
174      * @param client
175      * @throws Exception 
176      */
177     public static void testDeleteNode(CuratorFramework client) throws Exception{
178         DeleteBuilder delteBuilder=client.delete();
179         delteBuilder.guaranteed();//保证节点必须删除，如果删除出现错误，则后台程序会不断去尝试删除。
180         delteBuilder.deletingChildrenIfNeeded();//如果存在子节点，先删除子节点
181         delteBuilder.withVersion(-1);//指定版本号
182         delteBuilder.forPath("/node_135");//指定路径
183     }
184     
185     /**
186      * 获取子节点列表
187      * @param client
188      * @throws Exception 
189      */
190     public static void getChildList(CuratorFramework client) throws Exception{
191         GetChildrenBuilder getChildBuilder=client.getChildren();
192         List<String> nodelist=getChildBuilder.forPath("/");
193         System.out.println("TestRurator.getChildList()"+nodelist);
194         
195     }
196     /**
197      * 获取节点中存取的值
198      * @throws Exception 
199      */
200     public static void getNodeValue(CuratorFramework client) throws Exception{
201         
202         GetDataBuilder dataBuilder=    client.getData();
203         
204         Stat stat=new Stat();
205         dataBuilder.storingStatIn(stat);
206         byte[] bytes=dataBuilder.forPath("/node_135/node_135_2");
207         System.out.println("TestRurator.getNodeValue(data==>)"+new String(bytes));
208         System.out.println("TestRurator.getNodeValue()"+stat);
209     }
210     
211     /**
212      * 修改节点中的数据
213      * @param client
214      * @throws Exception 
215      */
216     public static void updateNodeValue(CuratorFramework client) throws Exception{
217         SetDataBuilder setDataBuilder=client.setData();
218         Stat stat=setDataBuilder.withVersion(-1).forPath("/node_135/node_135_2", "tianxiawushuang".getBytes());
219         System.out.println("TestRurator.updateNodeValue()"+stat);
220     }
221     
222     /**
223      * 判断一个节点是否存在
224      * @param client
225      * @throws Exception 
226      */
227     public static void isExistsNode(CuratorFramework client) throws Exception{
228         ExistsBuilder existsBuilder=client.checkExists();
229         Stat stat=existsBuilder.forPath("/node_135/node_135_2");
230         if(stat!=null){
231             System.out.println("TestRurator.isExistsNode()节点存在");
232         }else{
233             System.out.println("TestRurator.isExistsNode()节点不存在");
234         }
235     }
236     /**
237      * 以判断节点是否存在，讲解异步调用的使用
238      * @param client
239      * @throws Exception 
240      */
241     public static void testSynCallBack(CuratorFramework client) throws Exception{
242         ExistsBuilder existsBuilder=client.checkExists();
243         existsBuilder.inBackground(new myBalckCallBack(),"sxf",es).forPath("/node_135/node_135_2");
244         //System.out.println("TestRurator.testSynCallBack(异步调用返回)"+stat);
245     }
246     
247     /**
248      * 异步调用回调接口的实现类
249      * @author sxf
250      *
251      */
252     private static class myBalckCallBack implements BackgroundCallback{
253 
254         @Override
255         public void processResult(CuratorFramework client, CuratorEvent event)
256                 throws Exception {
257                 //回调类型
258                 CuratorEventType curatorEventType=event.getType();
259                 //节点路径
260                 String path=event.getPath();
261                 //节点列表
262                 List<String> list=event.getChildren();
263                 //获取上下文
264                 Object context=event.getContext();
265                 //获取返回碼  异步操作成功，返回0
266                 int code=event.getResultCode();
267                 //获取数据内容
268                 byte[] data=event.getData();
269                 
270                 StringBuffer sb=new StringBuffer();
271                 sb.append("curatorEventType="+curatorEventType).append("\n");
272                 sb.append("path="+path).append("\n");
273                 sb.append("list="+list).append("\n");
274                 sb.append("context="+context).append("\n");
275                 sb.append("data="+new String(data)).append("\n");
276                 
277                 System.out.println(sb.toString());
278         }
279         
280     }
281     
282     
283     /**
284      *节点事件监听(节点被创建，节点中存储的数据被修改)
285      * @param client
286      * @throws Exception 
287      */
288     public  static void nodeValueChangeListern(CuratorFramework client) throws Exception{
289         //注册监听
290         final NodeCache nodeCache =new NodeCache(client, "/node_135/node_135_2");;
291         //开启监听
292         nodeCache.start();
293         //添加监听器(实现指定接口的实例对象)
294         nodeCache.getListenable().addListener(new NodeCacheListener() {
295             
296             @Override
297             public void nodeChanged() throws Exception {
298                 //获取当前节点的最近数据
299                 byte[] bytes=nodeCache.getCurrentData().getData();
300                 System.out.println(new String(bytes));
301                 
302             }
303         });
304     }
305     
306     /**
307      * 监听节点的子节点事件注册
308      * @param client
309      * @throws Exception 
310      */
311     public static void nodeChildListernDothing(CuratorFramework client) throws Exception{
312         //注册监听器
313         //第三个参数：系统在监听到子节点列表发生变化时同时获取子节点的数据内容
314         final PathChildrenCache cache=new PathChildrenCache(client, "/node_135", true);
315         //开启监听
316         cache.start();
317         //注册监听器
318         cache.getListenable().addListener(new PathChildrenCacheListener() {
319             
320             @Override
321             public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
322                     throws Exception {
323                 Type type=event.getType();
324                 switch (event.getType()) {
325                 case CHILD_ADDED://添加子节点
326                         System.out.println("新添加子节点中存储的值==>"+new String(event.getData().getData()));
327                     break;
328 
329                 case CHILD_UPDATED://子节点被修改
330                         System.out.println("被修改的子节点中存储的值==>"+new String(event.getData().getData()));
331                     break;
332                 case CHILD_REMOVED://子节点被删除    
333                         System.out.println("被删除的字节点的路径==>"+event.getInitialData());
334                     break;
335                 
336                 default:
337                     break;
338                 }
339                 
340             }
341         });
342     }
343     
344 }