基于Zookeeper的分布式锁详解大数据

<dependency>  
    <groupId>org.apache.curator</groupId>  
    <artifactId>curator-recipes</artifactId>  
    <version>4.0.0</version>  
</dependency>

public static void main(String[] args) throws Exception {  
    //创建zookeeper的客户端  
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);  
    CuratorFramework client = CuratorFrameworkFactory.newClient("10.21.41.181:2181,10.21.42.47:2181,10.21.49.252:2181", retryPolicy);  
    client.start();  
  
    //创建分布式锁, 锁空间的根节点路径为/curator/lock  
    InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");  
    mutex.acquire();  
    //获得了锁, 进行业务流程  
    System.out.println("Enter mutex");  
    //完成业务流程, 释放锁  
    mutex.release();  
      
    //关闭客户端  
    client.close();  
} 

/*  
 * 获取锁，当锁被占用时会阻塞等待，这个操作支持同线程的可重入（也就是重复获取锁），acquire的次数需要与release的次数相同。  
 * @throws Exception ZK errors, connection interruptions  
 */  
@Override  
public void acquire() throws Exception  
{  
    if ( !internalLock(-1, null) )  
    {  
        throw new IOException("Lost connection while trying to acquire lock: " + basePath);  
    }  
} 

private boolean internalLock(long time, TimeUnit unit) throws Exception  
{  
  
    //这里处理同线程的可重入性，如果已经获得锁，那么只是在对应的数据结构中增加acquire的次数统计，直接返回成功  
    Thread currentThread = Thread.currentThread();  
    LockData lockData = threadData.get(currentThread);  
    if ( lockData != null )  
    {  
        // re-entering  
        lockData.lockCount.incrementAndGet();  
        return true;  
    }  
  
    //这里才真正去zookeeper中获取锁  
    String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());  
    if ( lockPath != null )  
    {  
        //获得锁之后，记录当前的线程获得锁的信息，在重入时只需在LockData中增加次数统计即可  
        LockData newLockData = new LockData(currentThread, lockPath);  
        threadData.put(currentThread, newLockData);  
        return true;  
    }  
  
    //在阻塞返回时仍然获取不到锁，这里上下文的处理隐含的意思为zookeeper通信异常  
    return false;  
} 

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception  
{  
    //参数初始化，此处省略  
    //...  
     
    //自旋获取锁  
    while ( !isDone )  
    {  
        isDone = true;  
  
        try  
        {  
            //在锁空间下创建临时且有序的子节点  
            ourPath = driver.createsTheLock(client, path, localLockNodeBytes);  
            //判断是否获得锁（子节点序号最小），获得锁则直接返回，否则阻塞等待前一个子节点删除通知  
            hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);  
        }  
        catch ( KeeperException.NoNodeException e )  
        {  
            //对于NoNodeException，代码中确保了只有发生session过期才会在这里抛出NoNodeException，因此这里根据重试策略进行重试  
            if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, 
                     RetryLoop.getDefaultRetrySleeper()) )  
            {  
                isDone = false;  
            }  
            else  
            {  
                throw e;  
            }  
        }  
    }  
  
    //如果获得锁则返回该子节点的路径  
    if ( hasTheLock )  
    {  
        return ourPath;  
    }  
  
    return null;  
} 

//自旋直至获得锁  
while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )  
{  
    //获取所有的子节点列表，并且按序号从小到大排序  
    List<String>        children = getSortedChildren();  
      
    //根据序号判断当前子节点是否为最小子节点  
    String              sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash  
    PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);  
    if ( predicateResults.getsTheLock() )  
    {  
        //如果为最小子节点则认为获得锁  
        haveTheLock = true;  
    }  
    else  
    {  
        //否则获取前一个子节点  
        String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();  
  
        //这里使用对象监视器做线程同步，当获取不到锁时监听前一个子节点删除消息并且进行wait()，当前一个子节点删除（也就是锁释放）时，
          回调会通过notifyAll唤醒此线程，此线程继续自旋判断是否获得锁  
        synchronized(this)  
        {  
            try   
            {  
                //这里使用getData()接口而不是checkExists()是因为，如果前一个子节点已经被删除了那么会抛出异常而且不会设置事件监听器，
                  而checkExists虽然也可以获取到节点是否存在的信息但是同时设置了监听器，这个监听器其实永远不会触发，对于zookeeper来说属于资源泄露  
                client.getData().usingWatcher(watcher).forPath(previousSequencePath);  
  
                //如果设置了阻塞等待的时间  
                if ( millisToWait != null )  
                {  
                    millisToWait -= (System.currentTimeMillis() - startMillis);  
                    startMillis = System.currentTimeMillis();  
                    if ( millisToWait <= 0 )  
                    {  
                        doDelete = true;    // 等待时间到达，删除对应的子节点  
                        break;  
                    }  
                      
                    //等待相应的时间  
                    wait(millisToWait);  
                }  
                else  
                {  
                   //永远等待  
                    wait();  
                }  
            }  
            catch ( KeeperException.NoNodeException e )   
            {  
                //上面使用getData来设置监听器时，如果前一个子节点已经被删除那么会抛出NoNodeException，只需要自旋一次即可，无需额外处理  
            }  
        }  
    }  
}