InterProcessMutex 分布式锁
InterProcessMutex是常用的一个使用ZK的分布式锁，它会在我们指定的前缀下面创建类型为有序临时的临时节点，然后序号最靠前的节点会获取到锁。 如果某个持有锁的线程挂了之后，对应的节点就会被删除，如果这时候有线程在等待锁，就会获取到锁
类定义
public class InterProcessMutex implements InterProcessLock, Revocable<InterProcessMutex>变量
private final LockInternals internals; //一些内部实现，例如路径，锁名，客户端等
private final String basePath;
//用来存放线程和锁数据的关系，其中的LockData下面有
private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();
private static final String LOCK_NAME = "lock-"; //锁的前缀private static class LockData
{
final Thread owningThread; //拥有锁的线程
final String lockPath; //锁的路径
final AtomicInteger lockCount = new AtomicInteger(1); //锁的数量
private LockData(Thread owningThread, String lockPath)
{
this.owningThread = owningThread;
this.lockPath = lockPath;
}
}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 ) //如果抢到了锁，就返回，然后存在本地的map中
{
LockData newLockData = new LockData(currentThread, lockPath);
threadData.put(currentThread, newLockData);
return true;
}
return false; //没有拿到锁
}attempLock (尝试加锁的逻辑）
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
{ //创建锁，并把锁的path返回来
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;
}createsTheLock(创建节点）
public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception
{
String ourPath;
if ( lockNodeBytes != null )
{//尝试创建一个EPHEMERAL_SEQUENTIAL节点，也就是临时有序节点，如果客户端端口连接，就会被删除
ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, lockNodeBytes);
}
else
{
ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
}
return ourPath;
}
//上面使用了一个函数creatingParentContainersIfNeeded，这个函数会在父节点不存在的时候，自动创建父节点，如果不创建父节点的话，在添加子节点时会出错。
//会根据我们传入的path作为父节点，然后生成一个子节点的路径，创建子节点，并返回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); // 截取出来我们定义的锁路径
PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases); //尝试获取锁
if ( predicateResults.getsTheLock() )
{
haveTheLock = true; //表示拿到锁了
}
else
{
String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
synchronized(this) //如果没有拿到锁，就去监听前面一个子节点，这里用到了zk的wather，会另外一篇文章分析。
{
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; // 如果等待时间到了，还是没有拿到锁，就把我们建立的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;
}getsTheLock （已经建立了节点，判断是否获取锁）
public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
{
int ourIndex = children.indexOf(sequenceNodeName);
validateOurIndex(sequenceNodeName, ourIndex); //校验我们的节点path是否存在，如果没有找到，ourIndex会是-1
boolean getsTheLock = ourIndex < maxLeases; //maxLeases 默认为1，也就是只有排序最靠前的子节点，才能拿到锁
//需要监听的路径，如果已经拿到锁了，不需要监视路径，如果没有拿到，则监视当前节点的前面一个节点
String pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
//组装起来，扔回去
return new PredicateResults(pathToWatch, getsTheLock);
}release （释放锁）
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); //从map中取到当前线程的锁
if ( lockData == null ) //如果没有锁，抛出异常
{
throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
}
//进行原子减操作
int newLockCount = lockData.lockCount.decrementAndGet();
//如果大于0，返回
if ( newLockCount > 0 )
{
return;
}
//如果减后小于0，说明已经多释放了一次，抛异常
if ( newLockCount < 0 )
{
throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
}
//如果为0，说明当前是最后一个锁了，准备释放锁
try
{
internals.releaseLock(lockData.lockPath);
}
//把节点删掉
finally
{
threadData.remove(currentThread);
}
}void releaseLock(String lockPath) throws Exception
{
revocable.set(null); //把存有的锁的信息删掉
deleteOurPath(lockPath); //把路径删掉
}获取锁的时候的调用图
释放锁时候的流程
