curator 分布式锁InterProcessMutex

写这篇文章的目的主要是为了记录下本身在zookeeper 锁上踩过的坑,以及踩坑以后本身的一点认识;

---

从zk分布式锁原理提及,原理很简单,你们也应该都知道,简单的说就是zookeeper实现分布式锁是经过在zk集群上的路径实现的,在获取分布式锁的时候在zk服务器集群节点上建立临时顺序节点,释放锁的时候删除该临时节点.
多么简单的一句话,可是当你实现起来,想去作点优化的时候每每会变得很难,难的咱们后续说;

---

再从需求提及,需求就是加锁,可是因为原来吞吐量不是很大,只是配置了一个固定的锁路径,可是却不是每次都会去根据这个锁路径建立锁,而是将这个锁路径存放在一个本地的HashMap中,这样的话,我就没有必要每次都去重复的建立这个锁对象,简单高效的利用;

---

变动后的需求是这样的,为了下降锁的力度,每次我要动态的生成一个path去zk上进行建立,而后再根据这个path生成锁对象,可是,一开始我依旧是沿用老的思惟,想避免重复建立这个path的锁对象,因而,我想弄个三方缓存来存储这个锁对象,这时候坑就来了;

接下来,咱们开始分析个人踩坑之旅:

public class InterProcessMutex implements InterProcessLock, Revocable<InterProcessMutex> { ... } 

这是curator里面重入锁对象的结构,InterProcessLock这个是curator通用的锁接口,定义的跟jdk自己的也差很少,也是curator留给开发者本身去定制实现符合本身业务需求的锁对象的;Revocable接口是用来执行取消动做时触发动做用到的,若是你自定义锁对象的时候在释放锁对象时想触发一些动做,你能够实现它的方法,以上即是InterProcessLock结构的介绍;

看到这个代码结构咱们还看出什么东西没?它并无实现Serializable,致使其没法被序列化,也就是上面我本身想改进我业务中锁的场景就不支持了,由于相似于redis这种缓存,无法去存放一个对象,它顶多支持字符串以及byte[],因此个人想法就被loss掉了;

即便与业务无关了,可是咱们做为可爱的程序员仍是有必要去研究一下这个玩意的内部实现,由于咱们不知道下次咱们还会遇到什么场景,因此有必要让本身刻骨铭心一次;

接下来,咱们看其内部实现,也就是咱们高大上的源码之旅:

private final LockInternals internals; private final String basePath; private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap(); 

internals:这个是全部申请锁与释放锁的核心实现,待会咱们再来说内部实现;
basePath:锁定的路径;
threadData:内部缓存锁的容器;

实现流程主要是这样的:每次初始化InterProcessMutex对象的时候都会初始化一个StandardLockInternalsDriver对象,这个对象咱们后面再讲它的使用,同时也会初始化一个LockInternals对象,

接下来,咱们来看获取锁的代码:

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 { 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; } 

逻辑以下:
每次获取锁时会直接从本地缓存中先获取锁的元数据,若是存在,则在原有的计数器基础上+1,直接返回;
不然,尝试去获取锁,逻辑以下,

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; } 

首先设置一个是否有锁的标志hasTheLock = false,而后
ourPath = driver.createsTheLock(client, path, localLockNodeBytes);这个地方主要是经过StandardLockInternalsDriver在锁目录下建立EPHEMERAL_SEQUENTIAL节点,
hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);这里主要是循环获取锁的过程,代码看下面,首先是判断是否实现了revocable接口,若是实现了那么就对这个path设置监听,不然的话经过StandardLockInternalsDriver尝试获得PredicateResults(主要是否获得锁及须要监视的目录的两个属性);

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; } 

而后判断PredicateResults中的pathToWatch(主要保存sequenceNode)是不是最小的节点,若是是,则获得锁,getsTheLock为true,不然获得该序列的前一个节点,设为pathToWatch,并监控起来;再判断获取锁的时间是否超时,超时则删除节点,不竞争下次锁,不然,睡眠等待获取锁;最后把获取的锁对象的锁路径等信息封装成LockData存储在本地缓存中.

获取锁的逻辑主要就是这些,有兴趣的同窗能够打断点跟踪学习下,

---

下面是释放锁的过程;

public void release() 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 ) { throw new IllegalMonitorStateException("You do not own the lock: " + basePath); } int newLockCount = lockData.lockCount.decrementAndGet(); if ( newLockCount > 0 ) { return; } if ( newLockCount < 0 ) { throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath); } try { internals.releaseLock(lockData.lockPath); } finally { threadData.remove(currentThread); } } 

代码很简单,从本地缓存中拿到锁对象,计数器-1,只有到那个计数器=0的时候才会去执internals.releaseLock(lockData.lockPath);

final void releaseLock(String lockPath) throws Exception { client.removeWatchers(); revocable.set(null); deleteOurPath(lockPath); } 

只要逻辑见名知意,首先移除watcher监听,这个监听多是在循环获取锁的时候建立的,而后取消动做时触发动做时间置空,最后就是删除path;

最后作个小总结吧

• 1. curator的InterProcessLock接口提供了多种锁机制,互斥锁，读写锁,以及可定时数的互斥锁的机制(这个你们具体问题具体分析).
• 2. 全部申请锁都会建立临时顺序节点,保证了都可以有机会去获取锁.
• 3. 内部用了线程的wait()和notifyAll()这种等待机制,能够及时的唤醒最渴望获得锁的线程.避免常规利用Thread.sleep()这种无用的间隔等待机制.
• 4. 利用redis作锁的时候,通常都须要作锁的有效时间限定。而curator则利用了zookeeper的临时顺序节点特性,一旦客户端失去链接后,则就会自动清除该节点.

 

转自： https://www.jianshu.com/p/5fa6a1464076