锁
)的地方就有江湖(事务
),今天不谈江湖,来撩撩人。临时节点
,zk会保证只有一个客户端创建成功。获取锁->监听->获取锁
。/**
* @Description ZK分布式锁的接口
* @Author 陈某
* @Date 2020/4/7 22:52
*/
public interface ZKLock {
/**
* 获取锁
*/
void lock() throws Exception;
/**
* 解锁
*/
void unlock() throws Exception;
}
/**
* @Description 排他锁,模板类
* @Author 陈某
* @Date 2020/4/7 22:55
*/
public abstract class AbstractZKLockMutex implements ZKLock {
/**
* 节点路径
*/
protected String lockPath;
/**
* zk客户端
*/
protected CuratorFramework zkClient;
private AbstractZKLockMutex(){}
public AbstractZKLockMutex(String lockPath,CuratorFramework client){
this.lockPath=lockPath;
this.zkClient=client;
}
/**
* 模板方法,搭建的获取锁的框架,具体逻辑交于子类实现
* @throws Exception
*/
@Override
public final void lock() throws Exception {
//获取锁成功
if (tryLock()){
System.out.println(Thread.currentThread().getName()+"获取锁成功");
}else{ //获取锁失败
//阻塞一直等待
waitLock();
//递归,再次获取锁
lock();
}
}
/**
* 尝试获取锁,子类实现
*/
protected abstract boolean tryLock() ;
/**
* 等待获取锁,子类实现
*/
protected abstract void waitLock() throws Exception;
/**
* 解锁:删除节点或者直接断开连接
*/
@Override
public abstract void unlock() throws Exception;
}
/**
* @Description 排他锁的实现类,继承模板类 AbstractZKLockMutex
* @Author 陈某
* @Date 2020/4/7 23:23
*/
@Data
public class ZKLockMutex extends AbstractZKLockMutex {
/**
* 用于实现线程阻塞
*/
private CountDownLatch countDownLatch;
public ZKLockMutex(String lockPath,CuratorFramework zkClient){
super(lockPath,zkClient);
}
/**
* 尝试获取锁:直接创建一个临时节点,如果这个节点存在创建失败抛出异常,表示已经互斥了,
* 反之创建成功
* @throws Exception
*/
@Override
protected boolean tryLock() {
try {
zkClient.create()
//临时节点
.withMode(CreateMode.EPHEMERAL)
//权限列表 world:anyone:crdwa
.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
.forPath(lockPath,"lock".getBytes());
return true;
}catch (Exception ex){
return false;
}
}
/**
* 等待锁,一直阻塞监听
* @return 成功获取锁返回true,反之返回false
*/
@Override
protected void waitLock() throws Exception {
//监听节点的新增、更新、删除
final NodeCache nodeCache = new NodeCache(zkClient, lockPath);
//启动监听
nodeCache.start();
ListenerContainer<NodeCacheListener> listenable = nodeCache.getListenable();
//监听器
NodeCacheListener listener=()-> {
//节点被删除,此时获取锁
if (nodeCache.getCurrentData() == null) {
//countDownLatch不为null,表示节点存在,此时监听到节点删除了,因此-1
if (countDownLatch != null)
countDownLatch.countDown();
}
};
//添加监听器
listenable.addListener(listener);
//判断节点是否存在
Stat stat = zkClient.checkExists().forPath(lockPath);
//节点存在
if (stat!=null){
countDownLatch=new CountDownLatch(1);
//阻塞主线程,监听
countDownLatch.await();
}
//移除监听器
listenable.removeListener(listener);
}
/**
* 解锁,直接删除节点
* @throws Exception
*/
@Override
public void unlock() throws Exception {
zkClient.delete().forPath(lockPath);
}
}
ConcurrentMap
,key
是当前线程,value
是定义的数据,结构如下:private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();
public boolean tryLock(){
//判断当前线程是否在threadData保存过
//存在,直接return true
//不存在执行获取锁的逻辑
//获取成功保存在threadData中
}
读锁
或称作读节点
)。从小到大
排序写锁
或称作写节点
)。从小到大
排序。public class ZKLockRW {
/**
* 节点路径
*/
protected String lockPath;
/**
* zk客户端
*/
protected CuratorFramework zkClient;
/**
* 用于阻塞线程
*/
private CountDownLatch countDownLatch=new CountDownLatch(1);
private final static String WRITE_NAME="_W_LOCK";
private final static String READ_NAME="_R_LOCK";
public ZKLockRW(String lockPath, CuratorFramework client) {
this.lockPath=lockPath;
this.zkClient=client;
}
/**
* 获取锁,如果获取失败一直阻塞
* @throws Exception
*/
public void lock() throws Exception {
//创建节点
String node = createNode();
//阻塞等待获取锁
tryLock(node);
countDownLatch.await();
}
/**
* 创建临时有序节点
* @return
* @throws Exception
*/
private String createNode() throws Exception {
//创建临时有序节点
return zkClient.create()
.withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
.forPath(lockPath);
}
/**
* 获取写锁
* @return
*/
public ZKLockRW writeLock(){
return new ZKLockRW(lockPath+WRITE_NAME,zkClient);
}
/**
* 获取读锁
* @return
*/
public ZKLockRW readLock(){
return new ZKLockRW(lockPath+READ_NAME,zkClient);
}
private void tryLock(String nodePath) throws Exception {
//获取所有的子节点
List<String> childPaths = zkClient.getChildren()
.forPath("/")
.stream().sorted().map(o->"/"+o).collect(Collectors.toList());
//第一个节点就是当前的锁,直接获取锁。递归结束的条件
if (nodePath.equals(childPaths.get(0))){
countDownLatch.countDown();
return;
}
//1. 读锁:监听最前面的写锁,写锁释放了,自然能够读了
if (nodePath.contains(READ_NAME)){
//查找临近的写锁
String preNode = getNearWriteNode(childPaths, childPaths.indexOf(nodePath));
if (preNode==null){
countDownLatch.countDown();
return;
}
NodeCache nodeCache=new NodeCache(zkClient,preNode);
nodeCache.start();
ListenerContainer<NodeCacheListener> listenable = nodeCache.getListenable();
listenable.addListener(() -> {
//节点删除事件
if (nodeCache.getCurrentData()==null){
//继续监听前一个节点
String nearWriteNode = getNearWriteNode(childPaths, childPaths.indexOf(preNode));
if (nearWriteNode==null){
countDownLatch.countDown();
return;
}
tryLock(nearWriteNode);
}
});
}
//如果是写锁,前面无论是什么锁都不能读,直接循环监听上一个节点即可,直到前面无锁
if (nodePath.contains(WRITE_NAME)){
String preNode = childPaths.get(childPaths.indexOf(nodePath) - 1);
NodeCache nodeCache=new NodeCache(zkClient,preNode);
nodeCache.start();
ListenerContainer<NodeCacheListener> listenable = nodeCache.getListenable();
listenable.addListener(() -> {
//节点删除事件
if (nodeCache.getCurrentData()==null){
//继续监听前一个节点
tryLock(childPaths.get(childPaths.indexOf(preNode) - 1<0?0:childPaths.indexOf(preNode) - 1));
}
});
}
}
/**
* 查找临近的写节点
* @param childPath 全部的子节点
* @param index 右边界
* @return
*/
private String getNearWriteNode(List<String> childPath,Integer index){
for (int i = 0; i < index; i++) {
String node = childPath.get(i);
if (node.contains(WRITE_NAME))
return node;
}
return null;
}
}
//arg1:CuratorFramework连接对象,arg2:节点路径
lock=new InterProcessMutex(client,path);
//获取锁
lock.acquire();
//释放锁
lock.release();
分布式锁
获取。