閱讀(88k) 書(shū)簽贊(15) 我要糾錯(cuò)

Zookeeper API

2018-10-28 17:35 更新

ZooKeeper有一個(gè)綁定Java和C的官方API。Zookeeper社區(qū)為大多數(shù)語(yǔ)言（.NET，python等）提供非官方API。使用ZooKeeper API，應(yīng)用程序可以連接，交互，操作數(shù)據(jù)，協(xié)調(diào)，最后斷開(kāi)與ZooKeeper集合的連接。

ZooKeeper API具有豐富的功能，以簡(jiǎn)單和安全的方式獲得ZooKeeper集合的所有功能。ZooKeeper API提供同步和異步方法。

ZooKeeper集合和ZooKeeper API在各個(gè)方面都完全相輔相成，對(duì)開(kāi)發(fā)人員有很大的幫助。讓我們?cè)诒菊掠懻揓ava綁定。

ZooKeeper API的基礎(chǔ)知識(shí)

與ZooKeeper集合進(jìn)行交互的應(yīng)用程序稱(chēng)為 ZooKeeper客戶(hù)端或簡(jiǎn)稱(chēng)客戶(hù)端。

Znode是ZooKeeper集合的核心組件，ZooKeeper API提供了一小組方法使用ZooKeeper集合來(lái)操縱znode的所有細(xì)節(jié)。

客戶(hù)端應(yīng)該遵循以步驟，與ZooKeeper集合進(jìn)行清晰和干凈的交互。

連接到ZooKeeper集合。ZooKeeper集合為客戶(hù)端分配會(huì)話(huà)ID。
定期向服務(wù)器發(fā)送心跳。否則，ZooKeeper集合將過(guò)期會(huì)話(huà)ID，客戶(hù)端需要重新連接。
只要會(huì)話(huà)ID處于活動(dòng)狀態(tài)，就可以獲取/設(shè)置znode。
所有任務(wù)完成后，斷開(kāi)與ZooKeeper集合的連接。如果客戶(hù)端長(zhǎng)時(shí)間不活動(dòng)，則ZooKeeper集合將自動(dòng)斷開(kāi)客戶(hù)端。

Java綁定

讓我們來(lái)了解本章中最重要的一組ZooKeeper API。ZooKeeper API的核心部分是ZooKeeper類(lèi)。它提供了在其構(gòu)造函數(shù)中連接ZooKeeper集合的選項(xiàng)，并具有以下方法：

connect - 連接到ZooKeeper集合
create- 創(chuàng)建znode
exists- 檢查znode是否存在及其信息
getData - 從特定的znode獲取數(shù)據(jù)
setData - 在特定的znode中設(shè)置數(shù)據(jù)
getChildren - 獲取特定znode中的所有子節(jié)點(diǎn)
delete - 刪除特定的znode及其所有子項(xiàng)
close - 關(guān)閉連接

連接到ZooKeeper集合

ZooKeeper類(lèi)通過(guò)其構(gòu)造函數(shù)提供connect功能。構(gòu)造函數(shù)的簽名如下 :

ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

connectionString - ZooKeeper集合主機(jī)。
sessionTimeout - 會(huì)話(huà)超時(shí)（以毫秒為單位）。
watcher - 實(shí)現(xiàn)“監(jiān)視器”界面的對(duì)象。ZooKeeper集合通過(guò)監(jiān)視器對(duì)象返回連接狀態(tài)。

讓我們創(chuàng)建一個(gè)新的幫助類(lèi) ZooKeeperConnection ，并添加一個(gè)方法 connect 。 connect 方法創(chuàng)建一個(gè)ZooKeeper對(duì)象，連接到ZooKeeper集合，然后返回對(duì)象。

這里 CountDownLatch 用于停止（等待）主進(jìn)程，直到客戶(hù)端與ZooKeeper集合連接。

ZooKeeper集合通過(guò)監(jiān)視器回調(diào)來(lái)回復(fù)連接狀態(tài)。一旦客戶(hù)端與ZooKeeper集合連接，監(jiān)視器回調(diào)就會(huì)被調(diào)用，并且監(jiān)視器回調(diào)函數(shù)調(diào)用CountDownLatch的countDown方法來(lái)釋放鎖，在主進(jìn)程中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();
   }
}

