Java API操作ZooKeeper

创建会话

 package org.zln.zk;

 import org.apache.zookeeper.WatchedEvent;

 import org.apache.zookeeper.Watcher;

 import org.apache.zookeeper.ZooKeeper;

 import java.io.IOException;

 /**

  * Created by sherry on 16/8/27.

  */

 public class TestZooKeeperClientApi {

     private static ZooKeeper zooKeeper;

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

         createSession();

     }

     /**

      * 创建会话

      */

     private static ZooKeeper createSession() throws IOException, InterruptedException {

         //实例化的过程,同时也是与ZooKeeper建立连接的过程,参数说明:ip:port 超时时间  监听器(实现water接口,监听器用于接收通知)

         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Watcher() {

             @Override

             public void process(WatchedEvent watchedEvent) {

                 System.out.println("收到事件:"+watchedEvent);//收到事件:WatchedEvent state:SyncConnected type:None path:null

                 //TODO

             }

         });

         System.out.println("查看状态:"+zooKeeper.getState());//查看状态:CONNECTING

         //如果不停一段时间,那么,监听器还没收到监听,方法就已经退出了

         Thread.sleep(5000);

         return zooKeeper;

     }

 }

创建会话

创建节点

 package org.zln.zk;

 import org.apache.zookeeper.*;

 import org.apache.zookeeper.data.ACL;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

 import java.io.IOException;

 import java.io.UnsupportedEncodingException;

 import java.util.ArrayList;

 /**

  * Created by sherry on 16/8/27.

  */

 public class TestZooKeeperClientApi {

     private static Logger logger = LoggerFactory.getLogger(TestZooKeeperClientApi.class);

     private static ZooKeeper zooKeeper;

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

         createSession();

     }

     /**

      * 创建会话

      */

     private static ZooKeeper createSession() throws IOException, InterruptedException {

         //实例化的过程,同时也是与ZooKeeper建立连接的过程,参数说明:ip:port 超时时间  监听器(实现water接口,监听器用于接收通知)

         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Watcher() {

             @Override

             public void process(WatchedEvent watchedEvent) {

                 //TODO 与 ZooKeeper 的交互,一般都放在这里

                 if (watchedEvent.getState() == Event.KeeperState.SyncConnected){//已连接

                     logger.info("连接上了");

                     try {

                         //参数说明:节点路径  数据的字节数组 权限 创建节点模式

                         String nodePath = createNode(zooKeeper,"/node_1","123".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

                         logger.info("创建节点:"+nodePath);

                     } catch (UnsupportedEncodingException|KeeperException|InterruptedException e) {

                         e.printStackTrace();

                     }

                 }

             }

         });

         logger.info("查看状态:"+zooKeeper.getState());//查看状态:CONNECTING

         //如果不停一段时间,那么,监听器还没收到监听,方法就已经退出了

         Thread.sleep(5000);

         return zooKeeper;

     }

     /**

      * 创建ZooKeeper节点

      * @param zooKeeper ZooKeeper连接

      * @return 节点路径

      */

     public static String createNode(ZooKeeper zooKeeper, String path, byte[] bytes, ArrayList<ACL> acls,CreateMode createMode) throws UnsupportedEncodingException, KeeperException, InterruptedException {

         //参数说明:节点路径  数据的字节数组 权限 创建节点模式

         return zooKeeper.create(path,bytes, acls, createMode);

     }

 }

创建节点

创建模式

PERSISTENT 　　　　　　　　持久节点
PERSISTENT_SEQUENTIAL 　持久顺序节点
EPHEMERAL 　　　　　　　　临时节点
EPHEMERAL_SEQUENTIAL 　临时顺序节点

