基于Zookeeper实现分布式锁
程序员文章站
2022-05-07 14:29:12
...
锁的基概
锁：通过锁可以实现在多个线程或多个进程间在争抢资源时，能够合理的分配置资源的所有权。在单体应用中我们可以通过 synchronized 或ReentrantLock 来实现锁
分布式锁：
但在分布式系统中，仅仅是加synchronized 是不够的，需要借助第三组件来实现。比如一些简单的做法是使用 关系型数据行级锁来实现不同进程之间的互斥，但大型分布式系统的性能瓶颈往往集中在数据库操作上。为了提高性能得采用如Redis、Zookeeper之内的组件实现分布式锁。
共享锁：也称作只读锁，当一方获得共享锁之后，其它方也可以获得共享锁。但其只允许读取。在共享锁全部释放之前，其它方不能获得写锁。
排他锁：也称作读写锁，获得排它锁后，可以进行数据的读写。在其释放之前，其它方不能获得任何锁。
基于ZK实现分布式锁流程
某银行帐户，可以同时进行帐户信息的读取，但读取其间不能修改帐户数据。其帐户ID为:888
- 获得读锁流程：
-
1、基于资源ID创建临时序号读锁节点
/lock/888.R0000000002 Read
2、获取 /lock 下所有子节点，判断其最小的节点是否为读锁，如果是则获锁成功
3、最小节点不是读锁，则阻塞等待。添加lock/ 子节点变更监听。
4、当节点变更监听触发，执行第2步
-
- 获得写锁：
-
1、基于资源ID创建临时序号写锁节点
/lock/888.R0000000002 Write
2、获取 /lock 下所有子节点，判断其最小的节点是否为自己，如果是则获锁成功
3、最小节点不是自己，则阻塞等待。添加lock/ 子节点变更监听。
4、当节点变更监听触发，执行第2步
- 释放锁：
-
读取完毕后，手动删除临时节点，如果获锁期间宕机，则会在会话失效后自动删除。
核心代码
package com.mylearn.zookeeper.test;
import com.jd.common.util.ArrayUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import java.util.*;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
/**
* Created by IntelliJ IDEA.
* User: yingkuohao
* Date: 13-11-26
* Time: 上午10:54
* CopyRight:360buy
* Descrption:
* 分布式锁 ：
* 1. 创建虚拟节点
* 2. 虚拟节点排序 ，获取最小的节点名称
* 3. 比较当前jvm创建的节点路径和最小的节点名称，如果想当则持有锁，否则等待或退出竞争。
* To change this template use File | Settings | File Templates.
*/
public class ZkLock {
private static final int SESSION_TIMEOUT = 10000;
static CountDownLatch countDownLatch = new CountDownLatch(1);
CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() {
public void run() {
System.out.println("虚拟节点准备完毕，开始抢锁"); //后续线程执行
}
});
private static String parentLockPth = "/lockRoot";
volatile static boolean flag = false;
public static void main(String args[]) {
final Map<String, String> map = new HashMap<String, String>();
map.put("node0", "192.168.192.68:2181");
map.put("node1", "192.168.229.79:2181");
map.put("node2", "192.168.229.80:2181");
final ZkLock zkLock = new ZkLock();
zkLock.creatRootNode(map.get("node0")); //创建lock父节点
checkParentIsOk(map); //校验父节点是否ok
//三个节点开始抢锁
for (int i = 0; i < 3; i++) {
final int finalI = i;
Thread thread = new Thread(new Runnable() {
public void run() {
try {
countDownLatch.await(); //三个节点同时执行
String key = "node" + finalI;
zkLock.createEphemeralNode(map.get(key), finalI);
} catch (InterruptedException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
}
});
thread.start();
}
countDownLatch.countDown();//子线程一起开始
}
/**
* 换一个节点去校验是否父节点已经同步
*
* @param map
*/
private static void checkParentIsOk(Map<String, String> map) {
ZkApi zkApi = new ZkApi(map.get("node1"));
ZooKeeper zooKeeper = zkApi.getZk();
try {
String parentData = new String(zooKeeper.getData(parentLockPth, false, null));
System.out.println(Thread.currentThread().getName() + "parentData=" + parentData);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 创建锁的根节点:lockRoot
*
* @param serverPath
*/
public void creatRootNode(String serverPath) {
ZkApi zkApi = new ZkApi(serverPath);
ZooKeeper zooKeeper = zkApi.getZk();
String parentLockPth = "/lockRoot";
zkApi.createNod(parentLockPth, "lockRootData", ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); //创建一个虚节点
try {
String parentData = new String(zooKeeper.getData(parentLockPth, false, null));
System.out.println(Thread.currentThread().getName() + "parentData=" + parentData);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 创建虚拟节点，然后抢锁，生成的结构如：
* /lockRoot/lock0
* /lockRoot/lock1
* /lockRoot/lock2
*
* @param serverPath
* @param i
*/
public void createEphemeralNode(String serverPath, int i) {
ZkApi zkApi = new ZkApi(serverPath);
String currentPath = "/lock" + i;
String wholePath = parentLockPth + currentPath; //节点路径 ,,形如/lockRoot/lock1
String testData = "lockdata"; // 节点初始化数据
List<ACL> aclList = ZooDefs.Ids.OPEN_ACL_UNSAFE; //节点的权限
// zkApi.createNod(path, testData, aclList, CreateMode.EPHEMERAL_SEQUENTIAL); //创建一个带顺序的虚节点成功后会带有一个序号，如lock1000000018, lock1000000019
zkApi.createNod(wholePath, testData, aclList, CreateMode.EPHEMERAL); //创建一个虚节点
try {
cyclicBarrier.await(); //等待三个节点都创建成功后开始抢锁
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
getLock(zkApi.getZk(), parentLockPth, currentPath); //抢锁
}
/**
* 抢锁的核心逻辑
* 1. 创建虚拟节点
* 2. 虚拟节点排序 ，获取最小的节点名称
* 3. 比较当前jvm创建的节点路径和最小的节点名称，如果想当则持有锁，否则等待或退出竞争。
*
* @param zooKeeper
* @param parentLockPth
* @param currentPath
*/
private void getLock(ZooKeeper zooKeeper, String parentLockPth, String currentPath) {
try {
List<String> childNode = zooKeeper.getChildren(parentLockPth, false); //获取孩子列表
Collections.sort(childNode); //把孩子列表排序 ，如：lock0,lock1,lock2
System.out.println(Thread.currentThread().getName() + "孩子节点一共：" + ArrayUtils.join(childNode.toArray(), ","));
String firtNode = childNode.get(0); //获取最小的节点，默认排序是升序，这里获取lock0
System.out.println(Thread.currentThread().getName() + "firNode=" + firtNode + "cureentNode=" + currentPath);
firtNode = "/" + firtNode;
if (currentPath.equals(firtNode)) {
//如果当前节点和第一个孩子节点的顺序相等，则可以获取锁
System.out.println(Thread.currentThread().getName() + "[获取锁ok]" + zooKeeper.getSaslClient().getConfigStatus());
doAction();
zooKeeper.delete(parentLockPth + firtNode, -1);//删除此节点，唤醒其他node
Stat stat = zooKeeper.exists(parentLockPth + firtNode, false);
if (stat == null) {
System.out.println(Thread.currentThread().getName() + "删除虚节点成功");
}
} else {
//如果当前节点的序号不是最小的，则等待获取锁
waitLock(zooKeeper, parentLockPth, currentPath);
}
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 等待锁，其实真正的来讲，分布式环境下获取不到锁的server就直接退出了，
* 就由获取锁的server去执行任务了。
* @param zooKeeper
* @param parentLockPth
* @param currentPath
*/
private void waitLock(ZooKeeper zooKeeper, String parentLockPth, String currentPath) {
try {
Stat stat = zooKeeper.exists(parentLockPth + currentPath, false);
while (flag = false) {
this.wait(); //flag默认为false，该节点线程一直等待，直至拥有锁的线程修改flag状态，唤醒，
}
if (stat != null) {
//如果获取锁的节点还存在，说明上个节点还在用锁，当前节点只能等待。
System.out.println(Thread.currentThread().getName() + "其他线程正在占用，请等待");
flag = false;
Thread.sleep(2000);
}
System.out.println(Thread.currentThread().getName() + "watiLock.重试,node[0]已不存在");
//如果获取锁的节点不存在了，说明上个节点获取的锁已经释放，它对应的虚节点已经被删除，这时可以继续尝试获取了
getLock(zooKeeper, parentLockPth, currentPath);
} catch (InterruptedException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
} catch (KeeperException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
}
private void getLock() {
doAction();
flag = true;
}
private void doAction() {
System.out.println(Thread.currentThread().getName() + "获取锁成功，执行内容");
try {
Thread.currentThread().sleep(2000);
System.out.println(Thread.currentThread().getName() + "睡眠完成");
} catch (InterruptedException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
}
}