Zookeeper:Java API

来自Wikioe
Eijux讨论 | 贡献2021年9月28日 (二) 18:29的版本 (建立内容为“category:Zookeeper __TOC__ == 关于 == Java 客户端包括两种方式: # 原生 API; # '''Curator'''; === Java 客户端搭建 === 使用的 IDE…”的新页面)
(差异) ←上一版本 | 最后版本 (差异) | 下一版本→ (差异)
跳到导航 跳到搜索


关于

Java 客户端包括两种方式:

  1. 原生 API;
  2. Curator

Java 客户端搭建

使用的 IDE 为 IntelliJ IDEA,创建一个 maven 工程,命名为 zookeeper-demo,并且引入如下依赖,可以自行在maven中央仓库选择合适的版本:

<dependency>
  <groupId>junit</groupId>
  <artifactId>junit</artifactId>
  <version>4.11</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.4.8</version>
</dependency>
<dependency>
  <groupId>org.apache.curator</groupId>
  <artifactId>curator-framework</artifactId>
  <version>4.0.0</version>
</dependency>
<dependency>
  <groupId>org.apache.curator</groupId>
  <artifactId>curator-recipes</artifactId>
  <version>4.0.0</version>
</dependency>

maven 工程目录结构:

文件:Zookeeper:Java客户端:maven工程目录结构.png

原生 API(Zookeeper API)


ZooKeeper 有一个绑定 Java 和 C 的官方API。Zookeeper 社区为大多数语言(.NET,python等)提供非官方API。


使用ZooKeeper API,应用程序可以连接,交互,操作数据,协调,最后断开与ZooKeeper集合的连接。

  • ZooKeeper API具有丰富的功能,以简单和安全的方式获得ZooKeeper集合的所有功能。
  • ZooKeeper API提供同步异步方法。


ZooKeeper集合和ZooKeeper API在各个方面都完全相辅相成,对开发人员有很大的帮助。

ZooKeeper API的基础知识

与ZooKeeper集合进行交互的应用程序称为 ZooKeeper客户端或简称客户端

  • Znode 是ZooKeeper集合的核心组件,ZooKeeper API提供了一小组方法使用ZooKeeper集合来操纵znode的所有细节。


客户端应该遵循以步骤,与ZooKeeper集合进行清晰和干净的交互:

  1. 连接到ZooKeeper集合。ZooKeeper集合为客户端分配会话ID。
  2. 定期向服务器发送心跳。否则,ZooKeeper集合将过期会话ID,客户端需要重新连接。
  3. 只要会话ID处于活动状态,就可以获取/设置znode。
  4. 所有任务完成后,断开与ZooKeeper集合的连接。如果客户端长时间不活动,则ZooKeeper集合将自动断开客户端。

org.apache.zookeeper.ZooKeeper

ZooKeeper API的核心部分是ZooKeeper类。它提供了在其构造函数中连接ZooKeeper集合的选项,并具有以下方法:

  • connect:连接到 ZooKeeper集合;
  • create:创建 znode;
  • exists:检查 znode 是否存在及其信息;
  • getData:从特定的 znode 获取数据;
  • setData:在特定的 znode 中设置数据;
  • getChildren:获取特定 znode 中的所有子节点;
  • delete:删除特定的 znode 及其所有子项;
  • close:关闭连接;


示例:

  1. 连接到 ZooKeeper 集合
    ZooKeeper 类通过其构造函数提供connect功能。构造函数的签名如下 :
    ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)
    
    • connectionString - ZooKeeper集合主机。
    • sessionTimeout - 会话超时(以毫秒为单位)。
    • watcher - 实现“监视器”界面的对象。ZooKeeper集合通过监视器对象返回连接状态。
    // import java classes
    import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    // import zookeeper classes
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.AsyncCallback.StatCallback;
    import org.apache.zookeeper.KeeperException.Code;
    import org.apache.zookeeper.data.Stat;
    
    public class ZooKeeperConnection {
       // declare zookeeper instance to access ZooKeeper ensemble
       private ZooKeeper zoo;
       final CountDownLatch connectedSignal = new CountDownLatch(1);
    
       // Method to connect zookeeper ensemble.
       public ZooKeeper connect(String host) throws IOException,InterruptedException {
          zoo = new ZooKeeper(host,5000,new Watcher() {
             public void process(WatchedEvent we) {
                if (we.getState() == KeeperState.SyncConnected) {
                   connectedSignal.countDown();
                }
             }
          });
    		
          connectedSignal.await();
          return zoo;
       }
    
       // Method to disconnect from zookeeper server
       public void close() throws InterruptedException {
          zoo.close();
       }
    }
    
  2. create 方法
    ZooKeeper类提供了在ZooKeeper集合中创建一个新的 znode 的 create 方法。create 方法的签名如下:
    create(String path, byte[] data, List<ACL> acl, CreateMode createMode)
    
    • path - Znode 路径。例如,/myapp1,/myapp2,/myapp1/mydata1,myapp2/mydata1/myanothersubdata
    • data - 要存储在指定 znode 路径中的数据
    • acl - 要创建的节点的访问控制列表。ZooKeeper API提供了一个静态接口 ZooDefs.Ids 来获取一些基本的acl列表。例如,ZooDefs.Ids.OPEN_ACL_UNSAFE返回打开znode的acl列表。
    • createMode - 节点的类型,即临时,顺序或两者。这是一个枚举
    import java.io.IOException;
    
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.ZooDefs;
    
    public class ZKCreate {
       // create static instance for zookeeper class.
       private static ZooKeeper zk;
       
       // create static instance for ZooKeeperConnection class.
       private static ZooKeeperConnection conn;
    
       // Method to create znode in zookeeper ensemble
       public static void create(String path, byte[] data) throws 
          KeeperException,InterruptedException {
          zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
          CreateMode.PERSISTENT);
       }
    
       public static void main(String[] args) {
          // znode path
          String path = "/MyFirstZnode"; // Assign path to znode
          // data in byte array
          byte[] data = "My first zookeeper app".getBytes(); // Declare data
    		
          try {
             conn = new ZooKeeperConnection();
             zk = conn.connect("localhost");
             create(path, data); // Create the data to the specified path
             conn.close();
          } catch (Exception e) {
             System.out.println(e.getMessage()); //Catch error message
          }
       }
    }
    
  3. exists 方法
    ZooKeeper类提供了 exists 方法来检查znode的存在。如果指定的znode存在,则返回一个znode的元数据Stat)。exists 方法的签名如下:
    exists(String path, boolean watcher)
    
    • path- Znode路径
    • watcher - 布尔值,用于指定是否监视指定的 znode
    import java.io.IOException;
    
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.data.Stat;
    
    public class ZKExists {
       private static ZooKeeper zk;
       private static ZooKeeperConnection conn;
    
       // Method to check existence of znode and its status, if znode is available.
       public static Stat znode_exists(String path) throws
          KeeperException,InterruptedException {
          return zk.exists(path, true);
       }
    
       public static void main(String[] args) throws InterruptedException,KeeperException {
          String path = "/MyFirstZnode"; // Assign znode to the specified path
    			
          try {
             conn = new ZooKeeperConnection();
             zk = conn.connect("localhost");
             Stat stat = znode_exists(path); // Stat checks the path of the znode
    				
             if(stat != null) {
                System.out.println("Node exists and the node version is " +
                stat.getVersion());
             } else {
                System.out.println("Node does not exists");
             }
    				
          } catch(Exception e) {
             System.out.println(e.getMessage()); // Catches error messages
          }
       }
    }
    
    Node exists and the node version is 1.
    
  4. getData 方法
    ZooKeeper类提供 getData 方法来获取附加在指定znode中的数据及其状态。 getData 方法的签名如下:
    getData(String path, Watcher watcher, Stat stat)
    
    • path - Znode路径。
    • watcher - 监视器类型的回调函数。当指定的znode的数据改变时,ZooKeeper集合将通过监视器回调进行通知。这是一次性通知。
    • stat - 返回znode的元数据。
    import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.data.Stat;
    
    public class ZKGetData {
    
       private static ZooKeeper zk;
       private static ZooKeeperConnection conn;
       
       public static Stat znode_exists(String path) throws 
          KeeperException,InterruptedException {
          return zk.exists(path,true);
       }
    
       public static void main(String[] args) throws InterruptedException, KeeperException {
          String path = "/MyFirstZnode";
          final CountDownLatch connectedSignal = new CountDownLatch(1);
    		
          try {
             conn = new ZooKeeperConnection();
             zk = conn.connect("localhost");
             Stat stat = znode_exists(path);
    			
             if(stat != null) {
                byte[] b = zk.getData(path, new Watcher() {
                   public void process(WatchedEvent we) {
                      if (we.getType() == Event.EventType.None) {
                         switch(we.getState()) {
                            case Expired:
                            connectedSignal.countDown();
                            break;
                         }
                      } else {
                         String path = "/MyFirstZnode";
                         try {
                            byte[] bn = zk.getData(path,
                            false, null);
                            String data = new String(bn,
                            "UTF-8");
                            System.out.println(data);
                            connectedSignal.countDown();
                         } catch(Exception ex) {
                            System.out.println(ex.getMessage());
                         }
                      }
                   }
                }, null);
    				
                String data = new String(b, "UTF-8");
                System.out.println(data);
                connectedSignal.await();
             } else {
                System.out.println("Node does not exists");
             }
          } catch(Exception e) {
            System.out.println(e.getMessage());
          }
       }
    }
    
  5. setData 方法
    ZooKeeper类提供 setData 方法来修改指定znode中附加的数据。 setData 方法的签名如下:
    setData(String path, byte[] data, int version)
    
    • path- Znode路径
    • data - 要存储在指定znode路径中的数据。
    • version- znode的当前版本。每当数据更改时,ZooKeeper会更新znode的版本号。
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    
    import java.io.IOException;
    
    public class ZKSetData {
       private static ZooKeeper zk;
       private static ZooKeeperConnection conn;
    
       // Method to update the data in a znode. Similar to getData but without watcher.
       public static void update(String path, byte[] data) throws
          KeeperException,InterruptedException {
          zk.setData(path, data, zk.exists(path,true).getVersion());
       }
    
       public static void main(String[] args) throws InterruptedException,KeeperException {
          String path= "/MyFirstZnode";
          byte[] data = "Success".getBytes(); //Assign data which is to be updated.
    		
          try {
             conn = new ZooKeeperConnection();
             zk = conn.connect("localhost");
             update(path, data); // Update znode data to the specified path
          } catch(Exception e) {
             System.out.println(e.getMessage());
          }
       }
    }
    
  6. getChildren 方法
    ZooKeeper类提供 getChildren 方法来获取特定znode的所有子节点。 getChildren 方法的签名如下:
    getChildren(String path, Watcher watcher)
    
    • path - Znode路径。
    • watcher - 监视器类型的回调函数。当指定的znode被删除或znode下的子节点被创建/删除时,ZooKeeper集合将进行通知。这是一次性通知。
    import java.io.IOException;
    import java.util.*;
    
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.data.Stat;
    
    public class ZKGetChildren {
       private static ZooKeeper zk;
       private static ZooKeeperConnection conn;
    
       // Method to check existence of znode and its status, if znode is available.
       public static Stat znode_exists(String path) throws 
          KeeperException,InterruptedException {
          return zk.exists(path,true);
       }
    
       public static void main(String[] args) throws InterruptedException,KeeperException {
          String path = "/MyFirstZnode"; // Assign path to the znode
    		
          try {
             conn = new ZooKeeperConnection();
             zk = conn.connect("localhost");
             Stat stat = znode_exists(path); // Stat checks the path
    
             if(stat!= null) {
                //“getChildren" method- get all the children of znode.It has two args, path and watch
                List <String> children = zk.getChildren(path, false);
                for(int i = 0; i < children.size(); i++)
                   System.out.println(children.get(i)); //Print children's
             } else {
                System.out.println("Node does not exists");
             }
          } catch(Exception e) {
             System.out.println(e.getMessage());
          }
       }
    }
    
  7. delete 方法
    ZooKeeper类提供了 delete 方法来删除指定的znode。 delete 方法的签名如下:
    delete(String path, int version)
    
    • path - Znode路径。
    • version - znode的当前版本。
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.KeeperException;
    
    public class ZKDelete {
       private static ZooKeeper zk;
       private static ZooKeeperConnection conn;
    
       // Method to check existence of znode and its status, if znode is available.
       public static void delete(String path) throws KeeperException,InterruptedException {
          zk.delete(path,zk.exists(path,true).getVersion());
       }
    
       public static void main(String[] args) throws InterruptedException,KeeperException {
          String path = "/MyFirstZnode"; //Assign path to the znode
    		
          try {
             conn = new ZooKeeperConnection();
             zk = conn.connect("localhost");
             delete(path); //delete the node with the specified path
          } catch(Exception e) {
             System.out.println(e.getMessage()); // catches error messages
          }
       }
    }
    

