/*
 * Copyright 2002-2016 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.zuoxiaolong.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;

/**
 * 这是一个用zookeeper实现queue的示例.
 *
 * @author Xiaolong Zuo
 * @since 16/1/5 01:30
 *
 * @see com.zuoxiaolong.zookeeper.QueueProducerTest
 * @see com.zuoxiaolong.zookeeper.QueueConsumerTest
 */
public class Queue extends SyncPrimitive {

    //初始化时建立一个序列的根节点,name可以看作是queue的名称.
    public Queue(String address, String name)
            throws KeeperException, InterruptedException, IOException {
        super(address);
        this.root = name;
        if (zk != null) {
            Stat s = zk.exists(root, false);
            if (s == null) {
                zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }
    }

    //生产者调用的方法
    boolean produce(int i) throws KeeperException, InterruptedException {
        ByteBuffer b = ByteBuffer.allocate(4);
        byte[] value;

        // 将消息的值设置为i
        b.putInt(i);
        value = b.array();
        //在根节点下建立一个节点,相当于在queue当中插入一个消息
        String result = zk.create(root + "/element", value, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT_SEQUENTIAL);
        System.out.println("result:" + result);
        return true;
    }

    int consume() throws KeeperException, InterruptedException {
        Stat stat = null;

        // Get the first element available
        while (true) {
            synchronized (mutex) {
                List<String> list = zk.getChildren(root, true);
                System.out.println("size:" + list.size());
                if (list.isEmpty()) {
                    System.out.println("Going to wait");
                    mutex.wait();
                } else {
                    Integer min = new Integer(list.get(0).substring(7));
                    String id = list.get(0);
                    for (String s : list) {
                        System.out.println("queue's elements : " + s);
                        Integer tempValue = new Integer(s.substring(7));
                        if (tempValue < min) id = s;
                    }
                    byte[] b = zk.getData(root + "/" + id, false, stat);
                    zk.delete(root + "/" + id, 0);
                    ByteBuffer buffer = ByteBuffer.wrap(b);
                    int value = buffer.getInt();
                    System.out.println("element has been consumed: " + root + "/" + id + "  the value is:" + value);
                }
            }
        }
    }

}

@@ -2067,40 +2067,16 @@
   //
- Get the first element available
+%E6%97%A0%E9%99%90%E5%BE%AA%E7%8E%AF%E8%8E%B7%E5%8F%96%E6%B6%88%E6%81%AF
 %0A   
@@ -2126,24 +2126,77 @@
 d (mutex) %7B%0A
+                //%E8%8E%B7%E5%8F%96%E6%A0%B9%E8%8A%82%E7%82%B9%E7%9A%84%E5%AD%90%E8%8A%82%E7%82%B9,%E4%B9%9F%E5%B0%B1%E6%98%AF%E6%B6%88%E6%81%AF,%E5%B9%B6%E4%B8%94%E7%9B%91%E5%90%ACChildChanged%E4%BA%8B%E4%BB%B6%0A
             
@@ -2401,16 +2401,60 @@
 wait%22);%0A
+                    //%E5%A6%82%E6%9E%9C%E5%BD%93%E5%89%8D%E6%B2%A1%E6%9C%89%E6%B6%88%E6%81%AF,%E5%88%99%E5%9C%A8mutex%E5%AF%B9%E8%B1%A1%E4%B8%8A%E7%AD%89%E5%BE%85%0A
         
@@ -2504,16 +2504,43 @@
  else %7B%0A
+                    //%E6%88%AA%E5%8F%96%E6%95%B0%E5%AD%97%0A
         
@@ -2829,32 +2829,79 @@
 .substring(7));%0A
+                        //%E9%80%89%E5%8F%96%E6%9C%80%E5%B0%8F%E7%9A%84%E8%BF%9B%E8%A1%8C%E5%A4%84%E7%90%86,%E4%B9%9F%E5%B0%B1%E6%98%AF%E6%9C%80%E6%97%A9%E6%8F%92%E5%85%A5%E7%9A%84%E6%B6%88%E6%81%AF%0A
                 
@@ -2951,32 +2951,59 @@
               %7D%0A
+                    //%E8%8E%B7%E5%8F%96%E6%B6%88%E6%81%AF%0A
                 
@@ -3055,24 +3055,51 @@
 lse, stat);%0A
+                    //%E5%88%A0%E9%99%A4%E6%B6%88%E6%81%AF%0A
             
@@ -3197,16 +3197,16 @@
 rap(b);%0A
-
         
@@ -3242,24 +3242,63 @@
 r.getInt();%0A
+                    //%E6%B6%88%E8%B4%B9%E6%8E%89%E6%B6%88%E6%81%AF,%E6%9C%AC%E7%A4%BA%E4%BE%8B%E5%B0%B1%E6%98%AF%E7%AE%80%E5%8D%95%E7%9A%84%E6%89%93%E5%8D%B0%0A
             

添加zookeeper的例子.