使用ZooKeeper编程 - 基础教程
简介
在本教程中，我们将展示如何使用ZooKeeper简单实现屏障(barrier)和生产者-消费者队列。我们分别将这些类命名为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());
}
}
To enter the barrier, a process calls enter(). The process creates a node under the root to represent it, using its host name to form the node name. It then wait until enough processes have entered the barrier. A process does it by checking the number of children the root node has with "getChildren()", and waiting for notifications in the case it does not have enough. To receive a notification when there is a change to the root node, a process has to set a watch, and does it through the call to "getChildren()". In the code, we have that "getChildren()" has two parameters. The first one states the node to read from, and the second is a boolean flag that enables the process to set a watch. In the code the flag is 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");
}
}