zookeeperテスト
詳細
zookeeperベースのbarrierをコードで実現した.
1.zookeeperをダウンロードします.
2.zk,binzkServerの起動
3.javaコード.
Barrier.java
BarrierTest.java
zookeeperベースのbarrierをコードで実現した.
1.zookeeperをダウンロードします.
2.zk,binzkServerの起動
3.javaコード.
Barrier.java
package com.chenguan.barrier;
import java.io.IOException;
import java.net.InetAddress;
import java.util.List;
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 Barrier implements Watcher {
int size;
String root;
private ZooKeeper zk = null;
private Integer mutex;
Barrier(String address, String name, int size) throws KeeperException,
IOException, InterruptedException {
zk = new ZooKeeper(address, 10 * 1000, this);
mutex = new Integer(-1);
// TODO Auto-generated constructor stub
this.root = name;
this.size = size;
// Create barrier node
if (zk != null) {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
}
// My node name
name = new String(InetAddress.getLocalHost().getCanonicalHostName()
.toString());
}
/**
******* Join barrier
*******
******* @return
******* @throws KeeperException
******* @throws InterruptedException
*******/
boolean enter(String name) throws KeeperException, InterruptedException {
zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
System.out.println(name+" created one node...");
while (true) {
synchronized (mutex) {
List list = zk.getChildren(root, true);
if (list.size() < size) {
System.out.println(name+" wait...");
mutex.wait();
} else {
System.out.println(name+" notified...");
return true;
}
}
}
}
/**
* Wait until all reach barrier
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean leave(String name) throws KeeperException, InterruptedException {
zk.delete(root + "/" + name, 0);
System.out.println(name+" deleted one node...");
while (true) {
synchronized (mutex) {
List list = zk.getChildren(root, true);
if (list.size() > 0) {
System.out.println(name+" wait...");
mutex.wait();
} else {
System.out.println(name+" notified...");
return true;
}
}
}
}
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
synchronized (mutex) {
System.err.println("event:"+event.getPath());
mutex.notifyAll();
}
}
}
BarrierTest.java
package com.chenguan.barrier;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class BarrierTest {
/**
* , zk
*
* @param args
* @throws Exception
*/
public static void main(String args[]) throws Exception {
// for (int i = 0; i < 3; i++) {
// Process p = new Process("Thread-" + i, barrier);
// p.start();
// }
ExecutorService service = Executors.newFixedThreadPool(3);
Future future1 = service.submit(new Process("Thread-" + 1,
new Barrier("127.0.0.1:2181", "/test1", 3)));
Future future2 = service.submit(new Process("Thread-" + 2,
new Barrier("127.0.0.1:2181", "/test1", 3)));
Future future3 = service.submit(new Process("Thread-" + 3,
new Barrier("127.0.0.1:2181", "/test1", 3)));
List> tasks = new ArrayList>();
tasks.add(future1);
tasks.add(future2);
tasks.add(future3);
for (Future f : tasks) {
String result = f.get();
System.out.println(result);
}
service.shutdown();
}
}
/**
* callable
* @author Administrator
*
*/
class Process implements Callable {
private String name;
private Barrier barrier;
public Process(String name, Barrier barrier) {
this.name = name;
this.barrier = barrier;
}
@Override
public String call() throws Exception {
try {
barrier.enter(name);
System.out.println(name + " enter");
Thread.sleep(1000 + new Random().nextInt(2000));
boolean result = barrier.leave(name);
if (result) {
System.out.println(name + " leave");
}
return name + " finished";
} catch (Exception e) {
e.printStackTrace();
return " ";
}
}
}
/**
* Runnable
* @author Administrator
*
*/
class ProcessRunnable implements Runnable {
private String name;
private Barrier barrier;
public ProcessRunnable(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));
boolean result = barrier.leave(name);
if (result) {
System.out.println(name + " leave");
}
} catch (Exception e) {
e.printStackTrace();
}
}
}