六：ZooKeeper的java客户端api的使用

一：客户端链接测试

 package com.yeepay.sxf.createConnection;

 import java.io.IOException;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.ZooKeeper.States;

 /**

  * 测试Zookeeper的链接

  * @author sxf

  *

  */

 public class TestCreateSession {

     //zooKeeper实例

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

         //实例化zooKeeper的实例

         //参数：(ip地址：端口号  ，当前会话超时时间，自定义事件监听器）

         zooKeeper=new ZooKeeper("10.151.30.75:2181",5000, new MyWatcher());

         //获取链接状态

         States states=zooKeeper.getState();

         //此链接为异步链接

         System.out.println("TestCreateSession.main(链接状态)："+states.toString());//CONNECTING

         Thread.sleep(Integer.MAX_VALUE);

     }

 }

 package com.yeepay.sxf.createConnection;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 /**

  * zookeeper实例过程中的事件监听器

  * @author sxf

  *

  */

 public class MyWatcher implements Watcher{

     //该方法可以做相关的逻辑代码

     @Override

     public void process(WatchedEvent event) {

         //MyWatcher.process(接收到的事件：)WatchedEvent state:SyncConnected type:None path:null

         System.out.println("MyWatcher.process(接收到的事件：)"+event);

         //如果链接成功可以做一些事情

         if(event.getState()==KeeperState.SyncConnected){

             System.out.println("MyWatcher.process(链接成功做一些事情：)");

         }

     }

 }

二：客户端创建节点测试

 package com.yeepay.sxf.createNode;

 import java.io.IOException;

 import java.security.NoSuchAlgorithmException;

 import java.util.ArrayList;

 import java.util.List;

 import org.apache.zookeeper.CreateMode;

 import org.apache.zookeeper.KeeperException;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooDefs.Ids;

 import org.apache.zookeeper.ZooDefs.Perms;

 import org.apache.zookeeper.data.ACL;

 import org.apache.zookeeper.data.Id;

 import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;

 /**

  * zooKeeper同步创建节点测试

  * @author sxf

  *

  */

 public class TestCreateNodeSyn implements Watcher {

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException, NoSuchAlgorithmException {

 //        //实例化zooKeeper链接

 //        zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestCreateNodeSyn());

 //        //创建链接

 //        Thread.sleep(Integer.MAX_VALUE);

         System.out.println(DigestAuthenticationProvider.generateDigest("shangxiaofei:shangxiaofei"));

     }

     /**

      *权限模式（scheme）： ip,digest

      *授权对象（ID）：

      *            ip权限模式：具体的ip地址

      *            digest权限模式：username:Base64(SHA-1(username:password))

      *

      *权限（permission）:CREATE(C),DELETE(D),READ(R),WRITE(W),ADMIN(A)

      *        注：单个权限，完全权限，复合权限

      *

      *权限组合：scheme+ID+permission

      */

     public void process(WatchedEvent event) {

         //链接成功

                 if(event.getState()==KeeperState.SyncConnected){

                     //同步创建节点

                     try {

                         //基于ip的权限，意味着这个ip的客户端对此节点有读取权限

                         ACL ipacl=new ACL(Perms.READ, new Id("ip", "10.151.30.75"));

                         //基于digest的权限，意味着只有这个用户名和密码的客户端才能读取和写的权限

                         ACL digetacl=new ACL(Perms.READ|Perms.WRITE,new Id("digest",DigestAuthenticationProvider.generateDigest("shangxiaofei:shangxiaofei")));

                         List<ACL> myaclAcls=new ArrayList<ACL>();

                         myaclAcls.add(ipacl);

                         myaclAcls.add(digetacl);

                         String path=zooKeeper.create("/node_128", "shangxiaofei".getBytes(), myaclAcls, CreateMode.PERSISTENT);

                         System.out.println("MyWatcher2.process(创建节点返回的路径：)"+path);

                     }catch (NoSuchAlgorithmException e){

                         e.printStackTrace();

                     } catch (KeeperException e) {

                         // TODO Auto-generated catch block

                         e.printStackTrace();

                     } catch (InterruptedException e) {

                         // TODO Auto-generated catch block

                         e.printStackTrace();

                     }

                 }

     }

 }

 package com.yeepay.sxf.createNode;

 import java.io.IOException;

 import org.apache.zookeeper.AsyncCallback;

 import org.apache.zookeeper.CreateMode;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooDefs.Ids;

 import org.apache.zookeeper.ZooKeeper;

 /**

  * zooKeeper异步创建节点测试

  * @author sxf

  *

  */

 public class TestCreateNodeAsyn implements Watcher {

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

         //实例化zooKeeper链接

         zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestCreateNodeAsyn());

         //创建链接

         Thread.sleep(Integer.MAX_VALUE);

     }

     @Override

     public void process(WatchedEvent event) {

         //链接成功

                 if(event.getState()==KeeperState.SyncConnected){

                         //异步创建节点

                         zooKeeper.create("/node_124", "shangxiaoshuai".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT,new ISstringCallBack(),"sxf创建");

                 }

     }

     //该类数据异步回调接口实现

     static class ISstringCallBack implements AsyncCallback.StringCallback{

         //创建成功rc=0

         //path，创建节点的路径

         //ctx,创建节点的传入的上下位

         //name,创建节点的名字

         @Override

         public void processResult(int rc, String path, Object ctx, String name) {

             StringBuffer sb=new StringBuffer();

             sb.append("rc="+rc).append("\n");

             sb.append("path="+path).append("\n");

             sb.append("ctx="+ctx).append("\n");

             sb.append("name="+name);

             System.out.println(sb.toString());

             /**

              *rc=0

              *path=/node_124

              *ctx=sxf创建

              *name=/node_124

              *

              */

         }

     }

 }

