ZooKeeperを使ったプログラミング - 基本的なチュートリアル
はじめに
このチュートリアルでは、ZooKeeperを使ったバリアとプロデューサー/コンシューマーキューの簡単な実装を示します。それぞれをBarrierクラスとQueueクラスと呼びます。これらの例では、少なくとも1つの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のコンストラクターに渡すこともできます。
プロセス()メソッドを使用して、ウォッチによってトリガーされた通知を処理します。次の議論では、ウォッチを設定するコードを示します。ウォッチは、ZooKeeperがノードへの変更をクライアントに通知できるようにする内部構造です。たとえば、クライアントが他のクライアントがバリアを離れるのを待っている場合、ウォッチを設定して、待機が終了したことを示す特定のノードへの変更を待つことができます。この点は、例を詳しく見ていくと明確になります。
バリア
バリアは、プロセスのグループが計算の開始と終了を同期できるようにするプリミティブです。この実装の一般的な考え方は、個々のプロセスノードの親となるバリアノードを持つことです。バリアノードを「/b1」と呼ぶとします。各プロセス「p」は、「/b1/p」というノードを作成します。十分な数のプロセスが対応するノードを作成すると、結合されたプロセスは計算を開始できます。
この例では、各プロセスはBarrierオブジェクトをインスタンス化し、そのコンストラクターはパラメータとして以下を受け取ります。
- ZooKeeperサーバーのアドレス（例：「zoo1.foo.com:2181」）
- ZooKeeper上のバリアノードのパス（例：「/b1」）
- プロセスのグループサイズ
Barrierのコンストラクターは、Zookeeperサーバーのアドレスを親クラスのコンストラクターに渡します。親クラスは、ZooKeeperインスタンスが存在しない場合に作成します。次に、BarrierのコンストラクターはZooKeeperにバリアノードを作成します。これは、すべてのプロセスノードの親ノードであり、root（注: これは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()」に2つのパラメーターがあります。1つ目は読み取るノードを指定し、2つ目はプロセスがウォッチを設定できるようにするブールフラグです。コードでは、フラグは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()を呼び出してバリアを離れます。まず、対応するノードを削除し、次にルートノードの子を取得します。少なくとも1つの子がある場合は、通知を待ちます（注意：getChildren()の呼び出しの2番目のパラメーターは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;
}
要素を消費するために、コンシューマープロセスはルートノードの子を取得し、最小のカウンター値を持つノードを読み取り、要素を返します。競合がある場合、競合する2つのプロセスのいずれかがノードを削除できなくなり、削除操作で例外がスローされることに注意してください。
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");
}
}