保存上面的代碼，它將在下一節(jié)中用于連接ZooKeeper集合。

創(chuàng)建Znode

ZooKeeper類(lèi)提供了在ZooKeeper集合中創(chuàng)建一個(gè)新的znode的create方法。 create 方法的簽名如下：

create(String path, byte[] data, List<ACL> acl, CreateMode createMode)

path - Znode路徑。例如，/myapp1，/myapp2，/myapp1/mydata1，myapp2/mydata1/myanothersubdata
data - 要存儲(chǔ)在指定znode路徑中的數(shù)據(jù)
acl - 要?jiǎng)?chuàng)建的節(jié)點(diǎn)的訪問(wèn)控制列表。ZooKeeper API提供了一個(gè)靜態(tài)接口 ZooDefs.Ids 來(lái)獲取一些基本的acl列表。例如，ZooDefs.Ids.OPEN_ACL_UNSAFE返回打開(kāi)znode的acl列表。
createMode - 節(jié)點(diǎn)的類(lèi)型，即臨時(shí)，順序或兩者。這是一個(gè)枚舉。

讓我們創(chuàng)建一個(gè)新的Java應(yīng)用程序來(lái)檢查ZooKeeper API的 create 功能。創(chuàng)建文件 ZKCreate.java 。在main方法中，創(chuàng)建一個(gè)類(lèi)型為 ZooKeeperConnection 的對(duì)象，并調(diào)用 connect 方法連接到ZooKeeper集合。

connect方法將返回ZooKeeper對(duì)象 zk 。現(xiàn)在，請(qǐng)使用自定義path和data調(diào)用 zk 對(duì)象的 create 方法。

創(chuàng)建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
      }
   }
}

一旦編譯和執(zhí)行應(yīng)用程序，將在ZooKeeper集合中創(chuàng)建具有指定數(shù)據(jù)的znode。你可以使用ZooKeeper CLI zkCli.sh 進(jìn)行檢查。

cd /path/to/zookeeper
bin/zkCli.sh
>>> get /MyFirstZnode

Exists - 檢查Znode的存在

ZooKeeper類(lèi)提供了 exists 方法來(lái)檢查znode的存在。如果指定的znode存在，則返回一個(gè)znode的元數(shù)據(jù)。exists方法的簽名如下：

exists(String path, boolean watcher)

path- Znode路徑
watcher - 布爾值，用于指定是否監(jiān)視指定的znode

讓我們創(chuàng)建一個(gè)新的Java應(yīng)用程序來(lái)檢查ZooKeeper API的“exists”功能。創(chuàng)建文件“ZKExists.java”。在main方法中，使用“ZooKeeperConnection”對(duì)象創(chuàng)建ZooKeeper對(duì)象“zk”。然后，使用自定義“path”調(diào)用“zk”對(duì)象的“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
      }
   }
}

一旦編譯和執(zhí)行應(yīng)用程序，你將獲得以下輸出。

Node exists and the node version is 1.

getData方法

ZooKeeper類(lèi)提供 getData 方法來(lái)獲取附加在指定znode中的數(shù)據(jù)及其狀態(tài)。 getData 方法的簽名如下：

getData(String path, Watcher watcher, Stat stat)

path - Znode路徑。
watcher - 監(jiān)視器類(lèi)型的回調(diào)函數(shù)。當(dāng)指定的znode的數(shù)據(jù)改變時(shí)，ZooKeeper集合將通過(guò)監(jiān)視器回調(diào)進(jìn)行通知。這是一次性通知。
stat - 返回znode的元數(shù)據(jù)。

讓我們創(chuàng)建一個(gè)新的Java應(yīng)用程序來(lái)了解ZooKeeper API的 getData 功能。創(chuàng)建文件 ZKGetData.java 。在main方法中，使用 ZooKeeperConnection 對(duì)象創(chuàng)建一個(gè)ZooKeeper對(duì)象 zk 。然后，使用自定義路徑調(diào)用zk對(duì)象的 getData 方法。