三：客户端删除节点测试

 package com.yeepay.sxf.deleteNode;

 import java.io.IOException;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 /**

  *同步删除节点中存取的值

  * @author sxf

  *

  */

 public class TestDeleteNodeSyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestDeleteNodeSyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //删除一个节点

                     //第一个参数：删除节点的全路径

                     //第二个参素：节点的版本（类似乐观锁），当为-1时，对版本无限制

                     zooKeeper.delete("/node_127", -1);

                 }else if(event.getType()==EventType.NodeDataChanged){

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

 }

 package com.yeepay.sxf.deleteNode;

 import java.io.IOException;

 import org.apache.zookeeper.AsyncCallback;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 /**

  *异步删除节点

  * @author sxf

  *

  */

 public class TestDeleteNodeAsyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestDeleteNodeAsyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //第一个参数：要删除的节点的全路径

                     //第二个参数：要删除节点的版本号（类似乐观锁）

                     //第三个参数:异步删除的回调实现类

                     //第四个参数：删除传入的上下文

                     zooKeeper.delete("/node_126", -1, new IsDeleteCallBack(), "删除节点sxf");

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

     /**

      * 异步删除回调接口实现类

      * @author sxf

      *

      */

     static class IsDeleteCallBack implements AsyncCallback.VoidCallback{

         //第一个参数：返回删除成功rc

         @Override

         public void processResult(int rc, String path, Object ctx) {

             StringBuffer sb=new StringBuffer();

             sb.append("rc="+rc).append("\n");

             sb.append("path="+path).append("\n");

             sb.append("ctx="+ctx).append("\n");

             System.out.println(sb.toString());

             /**

              * rc=0

              *    path=/node_126

              *    ctx=删除节点sxf

              */

         }

     }

 }

四：客户端判断节点是否存在测试

 package com.yeepay.sxf.existsNode;

 import java.io.IOException;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.data.Stat;

 /**

  *同步判断节点是否存在

  * @author sxf

  *

  */

 public class TestExistsNodeSyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestExistsNodeSyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //第一个参数：判断节点的全路径

                     //第二个参数：是否注册监听器

                     Stat stat=zooKeeper.exists("/node_123",true);

                     System.out.println(stat);

                 }else if(event.getType()==EventType.NodeDataChanged){

                     //节点数据改变

                     Stat stat=zooKeeper.exists(event.getPath(),true);

                     System.out.println("节点数据改变=>"+stat);

                 }else if(event.getType()==EventType.NodeCreated){

                     //节点被创建

                     Stat stat=zooKeeper.exists(event.getPath(),true);

                     System.out.println("节点被创建=>"+stat);

                 }else if(event.getType()==EventType.NodeDeleted){

                     //节点被删除

                     Stat stat=zooKeeper.exists(event.getPath(),true);

                     System.out.println("节点被删除=>"+stat);

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

 }

 package com.yeepay.sxf.existsNode;

 import java.io.IOException;

 import org.apache.zookeeper.AsyncCallback;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.data.Stat;

 /**

  *异步判断节点是否存在

  * @author sxf

  *

  */

 public class TestExistsNodeAsyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestExistsNodeAsyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                         zooKeeper.exists("/node_123", true, new IsStatCallBack(), "sxf判断节点是否存在");

                 }else if(event.getType()==EventType.NodeDataChanged){

                     //节点数据改变

                     zooKeeper.exists("/node_123", true, new IsStatCallBack(), "sxf判断节点是否存在");

                     System.out.println("TestExistsNodeAsyn.process(节点数据改变)");

                 }else if(event.getType()==EventType.NodeCreated){

                     //节点被创建

                     zooKeeper.exists("/node_123", true, new IsStatCallBack(), "sxf判断节点是否存在");

                     System.out.println("TestExistsNodeAsyn.process(节点被创建)");

                 }else if(event.getType()==EventType.NodeDeleted){

                     //节点被删除

                     zooKeeper.exists("/node_123", true, new IsStatCallBack(), "sxf判断节点是否存在");

                     System.out.println("TestExistsNodeAsyn.process(节点被删除)");

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

     //节点是否存在的异步回调接口实现类

     static class IsStatCallBack implements AsyncCallback.StatCallback{

         @Override

         public void processResult(int rc, String path, Object ctx, Stat stat) {

             StringBuffer sb=new StringBuffer();

             sb.append("rc="+rc).append("\n");

             sb.append("path="+path).append("\n");

             sb.append("ctx="+ctx).append("\n");

             sb.append("stat="+stat).append("\n");

             System.out.println(sb.toString());

         }

     }

 }

