分布式锁-ZK实现

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
//创建zk连接。该构造器要求传入三个参数分别是：ip:端口(String)、会话超时时间、本次连接的监听器。
public String create(String path, byte[] data, List<ACL> acl, CreateMode createMode)
//创建节点。参数：节点路径、节点数据、权限策略、节点类型

public void delete(String path, int version)
删除指定节点。参数：节点路径、数据版本号
public synchronized void close()
断开zk链接

public class ZooKeeperLock implements Watcher {

    private ZooKeeper zk = null;
    private String rootLockNode;            // 锁的根节点
    private String lockName;                // 竞争资源，用来生成子节点名称
    private String currentLock;             // 当前锁
    private String waitLock;                // 等待的锁（前一个锁）
    private CountDownLatch countDownLatch;  // 计数器（用来在加锁失败时阻塞加锁线程）
    private int sessionTimeout = 30000;     // 超时时间
    
    // 1. 构造器中创建ZK链接，创建锁的根节点
    public ZooKeeperLock(String zkAddress, String rootLockNode, String lockName) {
        this.rootLockNode = rootLockNode;
        this.lockName = lockName;
        try {
            // 创建连接，zkAddress格式为：IP:PORT
            zk = new ZooKeeper(zkAddress,this.sessionTimeout,this);
            // 检测锁的根节点是否存在，不存在则创建
            Stat stat = zk.exists(rootLockNode,false);
            if (null == stat) {
                zk.create(rootLockNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
    
    // 2. 加锁方法，先尝试加锁，不能加锁则等待上一个锁的释放
    public boolean lock() {
        if (this.tryLock()) {
            System.out.println("线程【" + Thread.currentThread().getName() + "】加锁（" + this.currentLock + "）成功！");
            return true;
        } else {
            return waitOtherLock(this.waitLock, this.sessionTimeout);
        }
    }
    
    public boolean tryLock() {
        // 分隔符
        String split = "_lock_";
        if (this.lockName.contains("_lock_")) {
            throw new RuntimeException("lockName can't contains '_lock_' ");
        }
        try {
            // 创建锁节点（临时有序节点）
            this.currentLock = zk.create(this.rootLockNode + "/" + this.lockName + split, new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println("线程【" + Thread.currentThread().getName() 
                        + "】创建锁节点（" + this.currentLock + "）成功，开始竞争...");
            // 取所有子节点
            List<String> nodes = zk.getChildren(this.rootLockNode, false);
            // 取所有竞争lockName的锁
            List<String> lockNodes = new ArrayList<String>();
            for (String nodeName : nodes) {
                if (nodeName.split(split)[0].equals(this.lockName)) {
                    lockNodes.add(nodeName);
                }
            }
            Collections.sort(lockNodes);
            // 取最小节点与当前锁节点比对加锁
            String currentLockPath = this.rootLockNode + "/" + lockNodes.get(0);
            if (this.currentLock.equals(currentLockPath)) {
                return true;
            }
            // 加锁失败，设置前一节点为等待锁节点
            String currentLockNode = this.currentLock.substring(this.currentLock.lastIndexOf("/") + 1);
            int preNodeIndex = Collections.binarySearch(lockNodes, currentLockNode) - 1;
            this.waitLock = lockNodes.get(preNodeIndex);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    private boolean waitOtherLock(String waitLock, int sessionTimeout) {
        boolean islock = false;
        try {
            // 监听等待锁节点
            String waitLockNode = this.rootLockNode + "/" + waitLock;
            Stat stat = zk.exists(waitLockNode,true);
            if (null != stat) {
                System.out.println("线程【" + Thread.currentThread().getName() 
                            + "】锁（" + this.currentLock + "）加锁失败，等待锁（" + waitLockNode + "）释放...");
                // 设置计数器，使用计数器阻塞线程
                this.countDownLatch = new CountDownLatch(1);
                islock = this.countDownLatch.await(sessionTimeout,TimeUnit.MILLISECONDS);
                this.countDownLatch = null;
                if (islock) {
                    System.out.println("线程【" + Thread.currentThread().getName() + "】锁（" 
                                + this.currentLock + "）加锁成功，锁（" + waitLockNode + "）已经释放");
                } else {
                    System.out.println("线程【" + Thread.currentThread().getName() + "】锁（" 
                                + this.currentLock + "）加锁失败...");
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return islock;
    }
    
    // 3. 释放锁
    public void unlock() throws InterruptedException {
        try {
            Stat stat = zk.exists(this.currentLock,false);
            if (null != stat) {
                System.out.println("线程【" + Thread.currentThread().getName() + "】释放锁 " + this.currentLock);
                zk.delete(this.currentLock, -1);
                this.currentLock = null;
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        } finally {
            zk.close();
        }
    }
    
    // 4. 监听器回调
    @Override
    public void process(WatchedEvent watchedEvent) {
        if (null != this.countDownLatch && watchedEvent.getType() == Event.EventType.NodeDeleted) {
            // 计数器减一，恢复线程操作
            this.countDownLatch.countDown();
        }
    }
}

public class Test {
    public static void doSomething() {
        System.out.println("线程【" + Thread.currentThread().getName() + "】正在运行...");
    }

    public static void main(String[] args) {
        Runnable runnable = new Runnable() {
            public void run() {
                ZooKeeperLock lock = null;
                lock = new ZooKeeperLock("10.150.27.51:2181","/locks", "test1");
                if (lock.lock()) {
                    doSomething();
                    try {
                        Thread.sleep(1000);
                        lock.unlock();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };

        for (int i = 0; i < 5; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }
}

// 设置重试策略，创建zk客户端
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
CuratorFramework client = CuratorFrameworkFactory.newClient("10.150.27.51:2181",retryPolicy);
// 启动客户端
client.start();
// 创建分布式可重入排他锁，监听客户端为client，锁的根节点为/locks
InterProcessMutex mutex = new InterProcessMutex(client,"/locks");
try {
    // 加锁
    if (mutex.acquire(3,TimeUnit.SECONDS)) {
        // TODO-同步操作
        //释放锁
        mutex.release();
    }
} catch (Exception e) {
    e.printStackTrace();
} finally {
    client.close();
}

// InterProcessMutex.class
  public void acquire() throws Exception {
      if (!this.internalLock(-1L, (TimeUnit)null)) {
          throw new IOException("Lost connection while trying to acquire lock: " + this.basePath);
      }
  }
  
  public boolean acquire(long time, TimeUnit unit) throws Exception {
      return this.internalLock(time, unit);
  }
  
  private boolean internalLock(long time, TimeUnit unit) throws Exception {
      Thread currentThread = Thread.currentThread();
      InterProcessMutex.LockData lockData = (InterProcessMutex.LockData)this.threadData.get(currentThread);
      if (lockData != null) {
          // 锁的可重入性
          lockData.lockCount.incrementAndGet();
          return true;
      } else {
          // 加锁并返回锁节点
          String lockPath = this.internals.attemptLock(time, unit, this.getLockNodeBytes());
          if (lockPath != null) {
              InterProcessMutex.LockData newLockData = new InterProcessMutex.LockData(currentThread, lockPath);
              this.threadData.put(currentThread, newLockData);
              return true;
          } else {
              return false;
          }
      }
  }

// LockInternals.class
 String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception {
     long startMillis = System.currentTimeMillis();
     Long millisToWait = unit != null ? unit.toMillis(time) : null;
     byte[] localLockNodeBytes = this.revocable.get() != null ? new byte[0] : lockNodeBytes;
     int retryCount = 0;
     String ourPath = null;
     boolean hasTheLock = false;
     boolean isDone = false;

     while(!isDone) {
         isDone = true;
         try {
             // 创建锁节点
             ourPath = this.driver.createsTheLock(this.client, this.path, localLockNodeBytes);
             // 竞争锁
             hasTheLock = this.internalLockLoop(startMillis, millisToWait, ourPath);
         } catch (NoNodeException var14) {
             if (!this.client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++,  
                     System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) {
                 throw var14;
             }

             isDone = false;
         }
     }

     return hasTheLock ? ourPath : null;
 }

 // LockInternals.class
private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception {
    boolean haveTheLock = false;
    boolean doDelete = false;

    try {
        if (this.revocable.get() != null) {
            ((BackgroundPathable)this.client.getData().usingWatcher(this.revocableWatcher)).forPath(ourPath);
        }
        while(this.client.getState() == CuratorFrameworkState.STARTED && !haveTheLock) {
            // 获取所有子节点
            List<String> children = this.getSortedChildren();
            // 获取当前锁节点
            String sequenceNodeName = ourPath.substring(this.basePath.length() + 1);
            // 使用锁驱动加锁
            PredicateResults predicateResults = this.driver.getsTheLock(this.client, children, 
                        sequenceNodeName, this.maxLeases);
            if (predicateResults.getsTheLock()) {
                haveTheLock = true;
            } else {
                // 阻塞等待上一个锁释放
                String previousSequencePath = this.basePath + "/" + predicateResults.getPathToWatch();
                synchronized(this) {
                    try {
                        ((BackgroundPathable)this.client.getData().usingWatcher(this.watcher)).forPath(previousSequencePath);
                        if (millisToWait == null) {
                            // 未设置超时一直阻塞
                            this.wait();
                        } else {
                            millisToWait = millisToWait - (System.currentTimeMillis() - startMillis);
                            startMillis = System.currentTimeMillis();
                            // 根据时间设置阻塞时间
                            if (millisToWait > 0L) {
                                this.wait(millisToWait);
                            } else {
                                // 已经超时，设置删除节点标识
                                doDelete = true;
                                break;
                            }
                        }
                    } catch (NoNodeException var19) {
                        ;
                    }
                }
            }
        }
    } catch (Exception var21) {
        ThreadUtils.checkInterrupted(var21);
        doDelete = true;
        throw var21;
    } finally {
        if (doDelete) {
            // 删除已超时的锁节点
            this.deleteOurPath(ourPath);
        }
    }
    return haveTheLock;
}

// StandardLockInternalsDriver.class
public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, 
        int maxLeases) throws Exception {
    int ourIndex = children.indexOf(sequenceNodeName);
    validateOurIndex(sequenceNodeName, ourIndex);
    boolean getsTheLock = ourIndex < maxLeases;
    String pathToWatch = getsTheLock ? null : (String)children.get(ourIndex - maxLeases);
    return new PredicateResults(pathToWatch, getsTheLock);
}