分布式助手Zookeeper（八） - 三劫散仙

package com.qin.queue;

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.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;

import com.test.Lock1;

/**
 * 
 * 基于zookeeper实现的
 * 分布式队列
 * @author qindongliang
 * 
 * **/
public class DistributeQueue1 {
	
	
 	 private ZooKeeper zk;
 	 private CountDownLatch down=new CountDownLatch(1);
 	 
 	 
 	 public DistributeQueue1(String zkHost) {
		try{
 		 zk=new ZooKeeper(zkHost, 5000, new Watcher(){

			@Override
			public void process(WatchedEvent event) {
				
				if(event.getState()==Event.KeeperState.SyncConnected){
					down.countDown();
				}
				
			}
 			 
 			 
 		 });
 		 
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	
	
 	 
 	 /**
	  * 创建一个持久node，
	  * 
	  * **/
	 public void createPersist()throws Exception{
		 
		 zk.create("/a", "主节点".getBytes(), Ids.OPEN_ACL_UNSAFE	, CreateMode.PERSISTENT);
		 
		 System.out.println("创建主节点成功........");
		 
		 
	 }
	 
	 /**
		 * 
		 * 校验队列情况
		 * 
		 * **/
		 public void check()throws Exception{
			 List<String> list=zk.getChildren("/a", null);
			 
			 if(list.size()==3){
				 execute();
			 }else{
				 System.out.println("阻塞中......,队列成员数量：" list.size());
			 }
			 
			 
			  
				  
				  
			  }
			 
		 
	 
	 
  public void execute(){
	  
	  System.out.println("队列已满，开始执行任务......");
	  
  }

  /***
	  * 
	  * 注册队列
	  * 
	  * 
	  * */
	 public void createTemp()throws Exception{
		 
		 zk.create("/a/b", "a".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		 System.out.println("有任务注册队列成功");
	    check(); 
		 
	 }
  
  public static void main(String[] args)throws Exception {
		
	 DistributeQueue1 dq=new DistributeQueue1("192.168.120.128:2181");
	 //dq.createPersist();
	 dq.createTemp();
	 Thread.sleep(Integer.MAX_VALUE);
}
	

}