Zookeeper分布式锁的使用

旧城等待， 2022-05-08 22:12 196阅读 0赞

由于公司引入了dubbo+zookeeper框架，里面不可避免的引入的zookeeper分布式锁，所以自己大致了解了一下。由于是自己研究，有不正确的地方还请大佬批评指正。

首先先介绍一下自己对zookeeper分布式锁的理解，之后会引入一版别人的感觉比较好的描述给大家

1.dubbo的微服务后场生产者会暴露接口给前场的消费者。在zookeeper会生成一个相应的节点，比如时候节点名字是/lock。

2.当多个客户端访问的时候，会在这个节点下依次生成临时的顺序节点，只有第一个节点可以获取到锁，其他节点等待。

3.当第一个节点释放锁，这个节点将会被删除，这个行为会被客户端监听，一旦收到这个通知，剩余的客户端会继续抢锁。

4.这种行为导致了羊群效应，所有在等待的客户端都会被触发。于是优化方法是每个客户端应该对刚好在它之前的子节点设置事件监听。我前面这个节点哥们完了，我再抢锁，否则说明我前面有很多人，我就得继续排队

5.执行业务代码逻辑

6.完成业务代码，删除对应的临时节点，释放锁。

给大家推荐大牛的描述，简单易懂：[10分钟看懂！基于Zookeeper的分布式锁][10_Zookeeper]

说完zookeeper分布式锁逻辑上的实现，下面介绍一下Curator，它也提供了对zookeeper分布式锁实现

先看一下大致流程

public class ZKLockTestTread extends Thread{
    	private Logger logger=LoggerFactory.getLogger(getClass());
    	private final String ZKLOCK_NODE_PATH="/zklock/readfile";
        public ZKLockTestTread(){
        }
        @Override
        public void run() {
            AbstractMutexLock lock=null;
            boolean lockflag=false;
            try{
            	lock=ZKMutexLockFactory.getZKMutexLock(ZKLOCK_NODE_PATH);
            	//指定zk的方式
            	//lock=ZKMutexLockFactory.getZKMutexLock("127.0.0.1:8080", ZKLOCK_NODE_PATH);
            	//lock.acquire();//争锁，无限等待
            	lockflag=lock.acquire(10, TimeUnit.SECONDS);//争锁，超时时间10秒。
            	if(lockflag){
            		//获取到分布式锁，执行任务
            		logger.info("SUCESS线程【"+Thread.currentThread().getName()+"】获取到分布式锁，执行任务");
            		Thread.sleep(1000000000);
            	}
            	else{
            		//未获取到分布式锁，不执行任务
            		logger.info("FAILURE线程【"+Thread.currentThread().getName()+"】未获取到分布式锁，不执行任务");
            	}
            } catch (PaasException e) {
            	logger.error("线程【"+Thread.currentThread().getName()+"】获取分布式锁出错："+e.getMessage(),e);
    			//e.printStackTrace();
    		}
            catch(Exception e)
            {
            	logger.error("线程【"+Thread.currentThread().getName()+"】运行出错："+e.getMessage(),e);
    			//e.printStackTrace();
    
            }
            finally{
            	if(lock!=null&&lockflag){
            		try {
    					lock.release();
    					logger.error("线程【"+Thread.currentThread().getName()+"】释放分布式锁OK");
    				} catch (Exception e) {
    					logger.error("线程【"+Thread.currentThread().getName()+"】释放分布式锁出错："+e.getMessage(),e);
    					//e.printStackTrace();
    				}
            	}
            }
            //System.out.println("【"+Thread.currentThread().getName()+"】");
        } 
    }

这里指定了节点的路径：

AbstractMutexLock lock=ZKMutexLockFactory.getZKMutexLock(ZKLOCK_NODE_PATH);

获得锁的方式是：

boolean lockflag=lock.acquire(10, TimeUnit.SECONDS);

返回值就是是否拿到了锁，接着我们点进去看一眼，首先是接口

/**
         * Acquire the mutex - blocks until it's available or the given time expires. Each call to acquire that returns true must be balanced by a call
         * to {@link #release()}
         *
         * @param time time to wait
         * @param unit time unit
         * @return true if the mutex was acquired, false if not
         * @throws Exception ZK errors, connection interruptions
         */
        public boolean acquire(long time, TimeUnit unit) throws Exception;

意思是说直到他状态是可用或者超过了超时时间才获取互斥锁，需用release()平衡，我们再进一步看一下实现

@Override
        public boolean acquire(long time, TimeUnit unit) throws Exception
        {
            return internalLock(time, unit);
        }

调用了internalLock才真正的调用了zookeeper锁。多说一句，time的值如果为-1，说明锁被占用时永久阻塞等待

接着进入internalLock

private boolean internalLock(long time, TimeUnit unit) throws Exception
        {
            /*
               Note on concurrency: a given lockData instance
               can be only acted on by a single thread so locking isn't necessary
            */
    
            Thread currentThread = Thread.currentThread();
    
            LockData lockData = threadData.get(currentThread);
            if ( lockData != null )
            {
                // re-entering
                lockData.lockCount.incrementAndGet();
                return true;
            }
    
            String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
            if ( lockPath != null )
            {
                LockData newLockData = new LockData(currentThread, lockPath);
                threadData.put(currentThread, newLockData);
                return true;
            }
    
            return false;
        }

attemptLock尝试去获得锁处理了zookeeper锁

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
        {
            final long      startMillis = System.currentTimeMillis();
            final Long      millisToWait = (unit != null) ? unit.toMillis(time) : null;
            final byte[]    localLockNodeBytes = (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 = driver.createsTheLock(client, path, localLockNodeBytes);
                    hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
                }
                catch ( KeeperException.NoNodeException e )
                {
                    // gets thrown by StandardLockInternalsDriver when it can't find the lock node
                    // this can happen when the session expires, etc. So, if the retry allows, just try it all again
                    if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
                    {
                        isDone = false;
                    }
                    else
                    {
                        throw e;
                    }
                }
            }
    
            if ( hasTheLock )
            {
                return ourPath;
            }
    
            return null;
        }

其中internalLockLoop：阻塞等待直到获得锁

private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
        {
            boolean     haveTheLock = false;
            boolean     doDelete = false;
            try
            {
                if ( revocable.get() != null )
                {
                    client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
                }
    
                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();
    
                        synchronized(this)
                        {
                            try 
                            {
                                // use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leak
                                client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                                if ( millisToWait != null )
                                {
                                    millisToWait -= (System.currentTimeMillis() - startMillis);
                                    startMillis = System.currentTimeMillis();
                                    if ( millisToWait <= 0 )
                                    {
                                        doDelete = true;    // timed out - delete our node
                                        break;
                                    }
    
                                    wait(millisToWait);
                                }
                                else
                                {
                                    wait();
                                }
                            }
                            catch ( KeeperException.NoNodeException e ) 
                            {
                                // it has been deleted (i.e. lock released). Try to acquire again
                            }
                        }
                    }
                }
            }
            catch ( Exception e )
            {
                ThreadUtils.checkInterrupted(e);
                doDelete = true;
                throw e;
            }
            finally
            {
                if ( doDelete )
                {
                    deleteOurPath(ourPath);
                }
            }
            return haveTheLock;
        }

[10_Zookeeper]: https://blog.csdn.net/qiangcuo6087/article/details/79067136