五：客户端获取节点列表测试

 package com.yeepay.sxf.getNode;

 import java.io.IOException;

 import java.util.List;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 /**

  * 同步获取节点列表node

  * @author sxf

  *

  */

 public class TestGetNodeSyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestGetNodeSyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //获取制定节点的子节点列表

                     //第一参数：节点路径

                     //第二个参数：是否需要关注该节点的子节点列表变化。关注：true  不关注：false  (事件监听器)

                     List<String> nodesList=zooKeeper.getChildren("/", true);

                     System.out.println("/ 下的子节点列表==>"+nodesList);

                     /**

                      * / 下的子节点列表==>[node_123, node_124, zookeeper]

                      */

                 }else if(event.getType()==EventType.NodeChildrenChanged){

                     //如果节点发生变化，则会触发该事件.当别的地方添加一个节点，则该方法被调用

                     System.out.println("TestGetNodeSyn.process()变化的地址："+event.getPath());//TestGetNodeSyn.process()变化的地址：/

                     List<String> nodesList=zooKeeper.getChildren(event.getPath(), true);

                     System.out.println("TestGetNodeSyn.process(节点发生变化==>)"+nodesList);

                     /**

                      * TestGetNodeSyn.process(节点发生变化==>)[node_123, node_124, zookeeper, node_125]

                      */

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

 }

 package com.yeepay.sxf.getNode;

 import java.io.IOException;

 import java.util.List;

 import org.apache.zookeeper.AsyncCallback;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.data.Stat;

 /**

  * 异步获取节点列表

  * @author sxf

  *

  */

 public class TestGetNodeAsyn implements Watcher {

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

         //实例化zooKeeper链接

         zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestGetNodeAsyn());

         //创建链接

         Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //获取制定节点的子节点列表

                     //第一参数：节点路径

                     //第二个参数：是否需要关注该节点的子节点列表变化。关注：true  不关注：false  (事件监听器)

                     //第三个参数：异步回调函数

                     //第四个参数：异步获取传入上下文参数

                     zooKeeper.getChildren("/", true, new ISChildrenCallback(), "SXF获取子节点列表");

                     /**

                      * / 下的子节点列表==>[node_123, node_124, zookeeper]

                      */

                 }else if(event.getType()==EventType.NodeChildrenChanged){

                     zooKeeper.getChildren(event.getPath(), true, new ISChildrenCallback(), "SXF获取子节点列表");

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

     //异步获取子节点列表的回调

     static class ISChildrenCallback implements AsyncCallback.Children2Callback{

         @Override

         public void processResult(int rc, String path, Object ctx,List<String> children, Stat stat) {

             StringBuffer sb=new StringBuffer();

             sb.append("rc="+rc).append("\n");

             sb.append("path="+path).append("\n");

             sb.append("ctx="+ctx).append("\n");

             sb.append("children="+children).append("\n");

             sb.append("stat="+stat).append("\n");

             System.out.println(sb.toString());

             /**

              * rc=0

              *    path=/

              *    ctx=SXF获取子节点列表

              *    children=[node_126, node_127, node_123, node_124, zookeeper, node_125]

              *    stat=0,0,0,0,0,8,0,0,0,6,4294967354

              */

         }

     }

 }

