zookeeper实现分布式锁
zookeeper案例实现
- 加入jar包依赖（使用curator框架）
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.12.0</version>
</dependency>
- 配置curator
public class ZKCuratorManager {
private static InterProcessMutex lock;
private static CuratorFramework cf;
private static String zkAddr = "127.0.0.1:2181";
private static String lockPath = "/distribute-lock";
static {
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
cf = CuratorFrameworkFactory.builder()
.connectString(zkAddr)
.sessionTimeoutMs(2000)
.retryPolicy(retryPolicy)
.build();
cf.start();
}
public static InterProcessMutex getLock(){
lock = new InterProcessMutex(cf, lockPath);
return lock;
}
}
- 锁的获取和释放
public class ZKCuratorLock {
//从配置类中获取分布式锁对象
private static InterProcessMutex lock = ZKCuratorManager.getLock();
//加锁
public static boolean acquire(){
try {
lock.acquire();
System.out.println(Thread.currentThread().getName() + " acquire success");
} catch (Exception e) {
e.printStackTrace();
}
return true;
}
//锁的释放
public static void release(){
try {
lock.release();
System.out.println(Thread.currentThread().getName() + " release success");
} catch (Exception e) {
e.printStackTrace();
}
}
}
- 使用CyclicBarrier模拟并发获取分布式锁
/**
* 使用CyclicBarrier模拟并发获取分布式锁
*/
public class CucTest {
public static void main(String[] args) {
int N = 4;
CyclicBarrier barrier = new CyclicBarrier(N);
for(int i=0;i<N;i++){
new Writer(barrier).start();
}
System.out.println("END");
}
static class Writer extends Thread{
private CyclicBarrier cyclicBarrier;
public Writer(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run() {
System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");
try {
Thread.sleep(5000); //以睡眠来模拟写入数据操作
System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
cyclicBarrier.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("所有线程写入完毕，继续处理其他任务...");
//加锁
ZKCuratorLock.acquire();
System.out.println("线程"+ Thread.currentThread().getName() +"获得分布式锁");
try {
Thread.sleep(2000);
ZKCuratorLock.release();
System.out.println("线程"+Thread.currentThread().getName()+"释放分布式锁");
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("END");
}
}
}
-
测试结果
zk实现分布式锁测试结果.png
zookeeper实现原理
- 首先，zk里有一把锁，这个锁就是zk上的一个节点。然后客户端都要来获取这个锁。
- 假设客户端A先对zk发起了加分布式锁的请求，这个加锁请求是用到了zk中的一个特殊的概念，叫做“临时顺序节点”。就是直接在"my_lock"这个锁节点下，创建一个顺序节点，这个顺序节点有zk内部自行维护的一个节点序号。
- 接着他会查一下"my_lock"这个锁节点下的所有子节点，并且这些子节点是按照序号排序的，这个时候他会拿到一个集合。
- 如果是排在第一个的话，就可以加锁了。
- 客户端B过来想要加锁了，这个时候他会干一样的事儿：也先是在"my_lock"这个锁节点下创建一个临时顺序节点。
- 客户端B因为是第二个来创建顺序节点的，所以zk内部会维护序号为"2"。
- 接着客户端B会走加锁判断逻辑，查询"my_lock"锁节点下的所有子节点。同时检查自己创建的顺序节点，是不是集合中的第一个？加锁失败！
- 加锁失败以后，客户端B就会通过ZK的API，对他的上一个顺序节点加一个监听器。监听这个节点是否被删除等变化！
- 接着，客户端A加锁之后，可能处理了一些代码逻辑，然后就会释放锁。将这个节点给删除。
- 删除那个节点之后，zk会负责通知监听这个节点的监听器，也就是客户端B之前加的那个监听器，说：兄弟，你监听的那个节点被删除了，释放了锁。
- 此时客户端B的监听器感知到了上一个顺序节点被删除，也就是排在他之前的某个客户端释放了锁。
-
此时，就会通知客户端B重新尝试去获取锁，也就是获取"my_lock"节点下的子节点集合。循环以上判断。
原理流程图.png