以上代码，是属于同步创建

 /**

      * 异步创建节点

      * @param zooKeeper

      * @param path

      * @param bytes

      * @param acls

      * @param createMode

      * @throws KeeperException

      * @throws InterruptedException

      */

     public static void asCreateNode(ZooKeeper zooKeeper, String path, byte[] bytes, ArrayList<ACL> acls,CreateMode createMode) throws KeeperException, InterruptedException {

         //异步创建需要增加 AsyncCallback.StringCallback 接口的实现类 以及 一个上下文对象参数

         zooKeeper.create(path, bytes, acls, createMode, new AsyncCallback.StringCallback() {

             /**

              *

              * @param rc  节点创建结果返回码  0-节点创建成功

              * @param path 节点真实路径

              * @param ctx  异步调用上下文  就是 create方法本地调用的那个最后一个参数

              * @param name

              */

             @Override

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

                 StringBuilder stringBuilder = new StringBuilder();

                 stringBuilder.append("\nrc="+rc+"\n" +

                         "path="+path+"\n" +

                         "ctx="+ctx+"\n" +

                         "name="+name+"\n");

                 logger.info(stringBuilder.toString());

             }

         },"异步创建");

     }

异步方式创建节点

获取子节点

     /**

      * 同步方式获取子节点

      * @param zooKeeper     连接

      * @param parentPath    父路径

      * @return

      * @throws KeeperException

      * @throws InterruptedException

      */

     public static List<String> getChildList(ZooKeeper zooKeeper,String parentPath) throws KeeperException, InterruptedException {

         //参数说明:  父节点路径  是否需要关注子节点的变化

         List<String> childs = zooKeeper.getChildren(parentPath,false);

         return childs;

     }

同步方式获取子节点且不关注子节点的变化

异步方式获取子节点且关注子节点的变化

     /**

      * 异步方式获取子节点 关注子节点变化

      * @param zooKeeper     连接

      * @param parentPath    父路径

      */

     public static void asGetChildListAndWatch(ZooKeeper zooKeeper,String parentPath){

         zooKeeper.getChildren(parentPath, true, new AsyncCallback.Children2Callback() {

             @Override

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

                 logger.info("变化后的子节点:");

                 for (String name:children){

                     logger.info("子节点:"+name);

                 }

             }

         },"关注子节点变化");

     }

AsyncCallback.Children2Callback接口实现异步关注

目前为止可以发现这个规律，有回调函数的是异步方式调用，没有回调函数的是同步调用

问：同步调用和异步调用的使用场景是？？？

