zookeeper-分布式锁的代码实现-【每日五分钟搞定大数据】
本文涉及到几个zookeeper简单的知识点,永久节点、有序节点、watch机制。比较基础,熟悉的就别看了跳过这篇吧
- 每个线程在/locks节点下创建一个临时有序节点test_lock_0000000040
- 获得/locks节点下所有子节点A、B、C,排序获得最小值
- 若当前节点B为最小值则获得锁,执行业务逻辑
- 若当前节点B不是最小值则watch比自己小1的节点A,节点A存在则await,否则获得锁
总结:临时有序节点排序后watch比自己小1的节点。
下面看代码
1.线程初始化
创建一个名字为lockName的永久节点(只有永久节点才可以创建子节点)
DistributedLock(String url, String lockName) {
this.lockName = lockName;
try {
zkConn = new ZooKeeper(url, sessionTimeout, this);
if (zkConn.exists(ROOT_LOCK, false) == null)
zkConn.create(ROOT_LOCK, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (IOException | InterruptedException | KeeperException e) {
e.printStackTrace();
}
}
复习一下:zookeeper的节点有以下几种类型:永久,永久有序,临时,临时有序
public enum CreateMode {
PERSISTENT(0, false, false),
PERSISTENT_SEQUENTIAL(2, false, true),
EPHEMERAL(1, true, false),
EPHEMERAL_SEQUENTIAL(3, true, true);
2.线程尝试获得锁tryLock
创建临时有序节点,会在你的lockName后面加上一串编号,例如/locks/test_lock_0000000035
CURRENT_LOCK = zkConn.create(ROOT_LOCK + "/" + lockName + splitStr, new byte[0],
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
获取当前lockName下的所有子临时节点
List<String> subNodes = zkConn.getChildren(ROOT_LOCK, false);
把名字都取出来放进list排个序
List<String> lockObjects = new ArrayList<>();
for (String node : subNodes) {
String _node = node.split(splitStr)[0];
if (_node.equals(lockName)) lockObjects.add(node);
}
若当前节点为最小节点,则直接获取锁成功
if (CURRENT_LOCK.equals(ROOT_LOCK + "/" + lockObjects.get(0))) {
return true;
}
若不是最小节点,开始等待,见下一步。
3.开始等待锁
获得当前线程需要等待的节点名
- 获得当前节点的节点名即test_lock_0000000035
- 获得当前几点在所有排队等锁的线程中的排序
- 根据排序-1,获得排在自己前面的线程的节点名
String prevNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/") + 1);
int prevNodePosition = Collections.binarySearch(lockObjects, prevNode);
WAIT_LOCK = lockObjects.get(prevNodePosition - 1);
给exists加上watcher,监控当前线程等待的节点waitLock是否还存在
若存在即stat不为null,当前线程先开始await
同时watch线程也启动了开始监控exists操作
若是exists状态有变化了(即waitLock不存在了)触发watch线程的countDown操作。
countDown操作使当前线程结束waiting,获得锁,开始继续往后执行
private boolean waitForLock(String waitLock, long waitTime) throws KeeperException, InterruptedException {
Stat stat = zkConn.exists(ROOT_LOCK + "/" + waitLock, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
if (countDownLatch != null) {
countDownLatch.countDown();
}
}
});
if (stat != null) {
System.out.println(Thread.currentThread().getName() + " is waiting for " + ROOT_LOCK + "/" + waitLock);
this.countDownLatch = new CountDownLatch(1);
this.countDownLatch.await(waitTime, TimeUnit.MILLISECONDS);
this.countDownLatch = null;
System.out.println(Thread.currentThread().getName() + " get the lock ");
}
return true;
}
完整的代码可以在我的公众号后台回复9获得,感谢阅读。