zookeeper实现Barrier

用代码实现了一个基于zookeeper的barrier。 

1.先下载zookeeper。 

2.启动zk,bin\zkServer 

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(); } } }