zookeeper实现分布式锁

public interface DistributedLock {
 
   /**获取锁，如果没有得到就等待*/
   public void acquire() throws Exception;
 
   /**
   * 获取锁，直到超时
   * @param time超时时间
   * @param unit time参数的单位
   * @return是否获取到锁
   * @throws Exception
   */
   public boolean acquire (long time, TimeUnit unit) throws Exception;
 
   /**
    * 释放锁
    * @throws Exception
    */
   public void release() throws Exception;
}

/**锁接口的具体实现，主要借助于继承的父类BaseDistributedLock来实现的接口方法
 * 该父类是基于Zookeeper实现分布式锁的具体细节实现*/
public class SimpleDistributedLockMutex extends BaseDistributedLock implements DistributedLock {
  /*用于保存Zookeeper中实现分布式锁的节点，如名称为locker：/locker，
 *该节点应该是持久节点，在该节点下面创建临时顺序节点来实现分布式锁 */
 private final String basePath;
 
 /*锁名称前缀，locker下创建的顺序节点例如都以lock-开头，这样便于过滤无关节点
 *这样创建后的节点类似：lock-00000001，lock-000000002*/
 private staticfinal String LOCK_NAME ="lock-";
 
 /*用于保存某个客户端在locker下面创建成功的顺序节点，用于后续相关操作使用（如判断）*/
 private String ourLockPath;
 
 /**
 * 用于获取锁资源，通过父类的获取锁方法来获取锁
 * @param time获取锁的超时时间
 * @param unit time的时间单位
 * @return是否获取到锁
 * @throws Exception
 */
 private boolean internalLock (long time, TimeUnit unit) throws Exception {
  //如果ourLockPath不为空则认为获取到了锁，具体实现细节见attemptLock的实现
  ourLockPath = attemptLock(time, unit);
  return ourLockPath !=null;
 }
 
 /**
  * 传入Zookeeper客户端连接对象，和basePath
  * @param client Zookeeper客户端连接对象
  * @param basePath basePath是一个持久节点
  */
 public SimpleDistributedLockMutex(ZkClientExt client, String basePath){
  /*调用父类的构造方法在Zookeeper中创建basePath节点，并且为basePath节点子节点设置前缀
  *同时保存basePath的引用给当前类属性*/
  super(client,basePath,LOCK_NAME);
  this.basePath = basePath;
 }
 
 /**获取锁，直到超时，超时后抛出异常*/
 public void acquire() throws Exception {
  //-1表示不设置超时时间，超时由Zookeeper决定
  if (!internalLock(-1,null)){
   throw new IOException("连接丢失!在路径:'"+basePath+"'下不能获取锁!");
  }
 }
 
 /**
 * 获取锁，带有超时时间
 */
 public boolean acquire(long time, TimeUnit unit) throws Exception {
  return internalLock(time, unit);
 }
 
 /**释放锁*/
 public void release()throws Exception {
  releaseLock(ourLockPath);
 }
}

public class BaseDistributedLock {
 
 private final ZkClientExt client;
 private final String path;
 private final String basePath;
 private final String lockName;
 private static final Integer MAX_RETRY_COUNT = 10;
 
 public BaseDistributedLock(ZkClientExt client, String path, String lockName){
 this.client = client;
 this.basePath = path;
 this.path = path.concat("/").concat(lockName); 
 this.lockName = lockName;
 }
 
 private void deleteOurPath(String ourPath) throws Exception{
 client.delete(ourPath);
 }
 
 private String createLockNode(ZkClient client, String path) throws Exception{
 return client.createEphemeralSequential(path, null);
 }
 
