分布式任务&分布式锁（li）

//创建一个与服务器的连接
ZooKeeper zk = new ZooKeeper("localhost:" + CLIENT_PORT, 
     ClientBase.CONNECTION_TIMEOUT, new Watcher() { 
         // 监控所有被触发的事件
         public void process(WatchedEvent event) { 
             System.out.println("已经触发了" + event.getType() + "事件！"); 
         } 
     }); 
// 创建一个目录节点
zk.create("/testRootPath", "testRootData".getBytes(), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT); 
// 创建一个子目录节点
zk.create("/testRootPath/testChildPathOne", "testChildDataOne".getBytes(),
Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); 
System.out.println(new String(zk.getData("/testRootPath",false,null))); 
// 取出子目录节点列表
System.out.println(zk.getChildren("/testRootPath",true)); 
// 修改子目录节点数据
zk.setData("/testRootPath/testChildPathOne","modifyChildDataOne".getBytes(),-1); 
System.out.println("目录节点状态：["+zk.exists("/testRootPath",true)+"]"); 
// 创建另外一个子目录节点
zk.create("/testRootPath/testChildPathTwo", "testChildDataTwo".getBytes(), 
Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); 
System.out.println(new String(zk.getData("/testRootPath/testChildPathTwo",true,null))); 
// 删除子目录节点
zk.delete("/testRootPath/testChildPathTwo",-1); 
zk.delete("/testRootPath/testChildPathOne",-1); 
// 删除父目录节点
zk.delete("/testRootPath",-1); 
// 关闭连接
zk.close(); 

// #输出的结果如下：
// 已经触发了 None 事件！
// testRootData
// [testChildPathOne]
// 目录节点状态：[5,5,1281804532336,1281804532336,0,1,0,0,12,1,6]
// 已经触发了 NodeChildrenChanged 事件！
// testChildDataTwo
// 已经触发了 NodeDeleted 事件！
// 已经触发了 NodeDeleted 事件！

 void findLeader() throws InterruptedException { 
        byte[] leader = null; 
        try { 
            leader = zk.getData(root + "/leader", true, null); 
        } catch (Exception e) { 
            logger.error(e); 
        } 
        if (leader != null) { 
            following(); 
        } else { 
            String newLeader = null; 
            try { 
                byte[] localhost = InetAddress.getLocalHost().getAddress(); 
                newLeader = zk.create(root + "/leader", localhost, 
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); 
            } catch (Exception e) { 
                logger.error(e); 
            } 
            if (newLeader != null) { 
                leading(); 
            } else { 
                mutex.wait(); 
            } 
        } 
    } 

 void getLock() throws KeeperException, InterruptedException{ 
        List<String> list = zk.getChildren(root, false); 
        String[] nodes = list.toArray(new String[list.size()]); 
        Arrays.sort(nodes); 
        if(myZnode.equals(root+"/"+nodes[0])){ 
            doAction(); 
        } 
        else{ 
            waitForLock(nodes[0]); 
        } 
    } 
    void waitForLock(String lower) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(root + "/" + lower,true); 
        if(stat != null){ 
            mutex.wait(); 
        } 
        else{ 
            getLock(); 
        } 
    }

//#加锁：
//ZooKeeper 将按照如下方式实现加锁的操作：
//1 ） ZooKeeper 调用 create （）方法来创建一个路径格式为“ _locknode_/lock- ”的节点，此节点类型为sequence （连续）和 ephemeral （临时）。
　　　　也就是说，创建的节点为临时节点，并且所有的节点连续编号，即“ lock-i ”的格式。
//2 ）在创建的锁节点上调用 getChildren （）方法，来获取锁目录下的最小编号节点，并且不设置 watch 。
//3 ）步骤 2 中获取的节点恰好是步骤 1 中客户端创建的节点，那么此客户端获得此种类型的锁，然后退出操作。
//4 ）客户端在锁目录上调用 exists （）方法，并且设置 watch 来监视锁目录下比自己小一个的连续临时节点的状态。
//5 ）如果监视节点状态发生变化，则跳转到第 2 步，继续进行后续的操作，直到退出锁竞争。
// 
//#解锁： 
//ZooKeeper 解锁操作非常简单，客户端只需要将加锁操作步骤 1 中创建的临时节点删除即可。

 void addQueue() throws KeeperException, InterruptedException{ 
        zk.exists(root + "/start",true); 
        zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE, 
        CreateMode.EPHEMERAL_SEQUENTIAL); 
        synchronized (mutex) { 
            List<String> list = zk.getChildren(root, false); 
            if (list.size() < size) { 
                mutex.wait(); 
            } else { 
                zk.create(root + "/start", new byte[0], Ids.OPEN_ACL_UNSAFE,
                 CreateMode.PERSISTENT); 
            } 
        } 
 }

 public void process(WatchedEvent event) { 
        if(event.getPath().equals(root + "/start") &&
         event.getType() == Event.EventType.NodeCreated){ 
            System.out.println("得到通知"); 
            super.process(event); 
            doAction(); 
        } 
    } 

#生产者代码
boolean produce(int i) throws KeeperException, InterruptedException{ 
        ByteBuffer b = ByteBuffer.allocate(4); 
        byte[] value; 
        b.putInt(i); 
        value = b.array(); 
        zk.create(root + "/element", value, ZooDefs.Ids.OPEN_ACL_UNSAFE, 
                    CreateMode.PERSISTENT_SEQUENTIAL); 
        return true; 
    } 

 
#消费者代码
int consume() throws KeeperException, InterruptedException{ 
        int retvalue = -1; 
        Stat stat = null; 
        while (true) { 
            synchronized (mutex) { 
                List<String> list = zk.getChildren(root, true); 
                if (list.size() == 0) { 
                    mutex.wait(); 
                } else { 
                    Integer min = new Integer(list.get(0).substring(7)); 
                    for(String s : list){ 
                        Integer tempValue = new Integer(s.substring(7)); 
                        if(tempValue < min) min = tempValue; 
                    } 
                    byte[] b = zk.getData(root + "/element" + min,false, stat); 
                    zk.delete(root + "/element" + min, 0); 
                    ByteBuffer buffer = ByteBuffer.wrap(b); 
                    retvalue = buffer.getInt(); 
                    return retvalue; 
                } 
            } 
        } 
 }