分布式助手Zookeeper（八）

港控/mmm° 2022-04-21 15:54 195阅读 0赞

\[b\]\[color=olive\]\[size=large\]散仙，在前几篇关于zookeeper的文章中，介绍了基于zookeeper实现的分布式公平锁，以及非公平锁，那么本篇呢，散仙就来介绍下关于使用zookeeper如何模拟实现一个分布式队列。  
  
那么为什么需要分布式队列呢？，我们都知道队列，在我们的编程开发中，是一种比不可少的数据结构，最典型莫过于，生产者与消费者的例子了，我们在程序过经常使用的队列是基于非分布式的环境，JAVA JDK也自带了非常多的队列的实现，有基于阻塞模式的，也有基于非阻塞模式的，这些我们可以在不同的场景下使用。  
  
要想把JDK自带的队列给设计成分布式的队列，是一件非常繁琐的事，这时候我就可以轻而易举的使用zookeeper来完成这个功能，zookeeper由于其特殊的文件系统，所以在分布式环境下，能做许多有用的事。  
  
\[/size\]\[/color\]\[/b\]  
  
\[b\]\[size=large\]\[color=green\]例如在Hadoop中，map任务完成后，我们启动reduce，我们都知道Hadoop的MapReduce是一种分布式的计算框架，所以这时候我们就可以使用zookeeper来完成这里的分布式队列的运用，虽然，hadoop本身用的并不是zookeeper来实现的。  
  
代码如下：  
\[/color\]\[/size\]\[/b\]

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

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