使用 ZooKeeper 進行程式設計 - 基本教學
簡介
在本教學中，我們將展示如何使用 ZooKeeper 實作屏障和生產者-消費者佇列。我們將類別分別稱為 Barrier 和 Queue。這些範例假設您至少執行一個 ZooKeeper 伺服器。
兩個原語都使用以下常見程式碼摘錄
static ZooKeeper zk = null;
static Integer mutex;
String root;
SyncPrimitive(String address) {
if(zk == null){
try {
System.out.println("Starting ZK:");
zk = new ZooKeeper(address, 3000, this);
mutex = new Integer(-1);
System.out.println("Finished starting ZK: " + zk);
} catch (IOException e) {
System.out.println(e.toString());
zk = null;
}
}
}
synchronized public void process(WatchedEvent event) {
synchronized (mutex) {
mutex.notify();
}
}
兩個類別都延伸 SyncPrimitive。這樣一來，我們可以在 SyncPrimitive 的建構函式中執行所有原語的共通步驟。為了讓範例保持簡潔，我們會在第一次建立屏障物件或佇列物件時建立 ZooKeeper 物件，並宣告一個靜態變數，作為此物件的參考。後續的 Barrier 和 Queue 執行個體會檢查 ZooKeeper 物件是否存在。或者，我們也可以讓應用程式建立 ZooKeeper 物件，並將其傳遞給 Barrier 和 Queue 的建構函式。
我們使用 process() 方法來處理因監控觸發的通知。在以下討論中，我們會提供設定監控的程式碼。監控是內部結構，可讓 ZooKeeper 通知客戶端節點的變更。例如，如果客戶端正在等待其他客戶端離開屏障，則可以設定監控並等待特定節點的修改，這表示等待結束。我們在探討範例後，這一點就會變得清楚。
屏障
屏障是一種原語，讓一群程序能同步計算的開始與結束。此實作的一般概念是有一個屏障節點，作為個別程序節點的父節點。假設我們稱屏障節點為「/b1」。然後每個程序「p」建立一個節點「/b1/p」。一旦有足夠的程序建立了對應的節點，加入的程序就能開始計算。
在此範例中，每個程序實例化一個 Barrier 物件，其建構函式帶有下列參數
- ZooKeeper 伺服器的位址（例如「zoo1.foo.com:2181」）
- ZooKeeper 上屏障節點的路徑（例如「/b1」）
- 程序群組的大小
Barrier 的建構函式將 Zookeeper 伺服器的位址傳遞給父類別的建構函式。如果不存在 ZooKeeper 實例，父類別會建立一個。然後 Barrier 的建構函式會在 ZooKeeper 上建立一個屏障節點，這是所有程序節點的父節點，我們稱之為根（注意：這不是 ZooKeeper 根目錄「/」）。
/**
* Barrier constructor
*
* @param address
* @param root
* @param size
*/
Barrier(String address, String root, int size) {
super(address);
this.root = root;
this.size = size;
// Create barrier node
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out
.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
// My node name
try {
name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());
} catch (UnknownHostException e) {
System.out.println(e.toString());
}
}
若要進入屏障，程序會呼叫 enter()。程序會在根目錄下建立一個節點來表示它，使用其主機名稱來形成節點名稱。然後它會等到有足夠的程序進入屏障。程序會透過使用「getChildren()」檢查根節點有多少個子節點，並在子節點數量不足時等待通知來達成此目的。若要收到根節點變更的通知，程序必須設定監控，並透過呼叫「getChildren()」來執行此動作。在程式碼中，我們有「getChildren()」有兩個參數。第一個參數指定要讀取的節點，第二個參數是一個布林旗標，讓程序能設定監控。在程式碼中，旗標為 true。
/**
* Join barrier
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean enter() throws KeeperException, InterruptedException{
zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() < size) {
mutex.wait();
} else {
return true;
}
}
}
}
請注意，enter() 會擲出 KeeperException 和 InterruptedException，因此應用程式有責任捕捉並處理此類例外狀況。
計算完成後，一個程序會呼叫 leave() 來離開屏障。它首先會刪除其對應的節點，然後取得根節點的子節點。如果至少有一個子節點，則它會等待通知（注意：呼叫 getChildren() 的第二個參數為 true，表示 ZooKeeper 必須在根節點上設定監控）。在收到通知後，它會再次檢查根節點是否有任何子節點。
/**
* Wait until all reach barrier
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean leave() throws KeeperException, InterruptedException {
zk.delete(root + "/" + name, 0);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() > 0) {
mutex.wait();
} else {
return true;
}
}
}
}
生產者-消費者佇列
生產者-消費者佇列是一個分散式資料結構，由程序群組用於產生和使用項目。生產者程序會建立新元素並將其新增至佇列。消費者程序會從清單中移除元素並處理它們。在此實作中，元素是簡單的整數。佇列由根節點表示，若要將元素新增至佇列，生產者程序會建立一個新節點，作為根節點的子節點。
下列程式碼摘錄對應至物件的建構函式。與 Barrier 物件一樣，它首先會呼叫父類別 SyncPrimitive 的建構函式，如果不存在 ZooKeeper 物件，則會建立一個。然後它會驗證佇列的根節點是否存在，如果不存在，則會建立。
/**
* Constructor of producer-consumer queue
*
* @param address
* @param name
*/
Queue(String address, String name) {
super(address);
this.root = name;
// Create ZK node name
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out
.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
}
生產者程序會呼叫「produce()」來將元素新增至佇列，並傳遞一個整數作為引數。若要將元素新增至佇列，此方法會使用「create()」建立一個新節點，並使用 SEQUENCE 旗標指示 ZooKeeper 附加與根節點關聯的排序器計數器的值。這樣一來，我們就能對佇列的元素施加一個總順序，從而保證佇列中最舊的元素會是下一個被使用的元素。
/**
* Add element to the queue.
*
* @param i
* @return
*/
boolean produce(int i) throws KeeperException, InterruptedException{
ByteBuffer b = ByteBuffer.allocate(4);
byte[] value;
// Add child with value i
b.putInt(i);
value = b.array();
zk.create(root + "/element", value, Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT_SEQUENTIAL);
return true;
}
若要使用元素，消費者程序會取得根節點的子節點，讀取計數器值最小的節點，並傳回元素。請注意，如果發生衝突，則兩個競爭程序中的其中一個將無法刪除節點，而刪除作業會擲回例外狀況。
呼叫 getChildren() 會傳回依序排列的子項清單。由於依序排列不一定會遵循計數器值的數值順序，因此我們需要決定哪個元素最小。為了決定哪一個元素具有最小的計數器值，我們會遍歷清單，並從每個元素中移除前綴「element」。
/**
* Remove first element from the queue.
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
int consume() throws KeeperException, InterruptedException{
int retvalue = -1;
Stat stat = null;
// Get the first element available
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() == 0) {
System.out.println("Going to wait");
mutex.wait();
} else {
Integer min = new Integer(list.get(0).substring(7));
for(String s : list){
Integer tempValue = new Integer(s.substring(7));
//System.out.println("Temporary value: " + tempValue);
if(tempValue < min) min = tempValue;
}
System.out.println("Temporary value: " + root + "/element" + min);
byte[] b = zk.getData(root + "/element" + min,
false, stat);
zk.delete(root + "/element" + min, 0);
ByteBuffer buffer = ByteBuffer.wrap(b);
retvalue = buffer.getInt();
return retvalue;
}
}
}
}
}
完整範例
在以下區段中，您可以找到一個完整的命令列應用程式，以示範上述提到的範例。使用下列指令來執行它。
ZOOBINDIR="[path_to_distro]/bin"
. "$ZOOBINDIR"/zkEnv.sh
java SyncPrimitive [Test Type] [ZK server] [No of elements] [Client type]
佇列測試
啟動一個產生器來建立 100 個元素
java SyncPrimitive qTest localhost 100 p
啟動一個消費者來使用 100 個元素
java SyncPrimitive qTest localhost 100 c
屏障測試
啟動一個具有 2 個參與者的屏障（啟動次數與您想要輸入的參與者數量相同）
java SyncPrimitive bTest localhost 2
來源清單
SyncPrimitive.Java
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Random;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
public class SyncPrimitive implements Watcher {
static ZooKeeper zk = null;
static Integer mutex;
String root;
SyncPrimitive(String address) {
if(zk == null){
try {
System.out.println("Starting ZK:");
zk = new ZooKeeper(address, 3000, this);
mutex = new Integer(-1);
System.out.println("Finished starting ZK: " + zk);
} catch (IOException e) {
System.out.println(e.toString());
zk = null;
}
}
//else mutex = new Integer(-1);
}
synchronized public void process(WatchedEvent event) {
synchronized (mutex) {
//System.out.println("Process: " + event.getType());
mutex.notify();
}
}
/**
* Barrier
*/
static public class Barrier extends SyncPrimitive {
int size;
String name;
/**
* Barrier constructor
*
* @param address
* @param root
* @param size
*/
Barrier(String address, String root, int size) {
super(address);
this.root = root;
this.size = size;
// Create barrier node
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out
.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
// My node name
try {
name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());
} catch (UnknownHostException e) {
System.out.println(e.toString());
}
}
/**
* Join barrier
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean enter() throws KeeperException, InterruptedException{
zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() < size) {
mutex.wait();
} else {
return true;
}
}
}
}
/**
* Wait until all reach barrier
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean leave() throws KeeperException, InterruptedException{
zk.delete(root + "/" + name, 0);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() > 0) {
mutex.wait();
} else {
return true;
}
}
}
}
}
/**
* Producer-Consumer queue
*/
static public class Queue extends SyncPrimitive {
/**
* Constructor of producer-consumer queue
*
* @param address
* @param name
*/
Queue(String address, String name) {
super(address);
this.root = name;
// Create ZK node name
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out
.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
}
/**
* Add element to the queue.
*
* @param i
* @return
*/
boolean produce(int i) throws KeeperException, InterruptedException{
ByteBuffer b = ByteBuffer.allocate(4);
byte[] value;
// Add child with value i
b.putInt(i);
value = b.array();
zk.create(root + "/element", value, Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT_SEQUENTIAL);
return true;
}
/**
* Remove first element from the queue.
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
int consume() throws KeeperException, InterruptedException{
int retvalue = -1;
Stat stat = null;
// Get the first element available
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() == 0) {
System.out.println("Going to wait");
mutex.wait();
} else {
Integer min = new Integer(list.get(0).substring(7));
String minNode = list.get(0);
for(String s : list){
Integer tempValue = new Integer(s.substring(7));
//System.out.println("Temporary value: " + tempValue);
if(tempValue < min) {
min = tempValue;
minNode = s;
}
}
System.out.println("Temporary value: " + root + "/" + minNode);
byte[] b = zk.getData(root + "/" + minNode,
false, stat);
zk.delete(root + "/" + minNode, 0);
ByteBuffer buffer = ByteBuffer.wrap(b);
retvalue = buffer.getInt();
return retvalue;
}
}
}
}
}
public static void main(String args[]) {
if (args[0].equals("qTest"))
queueTest(args);
else
barrierTest(args);
}
public static void queueTest(String args[]) {
Queue q = new Queue(args[1], "/app1");
System.out.println("Input: " + args[1]);
int i;
Integer max = new Integer(args[2]);
if (args[3].equals("p")) {
System.out.println("Producer");
for (i = 0; i < max; i++)
try{
q.produce(10 + i);
} catch (KeeperException e){
} catch (InterruptedException e){
}
} else {
System.out.println("Consumer");
for (i = 0; i < max; i++) {
try{
int r = q.consume();
System.out.println("Item: " + r);
} catch (KeeperException e){
i--;
} catch (InterruptedException e){
}
}
}
}
public static void barrierTest(String args[]) {
Barrier b = new Barrier(args[1], "/b1", new Integer(args[2]));
try{
boolean flag = b.enter();
System.out.println("Entered barrier: " + args[2]);
if(!flag) System.out.println("Error when entering the barrier");
} catch (KeeperException e){
} catch (InterruptedException e){
}
// Generate random integer
Random rand = new Random();
int r = rand.nextInt(100);
// Loop for rand iterations
for (int i = 0; i < r; i++) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
}
try{
b.leave();
} catch (KeeperException e){
} catch (InterruptedException e){
}
System.out.println("Left barrier");
}
}