Curator


Curator 是 Netflix 公司开源的一套 zookeeper 客户端框架,解决了很多 Zookeeper 客户端非常底层的细节开发工作,包括连接重连、反复注册 Watcher 和 NodeExistsException 异常等。


Curator 包含了几个包:

  • curator-framework:对 zookeeper 的底层 api 的一些封装。
  • curator-client:提供一些客户端的操作,例如重试策略等。
  • curator-recipes:封装了一些高级特性,如:Cache 事件监听、选举、分布式锁、分布式计数器、分布式 Barrier 等。【!!!】


示例:

public class CuratorDemo {
    public static void main(String[] args) throws Exception {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.
                builder().connectString("192.168.3.33:2181," +
                "192.168.3.35:2181,192.168.3.37:2181").
                sessionTimeoutMs(4000).retryPolicy(new
                ExponentialBackoffRetry(1000,3)).
                namespace("").build();
        curatorFramework.start();
        
        Stat stat=new Stat();
        //查询节点数据
        byte[] bytes = curatorFramework.getData().storingStatIn(stat).forPath("/runoob");
        System.out.println(new String(bytes));
        curatorFramework.close();
    }
}
上一步设置了 /runoob 节点值,所以控制台输出。
文件:Zookeeper:Java客户端:Curator示例.png