下面是從指定節(jié)點(diǎn)獲取數(shù)據(jù)的完整程序代碼：

編碼：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());
      }
   }
}

一旦編譯和執(zhí)行應(yīng)用程序，你將獲得以下輸出

My first zookeeper app

應(yīng)用程序?qū)⒌却齔ooKeeper集合的進(jìn)一步通知。使用ZooKeeper CLI zkCli.sh 更改指定znode的數(shù)據(jù)。

cd /path/to/zookeeper
bin/zkCli.sh
>>> set /MyFirstZnode Hello

現(xiàn)在，應(yīng)用程序?qū)⒋蛴∫韵螺敵霾⑼顺觥?/span>

Hello

setData方法

ZooKeeper類(lèi)提供 setData 方法來(lái)修改指定znode中附加的數(shù)據(jù)。 setData 方法的簽名如下：

setData(String path, byte[] data, int version)

path- Znode路徑
data - 要存儲(chǔ)在指定znode路徑中的數(shù)據(jù)。
version- znode的當(dāng)前版本。每當(dāng)數(shù)據(jù)更改時(shí)，ZooKeeper會(huì)更新znode的版本號(hào)。

現(xiàn)在讓我們創(chuàng)建一個(gè)新的Java應(yīng)用程序來(lái)了解ZooKeeper API的 setData 功能。創(chuàng)建文件 ZKSetData.java 。在main方法中，使用 ZooKeeperConnection 對(duì)象創(chuàng)建一個(gè)ZooKeeper對(duì)象 zk 。然后，使用指定的路徑，新數(shù)據(jù)和節(jié)點(diǎn)版本調(diào)用 zk 對(duì)象的 setData 方法。

以下是修改附加在指定znode中的數(shù)據(jù)的完整程序代碼。

編碼：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());
      }
   }
}

編譯并執(zhí)行應(yīng)用程序后，指定的znode的數(shù)據(jù)將被改變，并且可以使用ZooKeeper CLI zkCli.sh 進(jìn)行檢查。

cd /path/to/zookeeper
bin/zkCli.sh
>>> get /MyFirstZnode

getChildren方法

ZooKeeper類(lèi)提供 getChildren 方法來(lái)獲取特定znode的所有子節(jié)點(diǎn)。 getChildren 方法的簽名如下：

getChildren(String path, Watcher watcher)

path - Znode路徑。
watcher - 監(jiān)視器類(lèi)型的回調(diào)函數(shù)。當(dāng)指定的znode被刪除或znode下的子節(jié)點(diǎn)被創(chuàng)建/刪除時(shí)，ZooKeeper集合將進(jìn)行通知。這是一次性通知。

編碼：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());
      }

   }

}

在運(yùn)行程序之前，讓我們使用ZooKeeper CLI zkCli.sh 為 /MyFirstZnode 創(chuàng)建兩個(gè)子節(jié)點(diǎn)。

cd /path/to/zookeeper
bin/zkCli.sh
>>> create /MyFirstZnode/myfirstsubnode Hi
>>> create /MyFirstZnode/mysecondsubmode Hi

現(xiàn)在，編譯和運(yùn)行程序?qū)⑤敵錾厦鎰?chuàng)建的znode。

myfirstsubnode
mysecondsubnode

刪除Znode

ZooKeeper類(lèi)提供了 delete 方法來(lái)刪除指定的znode。 delete 方法的簽名如下：

delete(String path, int version)

path - Znode路徑。
version - znode的當(dāng)前版本。

讓我們創(chuàng)建一個(gè)新的Java應(yīng)用程序來(lái)了解ZooKeeper API的 delete 功能。創(chuàng)建文件 ZKDelete.java 。在main方法中，使用 ZooKeeperConnection 對(duì)象創(chuàng)建一個(gè)ZooKeeper對(duì)象 zk 。然后，使用指定的路徑和版本號(hào)調(diào)用 zk 對(duì)象的 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
      }
   }
}

以上內(nèi)容是否對(duì)您有幫助：

← Zookeeper CLI

Zookeeper 應(yīng)用程序 →

寫(xiě)筆記

我要補(bǔ)充