一、思路
锁的原语一般有两个：
①加锁
②解锁
加锁：
加锁的一般算法步骤是：
1.尝试加锁
2.如果锁没有被占用，则加锁成功
3.如果锁被占用，则等待锁被释放
4.锁被释放后，收到锁释放通知，重复步骤 1
翻译成ZooKeeper的算法步骤就是：
1.尝试创建表示锁的临时节点
2.如果创建节点成功，则加锁成功
3.如果创建节点失败，则创建一个锁节点的监视器，等待锁节点的删除通知
4.锁节点被持有者删除后，收到锁节点的删除通知，重复步骤 1
解锁：
解锁的算法步骤是：
1.锁节点是否存在
2.如果不存在，完成解锁
3.如果锁节点存在，则判断锁节点的数据（ 锁的持有者 ）是否和解锁者相同
4.如果一样，则删除锁节点，完成解锁
关注公众号： “Java不睡觉”， 回复：“资源”。获取大数据全套视频和大量Java书籍
所以根据以上思路，我们写如下的示例程序：
二、代码
2.1 锁实体类
package com.zhb.zookeeper.lock;
public class LockInfo {
// 锁的名字，体现为zookeeper上的节点名
String lockname;
// 锁的持有者，体现为zookeeper锁节点的数据
String lockOwner;
public LockInfo(String lockname, String lockOwner) {
this.lockname = lockname;
this.lockOwner = lockOwner;
}
public String getLockname() {
return lockname;
}
public void setLockname(String lockname) {
this.lockname = lockname;
}
public String getLockOwner() {
return lockOwner;
}
public void setLockOwner(String lockOwner) {
this.lockOwner = lockOwner;
}
@Override
public String toString() {
return "LockInfo [lockname=" + lockname + ", lockOwner=" + lockOwner + "]";
}
}
2.2 锁接口
package com.zhb.zookeeper.lock;
import org.apache.zookeeper.KeeperException;
public interface MyLock {
/**
* 尝试加锁,对应加锁步骤1
*
* @return whether success or not, true: lock success, false: lock failed
*/
public boolean tryLock() throws KeeperException, InterruptedException;
/**
* 同步加锁，对应加锁步骤3
*/
public void lock()throws KeeperException, InterruptedException;
/**
* 解锁，释放锁
*/
public void unlock();
}
2.3 锁的实现类
package com.zhb.zookeeper.lock;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
public class ZkLock implements MyLock {
private String LockName = "dis_lock";
private LockInfo lockInfo;
// 此变量的目的是确保连接上zookeeper Server
private CountDownLatch latch = new CountDownLatch(1);
private String connectString = "127.0.0.1:2181";
private int sessionTimeout = 60 * 1000;
// zookeeperk客户端
private ZooKeeper zooKeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
if (KeeperState.SyncConnected == event.getState() && EventType.None == event.getType()) {
latch.countDown();
}
}
});
// 构造函数
public ZkLock(LockInfo lockInfo) throws IOException, InterruptedException {
this.lockInfo = lockInfo;
latch.await();
}
@Override
public boolean tryLock() {
try {
if (zooKeeper.exists(lockInfo.getLockname(), false) != null) {
String currentOwner = new String(zooKeeper.getData(lockInfo.getLockname(), false, null));
System.out.println("已经被加上锁了，锁的持有者是：" + currentOwner);
return false;
} else {
zooKeeper.create(lockInfo.lockname, lockInfo.lockOwner.getBytes(), Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
return true;
}
} catch (KeeperException e) {
} catch (InterruptedException e) {
e.printStackTrace();
System.out.println("####################222222222222222####################");
}
return false;
}
@Override
public void lock() {
if (!tryLock()) {
CountDownLatch releaseSignal = new CountDownLatch(1);
try {
zooKeeper.exists(lockInfo.lockname, new Watcher() {
@Override
public void process(WatchedEvent event) {
if (lockInfo.getLockname().equals(event.getPath())
&& EventType.NodeDeleted.equals(event.getType())) {
releaseSignal.countDown();
}
}
});
releaseSignal.await();
// 递归调用自己
lock();
} catch (KeeperException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
} else {
System.out.println(lockInfo.lockOwner + ":上锁成功");
}
}
@Override
public void unlock() {
try {
if (zooKeeper.exists(lockInfo.lockname, false) != null) {
String existOwner = new String(zooKeeper.getData(lockInfo.lockname, null, null), "UTF-8");
if (lockInfo.lockOwner.equals(existOwner)) {
zooKeeper.delete(lockInfo.lockname, -1);
System.out.println(lockInfo.lockOwner + ": 解锁成功");
} else {
System.out.println(lockInfo.lockOwner + ":无法释放锁，因为没有获得锁");
}
}
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
2.4 测试类
package com.zhb.zookeeper.lock;
import java.io.IOException;
import org.apache.zookeeper.KeeperException;
public class LockTest {
public static void main(String[] args) throws IOException, InterruptedException {
LockInfo lockInfo1 = new LockInfo("/zk-lock", "person1");
LockInfo lockInfo2 = new LockInfo("/zk-lock", "person2");
LockInfo lockInfo3 = new LockInfo("/zk-lock", "person3");
ZkLock zLock1 = new ZkLock(lockInfo1);
ZkLock zLock2 = new ZkLock(lockInfo2);
ZkLock zLock3 = new ZkLock(lockInfo3);
new Thread(new Runnable() {
@Override
public void run() {
zLock1.lock();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
zLock2.lock();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
zLock3.lock();
}
}).start();
//停3秒钟
Thread.sleep(3000);
new Thread(new Runnable() {
@Override
public void run() {
zLock1.unlock();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
zLock2.unlock();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
zLock3.unlock();
}
}).start();
}
}
2.5运行结果
首先开启zookeeper服务：
然后运行测试类，结果如下：
使用zookeeper客户端查看是否正确：
正确
四、总结
本文只是zookeeper分布式锁的基础版，他的缺点是：
1.排它锁的粒度大，没有区分读、写操作，如果读多写少，则十分影响性能
2.羊群效应：锁释放后会通知所有等待中的 ZooKeeper 客户端，然后同时发起加锁请求，瞬时压力很大。
解决方案是：
1.缩小通知范围：等待锁的小伙伴们按先来后到的顺序排队吧，排好队了，接下来我只需要关心我前面一个节点的状态，当前一个节点被释放，我再去抢锁。
2.缩小锁的粒度：锁不关心业务，但是可以简单地通过操作的读、写性质来二分锁的粒度：
读锁：又称共享锁，如果前面没有写节点，可以直接上锁；当前面有写节点时，则等待距离自己最近的写节点释放（ 删除 ）。
写锁：如果前面没有节点，可以直接上锁；如果前面有节点，则等待前一个节点释放（ 删除 ）。
关注公众号： “Java不睡觉”， 回复：“资源”。获取大数据全套视频和大量Java书籍
京公网安备 11010502036488号