Zookeeper分布式锁的使用

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

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

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

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

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

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

5.执行业务代码逻辑

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

给大家推荐大牛的描述，简单易懂：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(){}@Overridepublic 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()平衡，我们再进一步看一下实现

    @Overridepublic 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 instancecan 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-enteringlockData.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 againif ( 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 slashPredicateResults    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 leakclient.getData().usingWatcher(watcher).forPath(previousSequencePath);if ( millisToWait != null ){millisToWait -= (System.currentTimeMillis() - startMillis);startMillis = System.currentTimeMillis();if ( millisToWait <= 0 ){doDelete = true;    // timed out - delete our nodebreak;}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;}

Zookeeper分布式锁的使用相关推荐

[转载] zookeeper 分布式锁服务
转载自http://www.cnblogs.com/shanyou/archive/2012/09/22/2697818.html 分布式锁服务在大家的项目中或许用的不多,因为大家都把排他放在数据库那 ...
zookeeper 分布式锁原理
zookeeper 分布式锁原理: 1 大家也许都很熟悉了多个线程或者多个进程间的共享锁的实现方式了,但是在分布式场景中我们会面临多个Server之间的锁的问题,实现的复杂度比较高.利用基于googl ...
分布式锁原理——redis分布式锁，zookeeper分布式锁
首先分布式锁和我们平常讲到的锁原理基本一样,目的就是确保,在多个线程并发时,只有一个线程在同一刻操作这个业务或者说方法.变量. 在一个进程中,也就是一个jvm 或者说应用中,我们很容易去处理控制,在j ...
关于分布式锁原理的一些学习与思考：redis分布式锁，zookeeper分布式锁
点击上方好好学java ,选择星标公众号重磅资讯.干货,第一时间送达今日推荐:牛人 20000 字的 Spring Cloud 总结,太硬核了~ 作者:队长给我球. 出处:https://w ...
分布式锁（一） Zookeeper分布式锁
什么是Zookeeper? Zookeeper(业界简称zk)是一种提供配置管理.分布式协同以及命名的中心化服务,这些提供的功能都是分布式系统中非常底层且必不可少的基本功能,但是如果自己实现这些功能而 ...
zookeeper 分布式锁服务
分布式锁服务在大家的项目中或许用的不多,因为大家都把排他放在数据库那一层来挡.当大量的行锁.表锁.事务充斥着数据库的时候.一般web应用很多的瓶颈都在数据库上,这里给大家介绍的是减轻数据库锁负担的一种 ...
zookeeper 分布式锁_关于redis分布式锁，zookeeper分布式锁原理的一些学习与思考
编辑:业余草来源:https://www.xttblog.com/?p=4946 首先分布式锁和我们平常讲到的锁原理基本一样,目的就是确保,在多个线程并发时,只有一个线程在同一刻操作这个业务或者说方法 ...
redis cluster 分布式锁_关于分布式锁原理的一些学习与思考redis分布式锁，zookeeper分布式锁...
首先分布式锁和我们平常讲到的锁原理基本一样,目的就是确保,在多个线程并发时,只有一个线程在同一刻操作这个业务或者说方法.变量. 在一个进程中,也就是一个jvm 或者说应用中,我们很容易去处理控制,在j ...
zookeeper分布式锁原理及实现
前言本文介绍下 zookeeper方式实现分布式锁原理简介 zookeeper实现分布式锁的原理就是多个节点同时在一个指定的节点下面创建临时会话顺序节点,谁创建的节点序号最小,谁就获得了锁,并且 ...

Zookeeper分布式锁的使用

Zookeeper分布式锁的使用相关推荐

最新文章

热门文章