Zookeeper API
Zookeeper API
由 xpproen 創建,youj 最後一次修改 2016-12-27ZooKeeper有一個綁定Java和C的官方API。Zookeeper社區為大多數語言(.NET,python等)提供非官方API。使用ZooKeeper API,應用程序可以連接,交互,操作數據,協調,最後斷開與ZooKeeper集合的連接。
ZooKeeper API具有豐富的功能,以簡單和安全的方式獲得ZooKeeper集合的所有功能。ZooKeeper API提供同步和異步方法。
ZooKeeper集合和ZooKeeper API在各個方面都完全相輔相成,對開發人員有很大的幫助。讓我們在本章討論Java綁定。
ZooKeeper API的基礎知識
與ZooKeeper集合進行交互的應用程序稱為 ZooKeeper客戶端或簡稱客戶端。
Znode是ZooKeeper集合的核心組件,ZooKeeper API提供了一小組方法使用ZooKeeper集合來操縱znode的所有細節。
客戶端應該遵循以步驟,與ZooKeeper集合進行清晰和幹凈的交互。
-
連接到ZooKeeper集合。ZooKeeper集合為客戶端分配會話ID。
-
定期向服務器發送心跳。否則,ZooKeeper集合將過期會話ID,客戶端需要重新連接。
-
只要會話ID處於活動狀態,就可以獲取/設置znode。
-
所有任務完成後,斷開與ZooKeeper集合的連接。如果客戶端長時間不活動,則ZooKeeper集合將自動斷開客戶端。
Java綁定
讓我們來了解本章中最重要的一組ZooKeeper API。ZooKeeper API的核心部分是ZooKeeper類。它提供了在其構造函數中連接ZooKeeper集合的選項,並具有以下方法:
-
connect - 連接到ZooKeeper集合
-
create- 創建znode
-
exists- 檢查znode是否存在及其信息
-
getData - 從特定的znode獲取數據
-
setData - 在特定的znode中設置數據
-
getChildren - 獲取特定znode中的所有子節點
-
delete - 刪除特定的znode及其所有子項
-
close - 關閉連接
連接到ZooKeeper集合
ZooKeeper類通過其構造函數提供connect功能。構造函數的簽名如下 :
ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)
-
connectionString - ZooKeeper集合主機。
-
sessionTimeout - 會話超時(以毫秒為單位)。
-
watcher - 實現“監視器”界面的對象。ZooKeeper集合通過監視器對象返回連接狀態。
讓我們創建一個新的幫助類 ZooKeeperConnection ,並添加一個方法 connect 。 connect 方法創建一個ZooKeeper對象,連接到ZooKeeper集合,然後返回對象。
這裏 CountDownLatch 用於停止(等待)主進程,直到客戶端與ZooKeeper集合連接。
ZooKeeper集合通過監視器回調來回復連接狀態。一旦客戶端與ZooKeeper集合連接,監視器回調就會被調用,並且監視器回調函數調用CountDownLatch的countDown方法來釋放鎖,在主進程中await。
以下是與ZooKeeper集合連接的完整代碼。
編碼:ZooKeeperConnection.java
// 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(); } }
保存上面的代碼,它將在下一節中用於連接ZooKeeper集合。
創建Znode
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 - 節點的類型,即臨時,順序或兩者。這是一個枚舉。
讓我們創建一個新的Java應用程序來檢查ZooKeeper API的 create 功能。創建文件 ZKCreate.java 。在main方法中,創建一個類型為 ZooKeeperConnection 的對象,並調用 connect 方法連接到ZooKeeper集合。
connect方法將返回ZooKeeper對象 zk 。現在,請使用自定義path和data調用 zk 對象的 create 方法。
創建znode的完整程序代碼如下:
編碼:ZKCreate.java
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 } } }
一旦編譯和執行應用程序,將在ZooKeeper集合中創建具有指定數據的znode。你可以使用ZooKeeper CLIzkCli.sh 進行檢查。
cd /path/to/zookeeper bin/zkCli.sh >>> get /MyFirstZnode
Exists - 檢查Znode的存在
ZooKeeper類提供了 exists 方法來檢查znode的存在。如果指定的znode存在,則返回一個znode的元數據。exists方法的簽名如下:
exists(String path, boolean watcher)
-
path- Znode路徑
-
watcher - 布爾值,用於指定是否監視指定的znode
讓我們創建一個新的Java應用程序來檢查ZooKeeper API的“exists”功能。創建文件“ZKExists.java”。在main方法中,使用“ZooKeeperConnection”對象創建ZooKeeper對象“zk”。然後,使用自定義“path”調用“zk”對象的“exists”方法。完整的列表如下:
編碼:ZKExists.java
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.
getData方法
ZooKeeper類提供 getData 方法來獲取附加在指定znode中的數據及其狀態。 getData 方法的簽名如下:
getData(String path, Watcher watcher, Stat stat)
-
path - Znode路徑。
-
watcher - 監視器類型的回調函數。當指定的znode的數據改變時,ZooKeeper集合將通過監視器回調進行通知。這是一次性通知。
-
stat - 返回znode的元數據。
讓我們創建一個新的Java應用程序來了解ZooKeeper API的 getData 功能。創建文件 ZKGetData.java 。在main方法中,使用 ZooKeeperConnection 對象創建一個ZooKeeper對象 zk 。然後,使用自定義路徑調用zk對象的 getData 方法。
下面是從指定節點獲取數據的完整程序代碼:
編碼:ZKGetData.java
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()); } } }
一旦編譯和執行應用程序,你將獲得以下輸出
My first zookeeper app
應用程序將等待ZooKeeper集合的進一步通知。使用ZooKeeper CLI zkCli.sh 更改指定znode的數據。
cd /path/to/zookeeper bin/zkCli.sh >>> set /MyFirstZnode Hello
現在,應用程序將打印以下輸出並退出。
Hello
setData方法
ZooKeeper類提供 setData 方法來修改指定znode中附加的數據。 setData 方法的簽名如下:
setData(String path, byte[] data, int version)
-
path- Znode路徑
-
data - 要存儲在指定znode路徑中的數據。
-
version- znode的當前版本。每當數據更改時,ZooKeeper會更新znode的版本號。
現在讓我們創建一個新的Java應用程序來了解ZooKeeper API的 setData 功能。創建文件 ZKSetData.java。在main方法中,使用 ZooKeeperConnection 對象創建一個ZooKeeper對象 zk 。然後,使用指定的路徑,新數據和節點版本調用 zk 對象的 setData 方法。
以下是修改附加在指定znode中的數據的完整程序代碼。
編碼:ZKSetData.java
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()); } } }
編譯並執行應用程序後,指定的znode的數據將被改變,並且可以使用ZooKeeper CLI zkCli.sh 進行檢查。
cd /path/to/zookeeper bin/zkCli.sh >>> get /MyFirstZnode
getChildren方法
ZooKeeper類提供 getChildren 方法來獲取特定znode的所有子節點。 getChildren 方法的簽名如下:
getChildren(String path, Watcher watcher)
-
path - Znode路徑。
-
watcher - 監視器類型的回調函數。當指定的znode被刪除或znode下的子節點被創建/刪除時,ZooKeeper集合將進行通知。這是一次性通知。
編碼:ZKGetChildren.java
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()); } } }
在運行程序之前,讓我們使用ZooKeeper CLI zkCli.sh 為 /MyFirstZnode 創建兩個子節點。
cd /path/to/zookeeper bin/zkCli.sh >>> create /MyFirstZnode/myfirstsubnode Hi >>> create /MyFirstZnode/mysecondsubmode Hi
現在,編譯和運行程序將輸出上面創建的znode。
myfirstsubnode mysecondsubnode
刪除Znode
ZooKeeper類提供了 delete 方法來刪除指定的znode。 delete 方法的簽名如下:
delete(String path, int version)
-
path - Znode路徑。
-
version - znode的當前版本。
讓我們創建一個新的Java應用程序來了解ZooKeeper API的 delete 功能。創建文件 ZKDelete.java 。在main方法中,使用 ZooKeeperConnection 對象創建一個ZooKeeper對象 zk 。然後,使用指定的路徑和版本號調用 zk 對象的 delete 方法。
刪除znode的完整程序代碼如下:
編碼:ZKDelete.java
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 } } }
Zookeeper API