Zookeeper - API

上一頁
下一頁

ZooKeeper 為 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 類透過其建構函式提供連線功能。建構函式的簽名如下：

ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

其中，

  • connectionString - ZooKeeper 叢集主機。

  • sessionTimeout - 以毫秒為單位的會話超時時間。

  • watcher - 實現“Watcher”介面的物件。ZooKeeper 叢集透過 watcher 物件返回連線狀態。

讓我們建立一個新的輔助類ZooKeeperConnection並新增一個方法connect。connect方法建立一個 ZooKeeper 物件，連線到 ZooKeeper 叢集，然後返回該物件。

這裡CountDownLatch用於停止（等待）主程序，直到客戶端連線到 ZooKeeper 叢集。

ZooKeeper 叢集透過Watcher 回撥回覆連線狀態。一旦客戶端連線到 ZooKeeper 叢集，Watcher 回撥將被呼叫，並且 Watcher 回撥呼叫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 類提供create 方法以在 ZooKeeper 叢集中建立一個新的 znode。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 CLI zkCli.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 - 型別為Watcher的回撥函式。當指定 znode 的資料發生更改時，ZooKeeper 叢集將透過 Watcher 回調發出通知。這是一次性通知。

  • 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 - 型別為“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。然後，使用指定的path和節點版本呼叫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
      }
   }
}
列印頁面

上一頁
下一頁
廣告
© . All rights reserved.