答：下面的操作依赖调用结果的时候，就需要调用同步方法

 package org.zln.zk;

 import org.apache.zookeeper.*;

 import org.apache.zookeeper.data.ACL;

 import org.apache.zookeeper.data.Stat;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

 import java.io.IOException;

 import java.io.UnsupportedEncodingException;

 import java.util.ArrayList;

 import java.util.List;

 /**

  * Created by sherry on 16/8/27.

  */

 public class TestZooKeeperClientApi {

     private static Logger logger = LoggerFactory.getLogger(TestZooKeeperClientApi.class);

     private static ZooKeeper zooKeeper;

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

         createSession();

         Thread.sleep(Integer.MAX_VALUE);

     }

     /**

      * 创建会话

      */

     private static ZooKeeper createSession() throws IOException, InterruptedException {

         //实例化的过程,同时也是与ZooKeeper建立连接的过程,参数说明:ip:port 超时时间  监听器(实现water接口,监听器用于接收通知)

         zooKeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Watcher() {

             @Override

             public void process(WatchedEvent watchedEvent) {

                 //TODO 与 ZooKeeper 的交互,一般都放在这里

                 if (watchedEvent.getState() == Event.KeeperState.SyncConnected){//已连接

                     logger.info("连接上了");

                     try {

                     //同步方式创建节点

                     //String nodePath = sysCreateNode(zooKeeper,"/node_1","123".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

                     //logger.info("创建节点:"+nodePath);

                     //异步方式创建节点

                     //asCreateNode(zooKeeper,"/node_2","234".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

                     //同步方式获取子节点  不关注子节点变化

 //                  List<String> list = getChildListNoWatch(zooKeeper,"/");

 //                  for (String name:list){

 //                      logger.info("子节点:"+name);

 //                  }

                     //异步方式获取节点 关注子节点变化

 //                    asGetChildListAndWatch(zooKeeper,"/");

                     //同步方式获取节点数据 sysGetNodeData

                         byte[] bytes = sysGetNodeDataNoWatch(zooKeeper,"/node_1");

                         logger.info("获取节点数据"+new String(bytes,"UTF-8"));

                         deleteNode(zooKeeper,"/node_1",0);

                     } catch (KeeperException|InterruptedException|UnsupportedEncodingException e) {

                         e.printStackTrace();

                     }

                 }

             }

         });

         logger.info("查看状态:"+zooKeeper.getState());//查看状态:CONNECTING

         return zooKeeper;

     }

     /**

      * 同步创建节点

      * @param zooKeeper  连接

      * @param path       节点路径

      * @param bytes      字节数组数据

      * @param acls       权限

      * @param createMode 创建模式

      * @return

      * @throws UnsupportedEncodingException

      * @throws KeeperException

      * @throws InterruptedException

      */

     public static String sysCreateNode(ZooKeeper zooKeeper, String path, byte[] bytes, ArrayList<ACL> acls,CreateMode createMode) throws UnsupportedEncodingException, KeeperException, InterruptedException {

         return zooKeeper.create(path,bytes, acls, createMode);

     }

     /**

      * 异步创建节点

      * @param zooKeeper  连接

      * @param path       节点路径

      * @param bytes      字节数组数据

      * @param acls       权限

      * @param createMode 创建模式

      * @throws KeeperException

      * @throws InterruptedException

      */

     public static void asCreateNode(ZooKeeper zooKeeper, String path, byte[] bytes, ArrayList<ACL> acls,CreateMode createMode) throws KeeperException, InterruptedException {

         //异步创建需要增加 AsyncCallback.StringCallback 接口的实现类 以及 一个上下文对象参数

         zooKeeper.create(path, bytes, acls, createMode, new AsyncCallback.StringCallback() {

             /**

              *

              * @param rc  节点创建结果返回码  0-节点创建成功

              * @param path 节点真实路径

              * @param ctx  异步调用上下文  就是 create方法本地调用的那个最后一个参数

              * @param name

              */

             @Override

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

                 StringBuilder stringBuilder = new StringBuilder();

                 stringBuilder.append("\nrc="+rc+"\n" +

                         "path="+path+"\n" +

                         "ctx="+ctx+"\n" +

                         "name="+name+"\n");

                 logger.info(stringBuilder.toString());

             }

         },"异步创建");

     }

     /**

      * 同步方式获取子节点  不关注子节点变化

      * @param zooKeeper     连接

      * @param parentPath    父路径

      * @return

      * @throws KeeperException

      * @throws InterruptedException

      */

     public static List<String> sysGetChildListNoWatch(ZooKeeper zooKeeper,String parentPath) throws KeeperException, InterruptedException {

         //参数说明:  父节点路径  是否需要关注子节点的变化  如果 true,则子节点发生变化后,会产生  NodeChildrenChanged 事件

         List<String> childs = zooKeeper.getChildren(parentPath,false);

         return childs;

     }

     /**

      * 异步方式获取子节点 关注子节点变化

      * @param zooKeeper     连接

      * @param parentPath    父路径

      */

     public static void asGetChildListAndWatch(ZooKeeper zooKeeper,String parentPath){

         zooKeeper.getChildren(parentPath, true, new AsyncCallback.Children2Callback() {

             @Override

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

                 logger.info("变化后的子节点:");

                 for (String name:children){

                     logger.info("子节点:"+name);

                 }

             }

         },"关注子节点变化");

     }

     /**

      * 同步方式获取数据

      * @param zooKeeper

      * @param path

      * @return

      * @throws KeeperException

      * @throws InterruptedException

      */

     public static byte[] sysGetNodeDataNoWatch(ZooKeeper zooKeeper,String path) throws KeeperException, InterruptedException {

         //路径 是否关注数据变化 状态

         return zooKeeper.getData(path,false,new Stat());

     }

     /**

      * 删除节点

      * @param zooKeeper

      * @param nodePath

      * @param version

      * @throws KeeperException

      * @throws InterruptedException

      */

     public static void deleteNode(ZooKeeper zooKeeper,String nodePath,int version) throws KeeperException, InterruptedException {

         zooKeeper.delete(nodePath,version);

     }

 }

客户端代码汇总

除了ZooKeeper提供的Java API外，还有两种客户端，ZKClient和Curator两种客户端，都是对原生API的封装，使得操作更方便

《从PAXOS到ZOOKEEPER分布式一致性原理与实践》，可以参考这本书

巴特西

Java API操作ZooKeeper

最新文章

热门文章