排它锁概念:
Exclusive Locks,被称为X锁,写锁,独占锁.如果事物T1对数据对象O1加上了排它锁,那么在整个加锁期间,只允许事务T1对O1进行读写操作,其他事务必须等到T1释放锁后才能进行操作.在单机环境中,JDK提供了synchronized关键字和ReentrantLock
重用锁来提供排它锁的功能.
zookeeper实现排它锁原理:
在需要获取排它锁时,所有的客户端都会调用create方法在固定路径下创建节点,并发环境下,只有一个客户端可以创建成功,相当于获取了锁,当该客户端完成事务后,删除该节点.对于没有获取成功的其他节点,则在该路径上设置监听,如果该路径子节点删除事件触发,继续尝试获取锁(create).
具体实现:
定义锁的接口:(我想实现的几个方法)
void lock();
boolean isLocked();
boolean tryLock();
boolean tryLock(long timeout);
void unlock();
直接上代码：
DistributedLock类中拥有的成员变量：
private static CuratorFramework client = null;
private static Logger logger = Logger.getLogger(DistributedLock.class);
protected static CountDownLatch latch = new CountDownLatch(1);
client：为了之后操作zk设置
logger：打印测试日志
latch：lock中要循环等待获取锁，属于一方释放锁和其他再次尝试获取锁之间的沟通桥梁
初始化方法：
public static synchronized void init(String connectString) {
if (client != null)
return;
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
client = CuratorFrameworkFactory.builder().connectString(connectString)
.sessionTimeoutMs(10000).retryPolicy(retryPolicy)
.namespace("LockService").build();
client.start();
// 创建锁目录
try {
if (client.checkExists().forPath("/ExclusiveLockDemo") == null) {
client.create().creatingParentsIfNeeded()
.withMode(CreateMode.PERSISTENT)
.withACL(Ids.OPEN_ACL_UNSAFE)
.forPath("/ExclusiveLockDemo");
}
// 创建锁监听
addChildWatcher("/ExclusiveLockDemo");
} catch (Exception e) {
logger.error("ZK服务器连接不上");
throw new RuntimeException("ZK服务器连接不上");
}
}init方法里做了以下几个事：
1.初始化client
2.创建根目录
3.加根目录的子节点监听（为了unlock）
监听处理函数：
private static void addChildWatcher(String path) throws Exception {
final PathChildrenCache cache = new PathChildrenCache(client, path,
true);
cache.start(StartMode.POST_INITIALIZED_EVENT);
cache.getListenable().addListener(new PathChildrenCacheListener() {
public void childEvent(CuratorFramework client,
PathChildrenCacheEvent event) throws Exception {
if (event.getType().equals(
PathChildrenCacheEvent.Type.INITIALIZED)) {
} else if (event.getType().equals(
PathChildrenCacheEvent.Type.CHILD_ADDED)) {
} else if (event.getType().equals(
PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
String path = event.getData().getPath();
System.out.println("收到监听"+path);
if(path.contains("ExclusiveLockDemo")){
logger.info("排他锁,收到锁释放通知");
latch.countDown();
}
} else if (event.getType().equals(
PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
}
}
});
}主要监听的是子节点的删除行为，当子节点lock删除时，就代表着有节点释放了锁，同时，应该通知等待获取锁的client发起新一轮的获取锁行为，这里用latch控制，latch.countDown()就是通知他们进行下一轮获取，具体代码可以看lock函数的实现。
lock函数：
public void lock() {
while (true) {
try {
client.create().creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL)
.withACL(Ids.OPEN_ACL_UNSAFE)
.forPath("/ExclusiveLockDemo/lock");
logger.info("成功获取到锁");
return;// 如果节点创建成功，即说明获取锁成功
} catch (Exception e) {
logger.info("此次获取锁没有成功");
try {
//如果没有获取到锁，需要重新设置同步资源值
if(latch.getCount()<=0){
latch = new CountDownLatch(1);
}
latch.await();
} catch (InterruptedException e1) {
e1.printStackTrace();
logger.error("", e1);
}
}
}
}lock函数：尝试在 “/ExclusiveLockDemo/lock”创建znode，当有多个客户端同时调用lock()时，有且只有一个client可以成功创建，打印“成功获取到锁”，其他client将会进入catch语句，打印“此次获取没有成功”，然后重置latch，进入阻塞状态（latch.await())。唤醒条件是持有锁的client释放锁。因为是阻塞的获取锁，所以整个函数处于while（true）死循环里。直到某个时间点成功创建锁，方可return。
islock（）函数：
public boolean isLocked() {
try {
Stat stat = client.checkExists().forPath("/ExclusiveLockDemo/lock");
return stat==null?false:true;
} catch (Exception e) {
e.printStackTrace();
}
return false;
}这个比较简单，就是判断一下制定path是否存在znode，也就是是否已经有client成功获取到锁并且还没有释放。
unlock函数：
public void unlock() {
try {
if (client.checkExists().forPath("/ExclusiveLockDemo/lock") != null) {
client.delete().forPath("/ExclusiveLockDemo/lock");
}
} catch (Exception e) {
e.printStackTrace();
}
}这个也比较简单，只是删除锁对应的znode节点即可，但是有个很严重的问题，这种删除太草率，你可以释放一个原本不是你获取的锁，这是不符合道理的，只有你获取的锁，你才有资格释放它，所以正规的写法应该是获取锁的时候加权限，然后释放的时候先检验权限，不是你的不能删。
public boolean tryLock() {
try {
client.create().creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL)
.withACL(Ids.OPEN_ACL_UNSAFE)
.forPath("/ExclusiveLockDemo/lock");
logger.info("成功获取到锁");
} catch (Exception e) {
logger.info("获取锁失败");
return false;
}
return true;
}仿照可重入锁的trylock，尝试获取锁，非阻塞，能获取就返回true，被占用就返回false
public boolean tryLock(long timeout) {
if(timeout<= 0L) return false;
final long deadline = System.currentTimeMillis() + timeout;
for(;;){
if(tryLock())
return true;
else{
timeout = deadline - System.currentTimeMillis();
if (timeout <= 0L)
return false;
}
}
}有获取时间限制的trylock，实现起来还是比较简单的，大家一看都能看懂