zookeeper学习(七)

色破飞机

　　       最近看到一个利用ZK的Watch机制实现Barrier的例子，因为Watch是一个很典型的类似观察者模式的机制，程序中很巧妙的使用一个Integer做为互斥量(mutex)。触发watch的process的时候，notifyAll。开始看的时候有一点晕，之后想了想恍然大悟,既然所有的类都继承自Object类，那么当然所有的类都会继承Object的wait,notify和notifyAll方法了。(基本类型如int，float等是不继承自Object的;但是数组是继承自Object的)。       
　　       实现Barrier的例子中用执行完enter或者leave之后，调用Integer型变量mutex的wait方法阻塞本线程，等待被唤醒。在watch的process中调用notifyAll，即一旦发布者有所变化会触发process方法。

　　         先把程序贴出来：

    public class Barrier implements Watcher {  
private static final String addr = "10.20.156.49:2181";  
private ZooKeeper           zk   = null;  
private Integer             mutex;  
private int                 size = 0;  
private String              root;  
public Barrier(String root, int size){  
this.root = root;  
this.size = size;  
try {  
zk = new ZooKeeper(addr, 10 * 1000, this);  
mutex = new Integer(-1);  
Stat s = zk.exists(root, false);  
if (s == null) {  
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);  
}  
} catch (Exception e) {  
e.printStackTrace();  
}  
}  
public synchronized void process(WatchedEvent event) {  
synchronized (mutex) {  
mutex.notify();  
}  
}  
public boolean enter(String name) throws Exception {  
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;  
}  
}  
}  
}  
public boolean leave(String name) 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;  
}  
}  
}  
}  
}   
　　测试代码；

    public class BarrierTest {  
public static void main(String args[]) throws Exception {  
for (int i = 0; i < 3; i++) {  
Process p = new Process("Thread-" + i, new Barrier("/test/barrier", 3));  
p.start();  
}  
}  
}  
class Process extends Thread {  
private String  name;  
private Barrier barrier;  
public Process(String name, Barrier barrier){  
this.name = name;  
this.barrier = barrier;  
}  
@Override  
public void run() {  
try {  
barrier.enter(name);  
System.out.println(name + " enter");  
Thread.sleep(1000 + new Random().nextInt(2000));  
barrier.leave(name);  
System.out.println(name + " leave");  
} catch (Exception e) {  
e.printStackTrace();  
}  
}  
}  
　　 
　　*****************************以上是实例程序******************************
　　接下来，我自己试了一下Integer的wait方法，却发生了运行时错误：
　　我的程序是这样的：

       public static void main(String args[]){
Integer mutex=new Integer(1);
try {
mutex.wait(5000);
System.out.println("OK");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
//运行结果：
Exception in thread "main" java.lang.IllegalMonitorStateException
at java.lang.Object.wait(Native Method)
　　 于是我查了一下原因；
　　违法的监控状态异常。当某个线程试图等待一个自己并不拥有的对象（O）的监控器或者通知其他线程等待该对象（O）的监控器时，抛出该异常。    原因是：在对某个对象上调用wait()方法进行线程等待（让其他竞争执行该代码的线程上锁）时，没有对该对象执行同步操作。
　　所以，给mutex.wait同步就行了：

public static void main(String args[]){
Integer mutex=new Integer(1);
try {
synchronized (mutex) {
mutex.wait(5000);
}
System.out.println("OK");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
//运行结果(5秒钟之后输出)；
OK