 /**
 * 获取锁的核心方法
 * @param startMillis
 * @param millisToWait
 * @param ourPath
 * @return
 * @throws Exception
 */
 private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{
 
 boolean haveTheLock = false;
 boolean doDelete = false;
 
 try{
  while ( !haveTheLock ) {
  //该方法实现获取locker节点下的所有顺序节点，并且从小到大排序 
  List children = getSortedChildren();
  String sequenceNodeName = ourPath.substring(basePath.length()+1);
  
  //计算刚才客户端创建的顺序节点在locker的所有子节点中排序位置，如果是排序为0，则表示获取到了锁
  int ourIndex = children.indexOf(sequenceNodeName);
  
  /*如果在getSortedChildren中没有找到之前创建的[临时]顺序节点，这表示可能由于网络闪断而导致
   *Zookeeper认为连接断开而删除了我们创建的节点，此时需要抛出异常，让上一级去处理
   *上一级的做法是捕获该异常，并且执行重试指定的次数 见后面的 attemptLock方法 */
  if ( ourIndex<0 ){
   throw new ZkNoNodeException("节点没有找到: " + sequenceNodeName);
  }

  //如果当前客户端创建的节点在locker子节点列表中位置大于0，表示其它客户端已经获取了锁
  //此时当前客户端需要等待其它客户端释放锁，
  boolean isGetTheLock = ourIndex == 0;
  
  //如何判断其它客户端是否已经释放了锁？从子节点列表中获取到比自己次小的哪个节点，并对其建立监听
  String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);

  if ( isGetTheLock ){
   haveTheLock = true;
  }else{
   //如果次小的节点被删除了，则表示当前客户端的节点应该是最小的了，所以使用CountDownLatch来实现等待
   String previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
   final CountDownLatch latch = new CountDownLatch(1);
   final IZkDataListener previousListener = new IZkDataListener() {
   
   //次小节点删除事件发生时，让countDownLatch结束等待
   //此时还需要重新让程序回到while，重新判断一次！
   public void handleDataDeleted(String dataPath) throws Exception {
    latch.countDown();  
   }
   
   public void handleDataChange(String dataPath, Object data) throws Exception {
    // ignore     
   }
   };

   try{   
   //如果节点不存在会出现异常
   client.subscribeDataChanges(previousSequencePath, previousListener);
   
   if ( millisToWait != null ){
    millisToWait -= (System.currentTimeMillis() - startMillis);
    startMillis = System.currentTimeMillis();
    if ( millisToWait <= 0 ){
    doDelete = true; // timed out - delete our node
    break;
    }

    latch.await(millisToWait, TimeUnit.MICROSECONDS);
   }else{
    latch.await();
   }
   
   }catch ( ZkNoNodeException e ){
   //ignore
   }finally{
   client.unsubscribeDataChanges(previousSequencePath, previousListener);
   }
  }
  }
 }catch ( Exception e ){
  //发生异常需要删除节点
  doDelete = true;
  throw e;
  
 }finally{
  //如果需要删除节点
  if ( doDelete ){
  deleteOurPath(ourPath);
  }
 }
 return haveTheLock;
 } 
 
 private String getLockNodeNumber(String str, String lockName) {
 int index = str.lastIndexOf(lockName);
 if ( index >= 0 ){
  index += lockName.length();
  return index <= str.length() ? str.substring(index) : "";
 }
 return str;
 }
 
 private List getSortedChildren() throws Exception {
 try{
  List children = client.getChildren(basePath);
  Collections.sort(
  children,
  new Comparator(){
   public int compare(String lhs, String rhs){
   return getLockNodeNumber(lhs, lockName).compareTo(getLockNodeNumber(rhs, lockName));
   }
  }
  );
  return children;
  
 }catch(ZkNoNodeException e){
  client.createPersistent(basePath, true);
  return getSortedChildren();
 }
 }
 
 
 protected void releaseLock(String lockPath) throws Exception{
 deleteOurPath(lockPath); 
 }
 
 
 protected String attemptLock(long time, TimeUnit unit) throws Exception{
 final long startMillis = System.currentTimeMillis();
 final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;

 String  ourPath = null;
 boolean  hasTheLock = false;
 boolean  isDone = false;
 int  retryCount = 0;
 
 //网络闪断需要重试一试
 while ( !isDone ){
  isDone = true;

  try{
  //createLockNode用于在locker（basePath持久节点）下创建客户端要获取锁的[临时]顺序节点
  ourPath = createLockNode(client, path);
  /**
   * 该方法用于判断自己是否获取到了锁，即自己创建的顺序节点在locker的所有子节点中是否最小
   * 如果没有获取到锁，则等待其它客户端锁的释放，并且稍后重试直到获取到锁或者超时
   */
  hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
  
  }catch ( ZkNoNodeException e ){
  if ( retryCount++ < MAX_RETRY_COUNT ){
   isDone = false;
  }else{
   throw e;
  }
  }
 }
 
 if ( hasTheLock ){
  return ourPath;
 }
 
 return null；
｝