使用 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");
}
}