在Java中操作Zookeeper的示例代码详解

 
    org.apache.zookeeper
    zookeeper
    3.6.0
  

//连接字符串,zkServer的ip、port，如果是集群逗号分隔
  String cOnnectStr= "192.168.1.9:2181";

  //zookeeper就是一个zkCli
  ZooKeeper zooKeeper = null;

  try {
　　　　　//初始次数为1。后面要在内部类中使用，三种写法：1、写成外部类成员变量，不用加final；2、作为函数局部变量，放在try外面，写成final；3、写在try中，不加final
  　　 CountDownLatch countDownLatch = new CountDownLatch(1);
    //超时时间ms，监听器
    zooKeeper = new ZooKeeper(connectStr, 5000, new Watcher() {
      public void process(WatchedEvent watchedEvent) {
        //如果状态变成已连接
        if (watchedEvent.getState().equals(Event.KeeperState.SyncConnected)) {
          System.out.println("连接成功");
          //次数-1
          countDownLatch.countDown();
        }
      }
    });
    //等待，次数为0时才会继续往下执行。等待监听器监听到连接成功，才能操作zk
    countDownLatch.await();
  } catch (IOException | InterruptedException e) {
    e.printStackTrace();
  }


  //...操作zk。后面的demo都是写在此处的


  //关闭连接
  try {
    zooKeeper.close();
  } catch (InterruptedException e) {
    e.printStackTrace();
  }

// 检测节点是否存在

  // 同步方式
  Stat exists = null;
  try {
    //如果存在，返回节点状态Stat；如果不存在，返回null。第二个参数是watch
    exists = zooKeeper.exists("/mall",false);
  } catch (KeeperException | InterruptedException e) {
    e.printStackTrace();
  }
  if (exists==null){
    System.out.println("节点不存在");
  }
  else {
    System.out.println("节点存在");
  }


  //异步回调
  zooKeeper.exists("/mall",false, new AsyncCallback.StatCallback() {
    //第二个是path znode路径，第三个是ctx 后面传入实参，第四个是znode的状态
    public void processResult(int i, String s, Object o, Stat stat) {
      //如果节点不存在，返回的stat是null
      if (stat==null){
        System.out.println("节点不存在");
      }
      else{
        System.out.println("节点存在");
      }
    }
  // 传入ctx，Object类型
  },null);

//创建节点

  //同步方式
  try {
    //数据要写成byte[]，不携带数据写成null；默认acl权限使用ZooDefs.Ids.OPEN_ACL_UNSAFE；最后一个是节点类型，P是永久，E是临时，S是有序
    zooKeeper.create("/mall", "abcd".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println("已创建节点/mall");
    //如果节点已存在，会抛出异常
  } catch (KeeperException | InterruptedException e) {　　　　 System.out.println("创建节点/mall失败，请检查节点是否已存在");
    e.printStackTrace();
  }

  //异步回调
  zooKeeper.create("/mall", "abcd".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new AsyncCallback.Create2Callback(){
    //第二个path，第三个ctx，第四个节点状态
    public void processResult(int i, String s, Object o, String s1, Stat stat) {
      //回调方式不抛出异常，返回的stat是创建节点的状态，如果节点已存在，返回的stat是null
      if (stat==null){
        System.out.println("创建节点/mall失败，请检查节点是否已存在");
      }
      else {
        System.out.println("节点创建成功");
      }
    }
    //ctx实参
  },null);

//删除节点

  //同步方式
  try {
    //第二个参数是版本号，-1表示可以是任何版本
    zooKeeper.delete("/mall1",-1);
    System.out.println("成功删除节点/mall");
  } catch (InterruptedException | KeeperException e) {
    System.out.println("删除节点/mall失败");
    e.printStackTrace();
  }


  //异步回调
  zooKeeper.delete("/mall2", -1, new AsyncCallback.VoidCallback() {
    //第二个是path，第三个是ctx
    public void processResult(int i, String s, Object o) {
      
    }
  //
  //ctx实参
  },null);

//获取子节点列表，List，比如/mall/user,/mall/order，返回的是["user"、"order"]

  //同步方式
  List children = null;
  try {
    //第二个参数是watch
    children = zooKeeper.getChildren("/mall", false);
  } catch (KeeperException | InterruptedException e) {
    e.printStackTrace();
  }
  System.out.println("子节点列表：" + children);


  //异步
  zooKeeper.getChildren("/mall", false, new AsyncCallback.ChildrenCallback() {
    //第二个起依次是：path、ctx、返回的子节点列表
    public void processResult(int i, String s, Object o, List list) {
      System.out.println("子节点列表：" + list);
    }
  //ctx实参
  }, null);

//获取节点数据，返回byte[]

  //同步方式
  byte[] data = null;
  try {
    //第二个参数是watch，第三个是stat
    data = zooKeeper.getData("/mall", false, null);
  } catch (KeeperException | InterruptedException e) {
    e.printStackTrace();
  }
  //调用new String()时要判断data是否为null，如果是null会抛NPE
  if (data==null){
    System.out.println("该节点没有数据");
  }
  else{
    System.out.println("节点数据："+new String(data));
  }


  //异步回调
  zooKeeper.getData("/mall", false, new AsyncCallback.DataCallback() {
    //第二个起依次是：path、ctx、返回的节点数据、节点状态
    public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
      //不必判断bytes是否是null，如果节点没有数据，不会调用回调函数；执行到此，说明bytes不是null
      System.out.println("节点数据：" + new String(bytes) );
    }
    //ctx实参
  }, null);

