Zookeeper 实现分布式锁
作者：互联网
ZooKeeper 实现分布式锁
利用 Watcher 机制和 ZooKeeper EPHEMERAL_SEQUENTIAL 节点的特点，实现分布式锁。
实现原理：EPHEMERAL_SEQUENTIAL 该类节点具有顺序递增特点，不会持久化到磁盘，在线程执行完毕后，会自动删除。
Watcher 机制使得在节点被删除时，能够获得通知，并且能接收到被 push 过来的消息。
分布式锁：在多进程环境下，公共的资源可能会被不同的进程占用，为了防止多个线程同时更改数据，需要为该资源加锁，当一个线程占用后，其他线程必须得等待。
具体实现思路为：
public class ZooDistributeLock implements Watcher {
private static final Logger LOG = LoggerFactory.getLogger(ZooDistributeLock.class);
private static final String LOCK_PATH = "/lock";
// 模拟开启的线程数
private static final int THREAD_NUM = 5;
// 用于等待所有线程都连接成功后再执行任务
private static CountDownLatch startFlag = new CountDownLatch(1);
// 用于确保所有线程执行完毕
private static CountDownLatch threadFlag = new CountDownLatch(THREAD_NUM);
private ZooKeeper zk = null;
private String currentPath;
private String lockPath;
public static void main(String[] args) {
for (int i = 0; i < THREAD_NUM; i++) {
new Thread() {
@Override
public void run() {
ZooDistributeLock zooDistributeLock = new ZooDistributeLock();
try {
zooDistributeLock.connection();
zooDistributeLock.createNode();
zooDistributeLock.getLock();
} catch (IOException | InterruptedException | KeeperException e) {
e.printStackTrace();
}
}
}.start();
}
try {
threadFlag.await();
LOG.info("所有线程执行完毕...");
} catch (InterruptedException e) {
LOG.error(e.getMessage(), e);
}
}
@Override
public void process(WatchedEvent event) {
Event.KeeperState state = event.getState();
Event.EventType type = event.getType();
if (Event.KeeperState.SyncConnected == state) {
if (Event.EventType.None == type) {
// 标识连接成功
LOG.info("成功连接上ZK服务器");
startFlag.countDown();
}
if (Event.EventType.NodeDeleted == type && event.getPath().equals(this.lockPath)) {
LOG.info("节点" + this.lockPath + "的锁已经被释放");
try {
// 上一个节点释放了,当前节点去获取锁
getLock();
} catch (KeeperException | InterruptedException e) {
LOG.error(e.getMessage(), e);
}
}
}
}
/**
* 连接到 ZK
*
* @throws IOException
*/
private void connection() throws IOException, InterruptedException {
zk = new ZooKeeper("127.0.0.1:2181", 5000, this);
// 等待连接成功后再执行下一步操作
startFlag.await();
}
// 创建节点,并初始化当前路径
private void createNode() throws KeeperException, InterruptedException, UnsupportedEncodingException {
this.currentPath = this.zk.create(LOCK_PATH, "".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode
.EPHEMERAL_SEQUENTIAL);
}
private void getLock() throws KeeperException, InterruptedException {
if (minNode()) {
doSomething();
// 释放锁
releaseLock();
}
}
/**
* 当前是否为最小节点
*
* @return
*/
private boolean minNode() {
// 当前序号
try {
initLockPath();
// 判断前一个节点存在不存在,如果存在,则表示当前节点不是最小节点
// zk.getData(this.lockPath, this, new Stat());
zk.getData(this.lockPath, true, new Stat());
LOG.info(this.currentPath + " 不是最小值,没有获取锁,等待 " + this.lockPath + " 释放锁");
return false;
} catch (KeeperException e) {
LOG.info(this.currentPath + " 是最小值,获得锁");
return true;
} catch (InterruptedException e) {
LOG.error(e.getMessage(), e);
}
return true;
}
private void doSomething() {
LOG.info("处理业务逻辑...");
}
/**
* 释放锁并关闭连接
*
* @throws KeeperException
* @throws InterruptedException
*/
private void releaseLock() throws KeeperException, InterruptedException {
Thread.sleep(2000);
if (this.zk != null) {
LOG.info(this.currentPath + " 业务处理完毕,释放锁...");
zk.delete(this.currentPath, -1);
this.zk.close();
LOG.info(Thread.currentThread().getName() + "关闭 zookeeper 连接");
}
threadFlag.countDown();
}
/**
* 初始化 lockpath
*/
private void initLockPath() {
int currentSeq = Integer.parseInt(this.currentPath.substring(LOCK_PATH.length()));
// 上一个序号
int preSeq = currentSeq - 1;
String preSeqStr = String.valueOf(preSeq);
while (preSeqStr.length() < 10) {
preSeqStr = "0" + preSeqStr;
}
this.lockPath = LOCK_PATH + preSeqStr;
}
@Override
public String toString() {
return "ZooDistributeLock{" +
"zk=" + zk +
", currentPath='" + currentPath + '\'' +
", lockPath='" + lockPath + '\'' +
'}';
}
}
控制台打印结果：
2017-07-10 14:53:13 [ Thread-0-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13 [ Thread-3-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13 [ Thread-4-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13 [ Thread-2-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13 [ Thread-1-EventThread:60 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:73) 成功连接上ZK服务器
2017-07-10 14:53:13 [ Thread-1:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000330 不是最小值,没有获取锁,等待 /lock0000000329 释放锁
2017-07-10 14:53:13 [ Thread-3:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000329 不是最小值,没有获取锁,等待 /lock0000000328 释放锁
2017-07-10 14:53:13 [ Thread-2:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000328 不是最小值,没有获取锁,等待 /lock0000000327 释放锁
2017-07-10 14:53:13 [ Thread-4:78 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:130) /lock0000000331 不是最小值,没有获取锁,等待 /lock0000000330 释放锁
2017-07-10 14:53:13 [ Thread-0:80 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000327 是最小值,获得锁
2017-07-10 14:53:13 [ Thread-0:80 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:15 [ Thread-0:2082 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000327 业务处理完毕,释放锁...
2017-07-10 14:53:15 [ Thread-0:2087 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e018b closed
2017-07-10 14:53:15 [ Thread-2-EventThread:2088 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000327的锁已经被释放
2017-07-10 14:53:15 [ Thread-0:2088 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-0关闭 zookeeper 连接
2017-07-10 14:53:15 [ Thread-2-EventThread:2089 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000328 是最小值,获得锁
2017-07-10 14:53:15 [ Thread-2-EventThread:2089 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:15 [ Thread-0-EventThread:2090 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e018b
2017-07-10 14:53:17 [ Thread-2-EventThread:4094 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000328 业务处理完毕,释放锁...
2017-07-10 14:53:17 [ Thread-3-EventThread:4097 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000328的锁已经被释放
2017-07-10 14:53:17 [ Thread-2-EventThread:4098 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e018c closed
2017-07-10 14:53:17 [ Thread-3-EventThread:4098 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000329 是最小值,获得锁
2017-07-10 14:53:17 [ Thread-3-EventThread:4098 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:17 [ Thread-2-EventThread:4098 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-2-EventThread关闭 zookeeper 连接
2017-07-10 14:53:17 [ Thread-2-EventThread:4098 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e018c
2017-07-10 14:53:19 [ Thread-3-EventThread:6103 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000329 业务处理完毕,释放锁...
2017-07-10 14:53:19 [ Thread-1-EventThread:6106 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000329的锁已经被释放
2017-07-10 14:53:19 [ Thread-3-EventThread:6107 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e0189 closed
2017-07-10 14:53:19 [ Thread-1-EventThread:6107 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000330 是最小值,获得锁
2017-07-10 14:53:19 [ Thread-3-EventThread:6107 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-3-EventThread关闭 zookeeper 连接
2017-07-10 14:53:19 [ Thread-1-EventThread:6108 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:19 [ Thread-3-EventThread:6108 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e0189
2017-07-10 14:53:21 [ Thread-1-EventThread:8111 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000330 业务处理完毕,释放锁...
2017-07-10 14:53:21 [ Thread-4-EventThread:8115 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.process(ZooDistributeLock.java:78) 节点/lock0000000330的锁已经被释放
2017-07-10 14:53:21 [ Thread-1-EventThread:8116 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e018a closed
2017-07-10 14:53:21 [ Thread-1-EventThread:8116 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-1-EventThread关闭 zookeeper 连接
2017-07-10 14:53:21 [ Thread-4-EventThread:8116 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.minNode(ZooDistributeLock.java:133) /lock0000000331 是最小值,获得锁
2017-07-10 14:53:21 [ Thread-4-EventThread:8116 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.doSomething(ZooDistributeLock.java:142) 处理业务逻辑...
2017-07-10 14:53:21 [ Thread-1-EventThread:8116 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e018a
2017-07-10 14:53:23 [ Thread-4-EventThread:10119 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:154) /lock0000000331 业务处理完毕,释放锁...
2017-07-10 14:53:23 [ Thread-4-EventThread:10123 ] - [ INFO ] org.apache.zookeeper.ZooKeeper.close(ZooKeeper.java:684) Session: 0x15d1c5a431e0188 closed
2017-07-10 14:53:23 [ Thread-4-EventThread:10124 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.releaseLock(ZooDistributeLock.java:157) Thread-4-EventThread关闭 zookeeper 连接
2017-07-10 14:53:23 [ Thread-4-EventThread:10124 ] - [ INFO ] org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:519) EventThread shut down for session: 0x15d1c5a431e0188
2017-07-10 14:53:23 [ main:10124 ] - [ INFO ] qingxiang.app.zoo.ZooDistributeLock.main(ZooDistributeLock.java:58) 所有线程执行完毕...
标签：INFO,10,EventThread,Thread,实现,ZooDistributeLock,Zookeeper,53,分布式 来源： https://blog.csdn.net/u013452337/article/details/100280712