zookeeper实现分布式锁

ysoren

通过在zookeeper的主目录下创建临时自增子节点，我们可以实现分布式锁功能，代码如下

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
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;
import org.apache.zookeeper.data.Stat;
public class Distributedlock {
// 超时时间
private static final int SESSION_TIMEOUT = 5000;
// zookeeper server列表
private String hosts = "localhost:2181,localhost:2182,localhost:2183";
//根目录
private String groupNode = "locks";
//子目录
private String subNode = "sub";
private ZooKeeper zk;
// 当前client创建的子节点
private String thisPath;
// 当前client等待的子节点
private String waitPath;
//闭锁
private CountDownLatch latch = new CountDownLatch(1);
/**
* 连接zookeeper
*/
public void connectZookeeper() throws Exception {
zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {
public void process(WatchedEvent event) {
try {
// 监听连接建立, 打开latch, 唤醒wait在该latch上的线程
if (event.getState() == KeeperState.SyncConnected) {
latch.countDown();
}
// 发生了waitPath的删除事件,当前线程获取锁
if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {
getLock();
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
// 等待连接建立
latch.await();
// 创建临时自增子节点
thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
// wait
Thread.sleep(10);
//得到子节点
List<String> childrenNodes = zk.getChildren("/" + groupNode, false);
// 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁
if (childrenNodes.size() == 1)
{
//得到锁
getLock();
}
else
{
//得到当前节点
String thisNode = thisPath.substring(("/" + groupNode + "/").length());
// 排序
Collections.sort(childrenNodes);
//得到当前节点在子节点的索引
int index = childrenNodes.indexOf(thisNode);
if (index == -1)
{
// never happened
}
else if (index == 0)
{
// inddx == 0, 说明thisNode在列表中最小, 当前client获得锁
getLock();
}
else
{
// 获得排名比thisPath前1位的节点
this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);
// 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法
zk.getData(waitPath, true, new Stat());
}
}
}
private void getLock() throws Exception {
try
{
System.out.println("gain lock: " + thisPath);
Thread.sleep(2000);
}
finally
{
System.out.println("finished: " + thisPath);
// 将thisPath删除, 监听thisPath的client将获得通知
// 相当于释放锁
zk.delete(this.thisPath, -1);
}
}
public static void main(String[] args) throws Exception {
//生成10个子节点
for (int i = 0; i < 10; i++) {
new Thread() {
public void run() {
try {
Distributedlock dl = new Distributedlock();
dl.connectZookeeper();
} catch (Exception e) {
e.printStackTrace();
}
}
}.start();
}
Thread.sleep(Long.MAX_VALUE);
}
}


运行结果如下
gain lock: /locks/sub0000000000
finished: /locks/sub0000000000
gain lock: /locks/sub0000000001
finished: /locks/sub0000000001
gain lock: /locks/sub0000000002
finished: /locks/sub0000000002
gain lock: /locks/sub0000000003
finished: /locks/sub0000000003
gain lock: /locks/sub0000000004
finished: /locks/sub0000000004
gain lock: /locks/sub0000000005
finished: /locks/sub0000000005
gain lock: /locks/sub0000000006
finished: /locks/sub0000000006
gain lock: /locks/sub0000000007
finished: /locks/sub0000000007
gain lock: /locks/sub0000000008
finished: /locks/sub0000000008
gain lock: /locks/sub0000000009
finished: /locks/sub0000000009