zookeeper应用 - FIFO 队列 分布式队列

package fifo;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
/**
 * 使用ZooKeeper实现的FIFO队列
 * @author lisg
 *
 */
public class ZKFIFO {
	private static final String HOSTS = "vm1";
	private ZooKeeper zk = null;
	private static final String PARENT_PATH = "/fifo";
	private static final String SEQ_PREFIX = "seq-";
	
	public ZKFIFO() {
		try {
			final CountDownLatch cdl = new CountDownLatch(1);
			zk = new ZooKeeper(HOSTS, 5000, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					if(KeeperState.SyncConnected.equals(event.getState())) {
						cdl.countDown();
					}
				}
			});
			
			cdl.await();
			
			//创建父节点
			Stat stat = zk.exists(PARENT_PATH, false);
			if(stat == null) {
				zk.create(PARENT_PATH, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
		} catch (Exception e) {
			System.out.println("zookeeper集群连接失败！");
			e.printStackTrace();
		}
	}
	
	/**
	 * 在父节点下创建顺序子节点
	 * @param data
	 */
	public void push(String data) {
		if(data == null) {
			data = "";
		}
		
		try {
			zk.create(PARENT_PATH + "/" + SEQ_PREFIX, 
					data.getBytes("UTF-8"), 
					ZooDefs.Ids.OPEN_ACL_UNSAFE, 
					CreateMode.PERSISTENT_SEQUENTIAL);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 删除字第一个子节点，并返回它的值
	 * @return
	 */
	public String pop() {
		try {
			final List<String> children = zk.getChildren(PARENT_PATH, false);
			if(children.isEmpty()) {
				return null;
			}
			
			Collections.sort(children);
			
			String firstChildPath = PARENT_PATH + "/" + children.get(0);
			
			final byte[] data = zk.getData(firstChildPath, false, null);
			zk.delete(firstChildPath, -1);
			
			return new String(data, "UTF-8");
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		return null;
	}
	
	public void close() {
		try {
			this.zk.close();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public static void main(String[] args) {
		final ZKFIFO fifo = new ZKFIFO();
		
		/*
		for(int i=0; i<10; i++) {
			new Thread() {
				public void run() {
					fifo.push("data-" + UUID.randomUUID().toString().replace("-", ""));
				};
			}.start();
		}
		*/
		
		System.out.println(fifo.pop());
		
		fifo.close();
	}
}