InterProcessMutex（可重入互斥锁）
注意：临时节点下不能创建临时子节点
具体流程图：
// 入口1
@Override
public void acquire() throws Exception {
if (!internalLock(-1, null)) {
throw new IOException("Lost connection while trying to acquire lock: " + basePath);
}
}
//入口2
@Override
public boolean acquire(long time, TimeUnit unit) throws Exception {
return internalLock(time, unit);
}
private boolean internalLock(long time, TimeUnit unit) throws Exception {
Thread currentThread = Thread.currentThread();
//可重入，是通过lockCount（AtomicInteger）实现的，加锁-递增，解锁-递减
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 newLockData = new LockData(currentThread, lockPath);
threadData.put(currentThread, newLockData);
return true;
}
return false;
}
org.apache.curator.framework.recipes.locks.LockInternals#attemptLock
//下面创建临时顺序节点后的 事件监听器
private final CuratorWatcher revocableWatcher = new CuratorWatcher() {
@Override
public void process(WatchedEvent event) throws Exception {
// 如果事件类型 = 没有数据更改
if ( event.getType() == Watcher.Event.EventType.NodeDataChanged ) {
checkRevocableWatcher(event.getPath());
}
}
};
//检查可撤销的事件监听器
private void checkRevocableWatcher(String path) throws Exception {
RevocationSpec entry = revocable.get();
if (entry != null) {
try {
byte[] bytes = client.getData().usingWatcher(revocableWatcher).forPath(path);
if (Arrays.equals(bytes, REVOKE_MESSAGE)) {
entry.getExecutor().execute(entry.getRunnable());
}
} catch (KeeperException.NoNodeException ignore) {
// ignore
}
}
}
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 = true
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;
}
}
}
//获得锁成功，返回路径，否则返回null
if ( hasTheLock ) {
return ourPath;
}
return null;
}
private final Watcher watcher = new Watcher() {
@Override
public void process(WatchedEvent event) {
client.postSafeNotify(LockInternals.this);
}
};
org.apache.curator.framework.Curatorframework#postSafeNotify
default CompletableFuture postSafeNotify(Object monitorHolder) {
return runSafe(() -> {
synchronized (monitorHolder) {
// 唤醒所有线程
monitorHolder.notifyAll();
}
});
}
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);
}
// 状态=启动，并未获得锁
// InterProcessMutex两种请求锁：acquire()/acquire(long time, TimeUnit unit)
// 释放本地锁，线程处于等待状态
while ((client.getState() == CuratorframeworkState.STARTED) && !haveTheLock) {
//获得排序后的子节点集合
List children = getSortedChildren();
//获得节点名
String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash 加1-包含斜杠
//通过节点 是否 子节点集合中的首个，判断是否获得锁
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
//使用getData()而不是exists()来避免留下不需要的监视者，这是一种资源泄漏类型
//添加事件监控器（请求安全通知事件监听），可看上方watcher
client.getData().usingWatcher(watcher).forPath(previousSequencePath);
// 等待millisToWait 时间，超时删除节点
if (millisToWait != null) {
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
// 如果超时，在finally代码块，删除节点
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;
}
org.apache.curator.framework.recipes.locks.StandardLockInternalsDriver
// 判断是否获得锁，通过判断 节点名 是否 children（子节点s）中第一个节点，通过maxLeases判断（默认为1）
@Override
public PredicateResults getsTheLock(Curatorframework client, List children, String sequenceNodeName, int maxLeases) throws Exception {
// 节点名在children中的下表
int ourIndex = children.indexOf(sequenceNodeName);
// 校验ourIndex<0
validateOurIndex(sequenceNodeName, ourIndex);
// ourIndex < maxLeases = true，即表明是children首个节点，获得锁成功
boolean getsTheLock = ourIndex < maxLeases;
// 如果获得锁失败，从子节点集合中获得 上一个下标的节点名；pathToWatch 即要监听的节点名
String pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
return new PredicateResults(pathToWatch, getsTheLock);
}
@Override
public String createsTheLock(Curatorframework client, String path, byte[] lockNodeBytes) throws Exception {
String ourPath;
// 创建临时顺序节点（CreateMode.EPHEMERAL_SEQUENTIAL）
// lockNodeBytes不为空，即需要存数据
if ( lockNodeBytes != null ) {
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;
}
