- Barriers解决什么问题?
当有些操作需要所有参与者全部准备好之后才能开始执行，并且对每个参与者来说必须等待所有参与者全部执行完毕，才算执行完毕。于是就需要一个屏障，来控制所有参与者同时开始，并等待所有参与者全部结束。
- 实现barriers需要注意的问题
1.如何控制所有线程同时开始？
- 参与者准备好时就在/barrier父节点下插入顺序临时节点。
- 调用getChidren方法，并设置watch,检查/barrier下所有children节点的数量是否等于所有参与者的数量。
- 如果等于，则说明所有参与者都已到达，本参与者开始执行自己的业务代码。
- 如果不是，则此参与者调用wait()方法阻塞。
- 当有别的参与者插入新的子节点时这个参与者会收到通知，调用notify唤醒线程，第二步重新开始执行。
2.如何等待所有线程结束？
- 参与者执行完毕后，调用delete方法删除自己的子节点
- 获取父节点下的子节点数量n，并设置watch
- 如果n=0,说明所有参与者都已执行完毕，本参与者是最后一个，则可以直接结束。
- 如果n>0,则本参与者调用wait方法阻塞
- 如果别的参与者执行完毕，删除了一个子节点，则本参与者会收到通知，调用notify唤醒线程，第二步重新开始执行。
3.用什么类型的节点？
根节点使用持久节点(persistent node),子节点使用临时节点(Ephemeral node)
根节点为什么要用持久节点？首先因为临时节点不能有子节点，所以根节点要用持久节点，并且在程序中要判断根节点是否存在。
子节点为什么要用临时节点？临时节点随着连接的断开而消失，在程序中，虽然会删除临时节点，但可能会出现程序在节点被删除之前就crash了，如果是持久节点，节点不会被删除。
- 源码
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;//这个类是barrier的父类，负责初始化zkclient,并且实现了Watcher接口，负责接收zookeeper通知
public class SyncPrimitive implements Watcher {static ZooKeeper zk = null;//zk client对象static Integer mutex; //这是barrier用到的锁String root;//这是barrieer父节点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);}//接收zookeeper通知，唤醒barrier中的enter()和leave()方法中的等待线程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;//本参与者对应的子节点path/*** 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 nodeif (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 nametry {name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());} catch (UnknownHostException e) {System.out.println(e.toString());}}/*** Join barrier 参与者执行enter进入屏障** @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 参与者执行leave()离开屏障** @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;}}}}}