//设置|更新节点据

  //同步方式
  try {
    //最后一个参数是版本号
    zooKeeper.setData("/mall", "1234".getBytes(), -1);
    System.out.println("设置节点数据成功");
  } catch (KeeperException | InterruptedException e) {
    System.out.println("设置节点数据失败");
    e.printStackTrace();
  }


  //异步回调
  zooKeeper.setData("/mall", "1234".getBytes(), -1, new AsyncCallback.StatCallback() {
    //第二个是path，第三个是ctx
    public void processResult(int i, String s, Object o, Stat stat) {

    }
  // ctx
  },null);

//设置acl权限
    
  //第一个参数指定权限,第二个是Id对象
  ACL acl = new ACL(ZooDefs.Perms.ALL, new Id("auth", "chy:abcd"));
  
  List aclList = new ArrayList<>();
  aclList.add(acl);
  
  //如果List中只有一个ACL对象，也可以这样写
  //List aclList = Collections.singletonList(auth);
    
  //验证权限，需写在设置权限之前。如果之前没有设置权限，也需要先验证本次即将设置的用户
  zooKeeper.addAuthInfo("digest","chy:abcd".getBytes());

  
  //方式一 setAcl
  try {
    //第二个参数是List，第三个参数是版本号
    zooKeeper.setACL("/mall", aclList, -1);
    System.out.println("设置权限成功");
  } catch (KeeperException | InterruptedException e) {
    System.out.println("设置权限失败");
    e.printStackTrace();
  }

  
  //方式二 在创建节点时设置权限
  try {
    zooKeeper.create("/mall","abcd".getBytes(),aclList,CreateMode.PERSISTENT);
    System.out.println("已创建节点并设置权限");
  } catch (KeeperException | InterruptedException e) {
    e.printStackTrace();
  }

//查看acl权限
    
  //设置权限之后，以后操作时需要先验证用户，一次session中验证一次即可
  zooKeeper.addAuthInfo("digest","chy:abcd".getBytes());

  
  //同步方式
  try {
    List aclList = zooKeeper.getACL("/mall", null);
    System.out.println("acl权限："+aclList);
  } catch (KeeperException | InterruptedException e) {
    System.out.println("获取acl权限失败");
    e.printStackTrace();
  }


  //异步回调
  zooKeeper.getACL("/mall3", null, new AsyncCallback.ACLCallback() {
    //第二个起：path、ctx、获取到的List、节点状态
    public void processResult(int i, String s, Object o, List list, Stat stat) {
      //就算没有手动设置acl权限，默认也是有值的
      System.out.println("acl权限："+list);
    }
  //ctx实参
  },null);

//添加监听 方式一
  try {
    CountDownLatch countDownLatch = new CountDownLatch(1);

    zooKeeper.getData("/mall", new Watcher() {
      public void process(WatchedEvent watchedEvent) {
        //watcher会监听该节点所有的事件，不管发生什么事件都会调用process()来处理，需要先判断一下事件类型
        if (watchedEvent.getType().equals(Event.EventType.NodeDataChanged)){
          System.out.println("节点数据改变了");
          //会一直监听，如果只监听一次数据改变，将下面这句代码取消注释即可
          //countDownLatch.countDown();
        }
      }
    }, null);
    //默认watcher是一次性的，如果要一直监听，需要借助CountDownLatch
    countDownLatch.await();
  } catch (KeeperException | InterruptedException e) {
    e.printStackTrace();
  }

//添加监听 方式二  
 　　try {
    CountDownLatch countDownLatch1 = new CountDownLatch(1);
    zooKeeper.addWatch("/mall", new Watcher() {
      @Override
      public void process(WatchedEvent watchedEvent) {
        if (watchedEvent.getType().equals(Event.EventType.NodeDataChanged)){
          System.out.println("节点数据改变了");
          //如果只监听一次数据改变，将下面这句代码注释掉
          //countDownLatch1.countDown();
        }
      }
    //监听模式，PERSISTENT是不监听子孙节点，PERSISTENT_RECURSIVE是递归监听子孙节点
    }, AddWatchMode.PERSISTENT_RECURSIVE);
    countDownLatch1.await();
  } catch (KeeperException | InterruptedException e) {
    e.printStackTrace();
  }

//移除监听 方式一
  try {
    zooKeeper.addWatch("/mall",null,AddWatchMode.PERSISTENT);
    System.out.println("已移除监听");
  } catch (KeeperException | InterruptedException e) {
    e.printStackTrace();
  }

//移除监听 方式二
  try {
    //第二个参数是Watcher，原来添加的那个Watcher监听对象，不能是null
    //第三个参数指定要移除监听的哪部分，Any是移除整个监听，Data是移除对数据的监听，Children是移除对子节点的递归监听
    //最后一个参数指定未连接到zkServe时，是否移除本地监听部分
    zooKeeper.removeWatches("/mall",watcher, Watcher.WatcherType.Children,true);
  } catch (InterruptedException | KeeperException e) {
    e.printStackTrace();
  }