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, 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, 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 Stringname;
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
页:
[1]