使用zookeeper如何模拟实现一个分布式队列

可以将程序设计语言可以分为两类：编译型语言和解释型语言。

介绍了基于zookeeper实现的分布式公平锁，以及非公平锁，那么本篇呢，散仙就来介绍下关于使用zookeeper如何模拟实现一个分布式队列。

那么为什么需要分布式队列呢？，我们都知道队列，在我们的编程开发中，是一种比不可少的数据结构，最典型莫过于，生产者与消费者的例子了，我们在程序过经常使用的队列是基于非分布式的环境，JAVA JDK也自带了非常多的队列的实现，有基于阻塞模式的，也有基于非阻塞模式的，这些我们可以在不同的场景下使用。

要想把JDK自带的队列给设计成分布式的队列，是一件非常繁琐的事，这时候我就可以轻而易举的使用zookeeper来完成这个功能，zookeeper由于其特殊的文件系统，所以在分布式环境下，能做许多有用的事。



例如在Hadoop中，map任务完成后，我们启动reduce，我们都知道Hadoop的MapReduce是一种分布式的计算框架，所以这时候我们就可以使用zookeeper来完成这里的分布式队列的运用，虽然，hadoop本身用的并不是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);  
}  
      
  
}  

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);
}
	

}

代码如上，散仙在代码里设定队列的总数为3，就执行干某一件事，我可以去zk的主节点上注册临时有序的节点，每注册一次都会校验，当前队列是不是以及满队，如果满队的话，就开始执行某个任务，从而达到分布式队列的模拟，这与JDK自带的CountDownLatch和CyclicBarrier是非常类似的，当然我们也可也使用zk模拟，这两个非分布式的功能，从而为我们的应用程序在分布式的环境下提供便利。