之前在利用zookeeper实现分布式锁时候,使用了ConcurrentHashMap保存currentPath等,现在有了ThreadLocal,我们可以直接使用ThreadLocal了
把MyZookeeperLock的类进行修改,其他代码不变,依然可以用
public class MyZookeeperLock implements Lock {
private String lockPath;
private ZkClient zkClient;
//保存当前节点
public static ThreadLocal<String> localCurrentPath = new ThreadLocal<>();
//记录要监听的节点
public static ThreadLocal<String> localBeforePath = new ThreadLocal<>();
//锁的层数
public static ThreadLocal<Integer> localDeepNum = new ThreadLocal<>();
public MyZookeeperLock(String lockPath) {
super();
this.lockPath = lockPath;
zkClient = new ZkClient("127.0.0.1:2181");
zkClient.setZkSerializer(new ZkSerializer() {
@Override
public Object deserialize(byte[] bytes) throws ZkMarshallingError {
try {
return new String(bytes, "UTF-8");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
throw new ZkMarshallingError(e);
}
}
@Override
public byte[] serialize(Object obj) throws ZkMarshallingError {
try {
return String.valueOf(obj).getBytes("UTF-8");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
throw new ZkMarshallingError(e);
}
}
});
if (!this.zkClient.exists(lockPath)) {
try {
//创建父节点
this.zkClient.createPersistent(lockPath);
} catch (ZkNodeExistsException e) {
}
}
}
@Override
public void lock() {
//不知道是第几次进,保存层数
Integer deep = localDeepNum.get();
if (deep == null) {
localDeepNum.set(1);
} else {
localDeepNum.set(deep + 1);
}
while (!tryLock()) {
//没拿到锁的时候,就去等待
waitForLock();
}
}
private void waitForLock() {
CountDownLatch countDownLatch = new CountDownLatch(1);
//监听
IZkDataListener listener = new IZkDataListener() {
@Override
public void handleDataChange(String s, Object o) throws Exception {
}
@Override
public void handleDataDeleted(String s) throws Exception {
//监听的节点被删除时,执行
//System.out.println("----监听节点被删除");
countDownLatch.countDown();
}
};
//找到我要监听的节点
String beforePath = localBeforePath.get();
//开始监听
zkClient.subscribeDataChanges(beforePath, listener);
if (this.zkClient.exists(beforePath)) {
try {
//走到这,说明被执行了
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//结束监听
zkClient.unsubscribeDataChanges(beforePath, listener);
}
@Override
public void lockInterruptibly() throws InterruptedException {
}
@Override
public boolean tryLock() {
String currentPath = localCurrentPath.get();
if (currentPath == null) {
//新增一个顺序节点
currentPath = this.zkClient.createEphemeralSequential(lockPath + "/", "zk");
localCurrentPath.set(currentPath);
}
//找到所有的节点
List<String> chilren = this.zkClient.getChildren(lockPath);
Collections.sort(chilren);
//判断自己是不是排在最前面
if (currentPath.equals(lockPath + "/" + chilren.get(0))) {
//排在最前面,我拿到锁了
return true;
} else {
//没有排在最前面,找到我要等谁
int curIndex = chilren.indexOf(currentPath.substring(lockPath.length() + 1));
String beforePath = lockPath + "/" + chilren.get(curIndex - 1);
localBeforePath.set(beforePath);
}
return false;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public void unlock() {
//判断层级
Integer deep = localDeepNum.get();
if (deep == null) {
System.out.println("出问题了");
return;
}
deep--;
if (deep == 0) {
localDeepNum.remove();
} else {
localDeepNum.set(deep);
//说明没有彻底出去
return;
}
//说明锁彻底结束了
String currentPath = localCurrentPath.get();
if (currentPath == null) {
System.out.println("出问题");
}
//移除这些没必要的
localCurrentPath.remove();
localBeforePath.remove();
//删除节点
this.zkClient.delete(currentPath);
}
@Override
public Condition newCondition() {
return null;
}
}