zookeeper实现fifo以及并发访问删除

package cn.sniper.zookeeper;
import java.io.IOException;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;
public class ZookeeperUtil {
 
 @Test
 public void helloword() {
  //端口默认是2181
  //String connectString = "192.168.1.231";
  String connectString = "192.168.1.231:2181";
  int sessionTimeout = 20000;
  
  try {
   ZooKeeper zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
    public void process(WatchedEvent event) {
     System.out.println(event);
    }
   });
   
   System.out.println(zk);
   zk.close();
  } catch (IOException e) {
   e.printStackTrace();
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
 }
 
 @Test
 public void create() {
  String connectString = "192.168.1.231:2181";
  int sessionTimeout = 20000;
  
  try {
   ZooKeeper zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
    public void process(WatchedEvent event) {
     System.err.println("事件类型：" + event.getType());
    }
   });
   
   //创建节点 
   zk.create("/sniper1", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
   
   zk.close();
  } catch (IOException e) {
   e.printStackTrace();
  } catch (InterruptedException e) {
   e.printStackTrace();
  } catch (KeeperException e) {
   e.printStackTrace();
  }
 }
 
 @Test
 public void fifoIn() {
  String connectString = "192.168.1.231,192.168.1.232,192.168.1.233";
  int sessionTimeout = 50000;
  
  try {
   ZooKeeper zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
    public void process(WatchedEvent event) {
     System.err.println("事件类型：" + event.getType());
    }
   });
   
   //每个客户端连进来的时候，都在fifo下创建一个有序节点  模拟10个客户端连接进入
   for(int i=0; i<10; i++) {
    zk.create("/fifo/", String.valueOf(System.currentTimeMillis()).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
   }
   
   zk.close();
  } catch (IOException e) {
   e.printStackTrace();
  } catch (InterruptedException e) {
   e.printStackTrace();
  } catch (KeeperException e) {
   e.printStackTrace();
  }
 }
 
 @Test
 public void fifoOut() {
  String connectString = "192.168.1.231,192.168.1.232,192.168.1.233";
  int sessionTimeout = 30000;
  
  try {
   ZooKeeper zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
    public void process(WatchedEvent event) {
     System.err.println("事件类型：" + event.getType());
    }
   });
   
   List<String> children = zk.getChildren("/fifo", new Watcher() {
    public void process(WatchedEvent event) {
     System.err.println("事件类型：" + event.getType());
    }
   });
   
   //由于节点的有序性，将节点用treeSet排序一下，取得第一个元素，就可以做到先进先出队列了
   TreeSet<String> set = new TreeSet<String>(children);
   
   String child = set.first();
   
   System.err.println(child);
   
   zk.delete("/fifo/"+child, -1);
   
   zk.close();
  } catch (IOException e) {
   e.printStackTrace();
  } catch (InterruptedException e) {
   e.printStackTrace();
  } catch (KeeperException e) {
   e.printStackTrace();
  }
 }
 
 /**
  * 多线程创建有序节点不存在问题
  * @throws Exception
  */
 public static void fifoInMultiThread() throws Exception {
  //int nThreads = Runtime.getRuntime().availableProcessors();
  
  int nThreads = 20;
  final ExecutorService service = Executors.newFixedThreadPool(nThreads);
  
  final ZooKeeper zk = new ZooKeeper("192.168.1.231,192.168.1.232,192.168.1.233", 30000, new Watcher() {
   public void process(WatchedEvent evet) {
    
   }
  });
  
  final long begin = System.currentTimeMillis();
  
  for(int i=0; i<1000; i++) {
   final int j = i;
   service.execute(new Runnable() {
    public void run() {
     try {
      zk.create("/fifo/", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
     } catch (KeeperException e) {
      e.printStackTrace();
     } catch (InterruptedException e) {
      e.printStackTrace();
     } finally {
      if(j == 999) {
       try {
        zk.close();
        service.shutdown();
        System.err.println(System.currentTimeMillis() - begin);
       } catch (InterruptedException e) {
        e.printStackTrace();
       }
      }
     }
    }
   });
  }
 }
 
 @Test
 public void size() throws Exception {
  final ZooKeeper zk = new ZooKeeper("192.168.1.231,192.168.1.232,192.168.1.233", 50000, new Watcher() {
   public void process(WatchedEvent evet) {
    
   }
  });
  
  System.out.println(zk.getChildren("/fifo", new Watcher() {
   public void process(WatchedEvent evet) {
    
   }
  }).size() + " =========================");
  
  zk.close();
 }
 
 /**
  * 多线程取出节点并且删除节点存在问题，synchronized解决方案
  * @throws Exception
  */
 public static void fifoOutMultiThread() throws Exception {
  int nThreads = 20;
  final ExecutorService service = Executors.newFixedThreadPool(nThreads);
  
  final ZooKeeper zk = new ZooKeeper("192.168.1.231,192.168.1.232,192.168.1.233", 50000, new Watcher() {
   public void process(WatchedEvent evet) {
    
   }
  });
  
  final long begin = System.currentTimeMillis();
  
  for(int i=0; i<1000; i++) {
   final int j = i;
   service.execute(new Runnable() {
    public void run() {
     try {
      synchronized (ZookeeperUtil.class) {
       List<String> children = zk.getChildren("/fifo", new Watcher() {
        public void process(WatchedEvent arg0) {
         
        }
       });
       
       if(children != null && children.size() > 0) {
        TreeSet<String> set = new TreeSet<String>(children);
        String pop = set.first();
        
        System.err.println(pop);
        
        zk.delete("/fifo/"+pop, -1);
       }
      }
     } catch (KeeperException e) {
      e.printStackTrace();
     } catch (InterruptedException e) {
      e.printStackTrace();
     } finally {
      if(j == 999) {
       try {
        zk.close();
        service.shutdown();
        System.err.println(System.currentTimeMillis() - begin);
       } catch (InterruptedException e) {
        e.printStackTrace();
       }
      }
     }
    }
   });
  }
 }
 
 /**
  * 多线程取出节点并且删除节点存在问题，读写锁方式
  * @throws Exception
  */
 public static void fifoOutMultiThreadLock() throws Exception {
  final ReadWriteLock rwl = new ReentrantReadWriteLock();
  
  int nThreads = 20;
  final ExecutorService service = Executors.newFixedThreadPool(nThreads);
  
  final ZooKeeper zk = new ZooKeeper("192.168.1.231,192.168.1.232,192.168.1.233", 50000, new Watcher() {
   public void process(WatchedEvent evet) {
    
   }
  });
  
  final long begin = System.currentTimeMillis();
  
  for(int i=0; i<1000; i++) {
   final int j = i;
   service.execute(new Runnable() {
    public void run() {
     try {
      rwl.readLock().lock();
       //取得fifo下的所有直接下级节点
       List<String> children = zk.getChildren("/fifo", new Watcher() {
        public void process(WatchedEvent event) {
         
        }
       });
       
       //用treeset排好序
       TreeSet<String> set = new TreeSet<String>(children);
       //取出第一个，即最先进入的一个
       String pop = set.first();
       //判断是否存在
       Stat stat = zk.exists("/fifo/"+pop, new Watcher() {
        public void process(WatchedEvent event) {
         
        }
       });
      
      if(stat != null) {
       rwl.readLock().unlock();
       
       rwl.writeLock().lock();
        //用于多线程读的因素，有可能第一条线程读到节点还存在，紧接着第二条线程就把该节点删除了，所以，加了写锁之后还需要判断一次节点是否存在
        stat = zk.exists("/fifo/"+pop, new Watcher() {
         public void process(WatchedEvent event) {
          
         }
        });
        
        if(stat != null) {
         System.err.println(pop);
         zk.delete("/fifo/"+pop, -1);
        }
       rwl.writeLock().unlock();
       rwl.readLock().lock();
      }
      rwl.readLock().unlock();
     } catch (KeeperException e) {
      e.printStackTrace();
     } catch (InterruptedException e) {
      e.printStackTrace();
     } finally {
      if(j == 999) {
       try {
        zk.close();
        service.shutdown();
        System.err.println(System.currentTimeMillis() - begin);
       } catch (InterruptedException e) {
        e.printStackTrace();
       }
      }
     }
    }
   });
  }
 }
 
 public static void main(String[] args) throws Exception {
  //fifoInMultiThread();
  //fifoOutMultiThread();
  //fifoOutMultiThreadLock();
 }
 
}