ZooKeeper屏障和队列的指南（五）
引言
在这个指南中，使用展现了使用ZooKeeper实现的屏障和生产-消费队列。咱们分别称这些类为Barrier和Queue。这些例子假定你至少有一个运行的ZooKeeper服务。java
两个原语都使用下面的代码片段:node
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对象的时候，而且咱们声明一个静态的变量引用这个对象。随后的Barrier对象和Queue检查一个ZooKeeper对象是否存在。另外，咱们能够有一个建立ZooKeeper的应用而且传递它给Barrier和Queue的构造函数。apache
咱们使用process()方法来处理监视器的通知。在下面的讨论，咱们呈现设置监视器的代码。一个监视器是一个内部的数据结构，它可以使ZooKeeper通知节点的改变。例如，若是一个客户端正在 等待其它客户端离开一个屏障，那么它能够给一个特定的节点设置一个监视器而且等待修改。这能表示它等待结束了。一旦你看完例子你就明白这一点。数据结构
屏障
一个屏障是一个原语，它使一组进程能够同步地开始和结束一个计算。这种实现的整体思想是有一个barrier节点做为每个进程节点的父节点。假如咱们这个屏障节点为"/b1"。每个进程"p"建立一个节点”/b1/p“。一旦有足够的进程已经建立的它们对象的节点，加入的进程能够开始计算。app
在这个例子中，每个进程表明一个Barrier对象，而且它的构造函数有这些参数:dom
- ZooKeeeper服务端的地址(例如"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()。一个进程在root下面建立一个节来表明它，用它的主机名来表示节点的名字。它而后等待直到足够的进程进入到屏障。一个进程经过用getChildren()检查root节点的孩子节点数量来实现。而且等待通知一旦它没有足够的孩子节点。为了收到一个通知当root节点改变的时候，一个进程不得不设置 一个监视器，而且经过调用"getChildren()"来实现。在代码中，咱们的"getChildren()"方法有两个参数。第一个是那一个节点来读数据的，而且第二个是布尔标识使进程设置 一个监视器，在代码中标识是true.oop
/**
* 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_SEQUENTIAL);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() < size) {
mutex.wait();
} else {
return true;
}
}
}
}
注意enter()抛出KeeperException 和InterruptedException异常，因此应用有责任来捕获和处理这样的异常。this
一旦计算结束，一个进程调用leave()来离开这个屏障。首先它删除它对应的节点，而后它获得root节点的孩子。若是至少有一个孩子，那么它就会等待一个通知(注意调用getChildred()的第二个参数是true)，意为着ZooKeeper不得不设置一个监视器在root节点上。一旦收到一个通知，它再次检查是否root节点有任何孩子。spa
/**
* 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;
}
}
}
}
}
生产者-消费者队列
一个生产者-消费者队列是一个数据分发结构，一组进程生产而且消费物品。生产者进程建立新的元素而且加入到队列。消费者进程从队列中删除元素，而且处理它们。在这个实现中，元素中简单的数字。队列被一个root节点表示，而且加入一个元素到这个队列，一个生产者进程建立一个新节点，root节点的孩子节点。
下面的代码片段对象对象的构造函数。就像Barrier对象同样，它首先父类的构造函数，SyncPrimitive，来建立一个ZooKeeper对象，若是这个对象不存在。它而后校验队列的root节点是否存在，而且若是不存在就建立一个。
/**
* 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增长序列计数器的值到root节点上。用这种方法，咱们暴露了队列中元素的总的顺序，所以保证队列中最老的元素是下一个要消费的元素。
/**
* 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;
}
为了消费一个元素，一个消费者进程获取root节点的孩子，读取最小值的节点，而且返回这个元素。注意若是有一个冲突，那么两个竞争进程中的一个将不能删除它的节点而且删除操做将会抛出异常。
调用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;
}
}
}
}
}
完整的源码列表
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_SEQUENTIAL);
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));
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;
}
}
}
}
}
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");
}
}
相关文章
- 1. RabbitMQ指南（三） 队列
- 2. zookeeper实现屏障_Barrier
- 3. zookeeper C API指南
- 4. 指令重排序和内存屏障
- 5. Laravel入坑指南(11)——列队
- 6. 小橙书阅读指南（七）——优先队列和索引优先队列
- 7. 剑指offer2.3.5栈和队列
- 8. 剑指offer——栈和队列
- 9. Zookeeper分布式锁和队列
- 10. 团队协作的五大障碍
- 更多相关文章...
- • C# 队列（Queue） - C#教程
- • SQL 指南 - 网站建设指南
- • 算法总结-双指针
- • 适用于PHP初学者的学习线路和建议
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. RabbitMQ指南（三） 队列
- 2. zookeeper实现屏障_Barrier
- 3. zookeeper C API指南
- 4. 指令重排序和内存屏障
- 5. Laravel入坑指南(11)——列队
- 6. 小橙书阅读指南（七）——优先队列和索引优先队列
- 7. 剑指offer2.3.5栈和队列
- 8. 剑指offer——栈和队列
- 9. Zookeeper分布式锁和队列
- 10. 团队协作的五大障碍