六：客户端获取节点值测试

 package com.yeepay.sxf.getNodeValue;

 import java.io.IOException;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.data.Stat;

 /**

  *同步获取节点中存取的值

  * @author sxf

  *

  */

 public class TestGetNodeValueSyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestGetNodeValueSyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //如果没有权限，可以注册权限

                     zooKeeper.addAuthInfo("digest", "shangxiaofei:shangxiaofei".getBytes());

                     //第一个参数:要获取节点存储值的全路径

                     //第二个参数：是否注册一个事件监听器

                     //第三个参数：一个状态实例

                     byte[] nodeValueByte=zooKeeper.getData("/node_128", true, new Stat());

                     System.out.println("TestGetNodeValueSyn.process(获取/node_128的值：)"+new String(nodeValueByte));

                     //TestGetNodeValueSyn.process(获取/node_127的值：)aaaaaaaa

                 }else if(event.getType()==EventType.NodeDataChanged){

                     //当关注的节点的值发生了变化则通知

                     byte[] nodeValueByte=zooKeeper.getData(event.getPath(), true, new Stat());

                     System.out.println("TestGetNodeValueSyn.process(/node_127的值发生了变化：)"+new String(nodeValueByte));

                     //TestGetNodeValueSyn.process(/node_127的值发生了变化：)shuihongjie

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

 }

 package com.yeepay.sxf.getNodeValue;

 import java.io.IOException;

 import org.apache.zookeeper.AsyncCallback;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.data.Stat;

 /**

  * 异步获取节点中存取的值

  * @author sxf

  *

  */

 public class TestGetNodeValueASyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestGetNodeValueASyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //异步获取节点中的值

                     zooKeeper.getData("/node_127", true, new IsDataCallBack(), "sxf修改");

                 }else if(event.getType()==EventType.NodeDataChanged){

                     zooKeeper.getData(event.getPath(), true, new IsDataCallBack(), "sxf修改");

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

     //异步获取数据的回调函数

     static class IsDataCallBack implements  AsyncCallback.DataCallback{

         @Override

         public void processResult(int rc, String path, Object ctx, byte[] data,

                 Stat stat) {

             StringBuffer sb=new StringBuffer();

             sb.append("rc="+rc).append("\n");

             sb.append("path="+path).append("\n");

             sb.append("ctx="+ctx).append("\n");

             sb.append("data="+new String(data)).append("\n");

             sb.append("stat="+stat).append("\n");

             System.out.println(sb.toString());

             /**

              * rc=0

              *path=/node_127

              *ctx=sxf修改

              *data=tianxiabangchang

              *stat=4294967354,4294967367,1456371531643,1456373368574,4,0,0,0,16,0,4294967354

              */

         }

     }

 }

七：客户端修改节点数据测试

 package com.yeepay.sxf.updateNodeData;

 import java.io.IOException;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.data.Stat;

 /**

  *同步修改节点数据

  * @author sxf

  *

  */

 public class TestUpdateNodeValueSyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestUpdateNodeValueSyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //第一个参数：要修改的节点的全路径

                     //第二个参数：要修改的节点的数据内容

                     //第三个参数：版本号

                     Stat stat=zooKeeper.setData("/node_124", "sxfupdatenodedata".getBytes(), -1);

                     System.out.println("TestUpdateNodeValueSyn.process()"+stat);

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

 }

 package com.yeepay.sxf.updateNodeData;

 import java.io.IOException;

 import org.apache.zookeeper.AsyncCallback;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.Watcher.Event.EventType;

 import org.apache.zookeeper.Watcher.Event.KeeperState;

 import org.apache.zookeeper.ZooKeeper;

 import org.apache.zookeeper.data.Stat;

 /**

  *异步修改节点数据

  * @author sxf

  *

  */

 public class TestUpdateNodeValueAsyn implements Watcher{

     private static ZooKeeper zooKeeper;

     public static void main(String[] args) throws IOException, InterruptedException {

                 //实例化zooKeeper链接

                 zooKeeper=new ZooKeeper("10.151.30.75:2181", 5000,new TestUpdateNodeValueAsyn());

                 //创建链接

                 Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * event

      * （1）zookeeper链接状态 event.getState()

      * （2）zookeeper事件类型 event.getType()

      * （3）zookeeper事件触发的节点路径 event.getPath()

      */

     @Override

     public void process(WatchedEvent event) {

         try{

             //已经连接做一件事情

             if(event.getState()==KeeperState.SyncConnected){

                 if(event.getType()==EventType.None&&event.getPath()==null){

                     //第一个参数：要修改的节点的全路径

                     //第二个参数：要修改的节点的数据内容

                     //第三个参数：版本号

                     zooKeeper.setData("/node_124", "yangkai".getBytes(), -1, new ISupdateDataCallBack(), "sxf修改节点数据");

                 }

             }

         }catch(Exception e){

             e.printStackTrace();

         }

     }

     /**

      * 异步修改节点数据的回调接口的实现

      * @author sxf

      *

      */

     static class ISupdateDataCallBack implements AsyncCallback.StatCallback{

         @Override

         public void processResult(int rc, String path, Object ctx, Stat stat) {

             StringBuffer sb=new StringBuffer();

             sb.append("rc="+rc).append("\n");

             sb.append("path="+path).append("\n");

             sb.append("ctx="+ctx).append("\n");

             sb.append("stat="+stat).append("\n");

             System.out.println(sb.toString());

         }

     }

 }

巴特西

六：ZooKeeper的java客户端api的使用

最新文章

热门文章