ZooKeeper实现读写锁
1 读写锁的概念
读写锁是计算机程序的并发控制的一种同步机制，用于解决读写问题，读操作可并发重入，写操作是互斥的。 读写锁有多种读写权限的优先级策略，可以设计为读优先、写优先或不指定优先级。
- 读优先：允许最大并发的读操作，但可能会饿死写操作；因为写操作必须在没有任何读操作的时候才能够执行。
- 写优先：只要排队队列中有写操作，读操作就必须等待；
- 不指定优先级：对读操作和写操作不做任何优先级的假设
不指定优先级的策略，最适合使用ZooKeeper的子节点模式来实现，今天就来尝试这种策略。
2 锁设计
同前面介绍的普通分布式锁，也使用子节点模式实现。先用容器模式（CreateMode.CONTAINER）创建唯一的锁节点，每个锁客户端在锁节点下使用临时循序模式（CreateMode. SEQUENTIAL）创建子节点。这些子节点会自动在名称后面追加10位数字。
2.1 如何标识读锁还是写锁？
有两种简单的方案：在子节点名中标识、在节点的值中标识。如果采用在值中标识，每次子节点列表后，还需要再分别读一下子节点的值，才能判断是读锁还是写锁，会比较耗时。如果在子节点名称中标识，会面临一个问题：在同一个节点中创建的子节点，如果给定的名称不同，追加的10位数字是否仍然是递归的？
写个测试用例验证一下。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
|
测试用例通过，说明在同一个Container中创建的子节点，不论提供的节点名是什么，后续追加的10位数字都是顺序递增的。这样，就可以使用节点名来区分读锁和写锁。
2.2 类设计
介绍分布式锁的时候，已经创建了阻塞锁 ChildrenBlockingLock，读写锁正好可以基于这个类做重载。
2.3 获取锁的逻辑
写锁是一个独占锁，逻辑跟普通分布式锁相同，只要它之前有锁就必须等待。所以，完全沿用阻塞锁的逻辑即可。
读锁允许并发，它之前可以有任意读锁，但不能有写锁。所以只需要判断有没有写锁即可。
3 关键代码
3.1 ChildrenNodeLock.java
这个类，主要是增加了一个获取排序后子节点列表的方法，这样方便实现读写锁的代码。当然，这个操作会增加一些耗时，如果子节点数量太大，可能不适用。
首先定义一个函数，用来返回子节点的前缀
/** 子节点的前缀，缺省是element，子类可以重载 */
protected String getChildPrefix() {
return "element";
}
然后定义一个内部类，子节点排序时会用到
/** 子节点名称比较 */
private class StringCompare implements Comparator<String> {
@Override
public int compare(String string1, String string2) {
return string1.substring(string1.length() - 10)
.compareTo(string2.substring(string2.length() - 10));
}
}
最后实现子节点排序方法，用于代替 getChildren 函数
/** 获取排好序的子节点列表 */
final public List<String> getOrderedChildren(String path, boolean watch)
throws KeeperException, InterruptedException {
List<String> children = getZooKeeper().getChildren(path, watch);
Collections.sort(children, new StringCompare());
return children;
}
3.2 ChildrenBlockingLock.java
在多客户端随机测试时，经常出现程序卡死的情况，无法正常退出。经过添加日志跟踪，发现WatchedEvent可能会丢失，也可能会发送给并不是注册事件的ZooKeeper客户端。在网上搜索，发现很多人也碰到类似问题。
简单修改了一下ChildrenBlockingLock#isLockSuccess等待信号的代码，从无参数的死等变成设置一定超时时间等待。关键代码如下
protected boolean isLockSuccess() {
boolean lockSuccess;
try {
while (true) {
String prevElementName = getPrevElementName();
if (prevElementName == null) {
log.trace("{} 没有更靠前的子节点，加锁成功", elementNodeName);
lockSuccess = true;
break;
} else {
// 有更小的节点，说明当前节点没抢到锁，注册前一个节点的监听。
log.trace("{} 监控 {} 的事件", elementNodeName, prevElementName);
getZooKeeper().exists(this.guidNodeName + "/" + prevElementName, true);
synchronized (mutex) {
// 等待最多一秒
mutex.wait(1000);
log.trace("{} 监控的 {} 有子节点变化", elementNodeName, guidNodeName);
}
}
}
} catch (KeeperException e) {
lockSuccess = false;
} catch (InterruptedException e) {
lockSuccess = false;
}
return lockSuccess;
}
3.3 写锁 ZooKeeperWriteLock.java
代码基本是沿用父类，只需要重载getChildPrefix()方法，
/** 返回写锁的前缀 */
protected String getChildPrefix() {
return "w-lock-";
}
3.4 读锁 ZooKeeperReadLock.java
同写锁相比，除了重载getChildPrefix()方法，还重载了getPrevElementName()用来查找最近一个写锁。
/** 返回读锁的前缀 */
protected String getChildPrefix() {
return "r-lock-";
}
/** 是写锁 */
private boolean isWriteLock(String elementName) {
return elementName.startsWith(ZooKeeperWriteLock.FLAG);
}
/** 读取前一个写锁 */
protected String getPrevElementName() throws KeeperException, InterruptedException {
List<String> elementNames = super.getOrderedChildren(this.guidNodeName, false);
super.traceOrderedChildren(this.guidNodeName, elementNames);
String prevWriteElementName = null;
for (String oneElementName : elementNames) {
if (this.elementNodeFullName.endsWith(oneElementName)) {
// 已经到了当前节点
break;
}
if (isWriteLock(oneElementName)) {
prevWriteElementName = oneElementName;
}
}
return prevWriteElementName;
}
4 测试用例
测试用例没想到好的判断方法，很难使用assert判断结果，因此做了简化，根据日志输出，靠人眼判断是否正确。
4.1 测试线程类
分别为都锁和写锁构建了两个内部类
/** 写锁线程 */
class WriteLockClient extends Thread {
ZooKeeperWriteLock writeLock;
public WriteLockClient() {
try {
this.writeLock = new ZooKeeperWriteLock(address);
} catch (IOException e) {
}
}
public void run() {
writeLock.lock(guidNodeName, this.getName());
try {
Thread.sleep(1000 + random.nextInt(20) * 100);
} catch (InterruptedException e) {
}
writeLock.release(guidNodeName, this.getName());
}
}
/** 读锁线程 */
class ReadLockClient extends Thread {
ZooKeeperReadLock readLock;
public ReadLockClient() {
try {
this.readLock = new ZooKeeperReadLock(address);
} catch (IOException e) {
}
}
public void run() {
readLock.lock(guidNodeName, this.getName());
try {
Thread.sleep(1000 + random.nextInt(20) * 100);
} catch (InterruptedException e) {
}
readLock.release(guidNodeName, this.getName());
try {
readLock.getZooKeeper().close();
} catch (InterruptedException e) {
}
}
}
4.2 读-读锁测试
代码
@Test
public void testReadRead() throws IOException, InterruptedException {
ReadLockClient readLock1 = new ReadLockClient();
ReadLockClient readLock2 = new ReadLockClient();
readLock1.start();
readLock2.start();
readLock1.join();
readLock2.join();
}
测试结果可以看到，两个读锁并发执行
22:18.861 [Thread-2 INFO] r-lock-0000000000 get read lock : true 22:18.865 [Thread-1 INFO] r-lock-0000000001 get read lock : true 22:20.065 [Thread-2 INFO] r-lock-0000000000 release read lock 22:21.366 [Thread-1 INFO] r-lock-0000000001 release read lock
4.3 读-写锁测试
代码
@Test
public void testReadWrite() throws IOException, InterruptedException {
ReadLockClient readLock1 = new ReadLockClient();
WriteLockClient writeLock1 = new WriteLockClient();
readLock1.start();
Thread.sleep(50);
writeLock1.start();
readLock1.join();
writeLock1.join();
}
测试结果可以看到，首先获取读锁，释放之后才获取到写锁。
27:40.800 [Thread-1 INFO] r-lock-0000000000 get read lock : true 27:43.310 [Thread-1 INFO] r-lock-0000000000 release read lock 27:43.423 [Thread-2 INFO] w-lock-0000000001 get write lock : true 27:44.423 [Thread-2 INFO] w-lock-0000000001 release write lock
4.4 写-读锁测试
代码
@Test
public void testWriteRead() throws IOException, InterruptedException {
ReadLockClient readLock1 = new ReadLockClient();
WriteLockClient writeLock1 = new WriteLockClient();
writeLock1.start();
Thread.sleep(50);
readLock1.start();
writeLock1.join();
readLock1.join();
}
测试结果可以看到，首先获取写锁，释放之后才获取到读锁。
29:17.661 [Thread-2 INFO] w-lock-0000000000 get write lock : true 29:19.966 [Thread-2 INFO] w-lock-0000000000 release write lock 29:19.976 [Thread-1 INFO] r-lock-0000000001 get read lock : true 29:22.476 [Thread-1 INFO] r-lock-0000000001 release read lock
4.5 多客户端随机读写锁测试
测试代码
@Test
public void testRandomReadWriteLock() throws IOException, InterruptedException {
int threadCount = 20;
Thread[] lockThreads = new Thread[threadCount];
for (int i = 0; i < threadCount; i++) {
// 一定概率是写锁
boolean writeLock = random.nextInt(5) == 0;
if (writeLock) {
lockThreads[i] = new WriteLockClient();
} else {
lockThreads[i] = new ReadLockClient();
}
lockThreads[i].start();
}
for (int i = 0; i < threadCount; i++) {
lockThreads[i].join();
}
}
测试结果可以看出，如果连续多个读锁会并发执行。为了方便查看，我添加了一些横线分隔。
30:31.317 [Thread-1 INFO] w-lock-0000000000 get write lock : true 30:32.824 [Thread-1 INFO] w-lock-0000000000 release write lock ------------------------------------------------------------------ 30:32.834 [Thread-17 INFO] r-lock-0000000004 get read lock : true 30:32.835 [Thread-19 INFO] r-lock-0000000002 get read lock : true 30:32.835 [Thread-20 INFO] r-lock-0000000001 get read lock : true 30:32.836 [Thread-18 INFO] r-lock-0000000003 get read lock : true 30:34.135 [Thread-20 INFO] r-lock-0000000001 release read lock 30:34.634 [Thread-17 INFO] r-lock-0000000004 release read lock 30:34.935 [Thread-19 INFO] r-lock-0000000002 release read lock 30:35.036 [Thread-18 INFO] r-lock-0000000003 release read lock ------------------------------------------------------------------ 30:35.053 [Thread-16 INFO] w-lock-0000000005 get write lock : true 30:36.154 [Thread-16 INFO] w-lock-0000000005 release write lock ------------------------------------------------------------------ 30:36.160 [Thread-14 INFO] r-lock-0000000007 get read lock : true 30:36.160 [Thread-15 INFO] r-lock-0000000006 get read lock : true 30:38.160 [Thread-14 INFO] r-lock-0000000007 release read lock 30:38.661 [Thread-15 INFO] r-lock-0000000006 release read lock ------------------------------------------------------------------ 30:38.669 [Thread-13 INFO] w-lock-0000000008 get write lock : true 30:39.969 [Thread-13 INFO] w-lock-0000000008 release write lock ------------------------------------------------------------------ 30:39.976 [Thread-12 INFO] r-lock-0000000009 get read lock : true 30:39.977 [Thread-8 INFO] r-lock-0000000014 get read lock : true 30:39.977 [Thread-6 INFO] r-lock-0000000015 get read lock : true 30:39.984 [Thread-10 INFO] r-lock-0000000011 get read lock : true 30:39.985 [Thread-3 INFO] r-lock-0000000018 get read lock : true 30:39.984 [Thread-7 INFO] r-lock-0000000013 get read lock : true 30:39.984 [Thread-11 INFO] r-lock-0000000010 get read lock : true 30:39.983 [Thread-9 INFO] r-lock-0000000012 get read lock : true 30:39.983 [Thread-2 INFO] r-lock-0000000019 get read lock : true 30:39.982 [Thread-5 INFO] r-lock-0000000016 get read lock : true 30:39.986 [Thread-4 INFO] r-lock-0000000017 get read lock : true 30:40.986 [Thread-3 INFO] r-lock-0000000018 release read lock 30:41.086 [Thread-2 INFO] r-lock-0000000019 release read lock 30:41.285 [Thread-6 INFO] r-lock-0000000015 release read lock 30:41.576 [Thread-12 INFO] r-lock-0000000009 release read lock 30:42.185 [Thread-10 INFO] r-lock-0000000011 release read lock 30:42.186 [Thread-5 INFO] r-lock-0000000016 release read lock 30:42.187 [Thread-11 INFO] r-lock-0000000010 release read lock 30:42.286 [Thread-9 INFO] r-lock-0000000012 release read lock 30:42.586 [Thread-7 INFO] r-lock-0000000013 release read lock 30:42.677 [Thread-8 INFO] r-lock-0000000014 release read lock 30:42.887 [Thread-4 INFO] r-lock-0000000017 release read lock
