分布式锁中-基于Zookeeper的实现是怎样

​篇幅太长看着也累，测验分成多个小章节，每天进步一点点。

前情回忆

分布式锁系列内容规划如下，本篇是第 2 篇：

1. 《分布式锁上-初探》
2. 《分布式锁中-根据 Zookeeper 的完结是怎样》（本篇）
3. 《分布式锁中-根据 etcd 的完结很高雅》
4. 《分布式锁中-根据 Redis 的完结需避坑 – Jedis 篇》
5. 《分布式锁中-根据 Redis 的完结很多样- Redission 篇》（写作中）
6. 《分布式锁中-根据 MySQL 的完结 – 有得用就好》（写作中）
7. 《分布式锁中-多维度的比照各种分布式锁完结》（写作中）
8. 《分布式锁下-分布式锁客户端的笼统、适配与加固》（写作中）

ZooKeeper介绍

Zookeeper（后续简称ZK）是一个分布式的，开放源码的分布式应用程序和谐服务，通常以集群形式运转，其和谐才能能够理解为是根据观察者设计形式来完结的；ZK服务会运用Znode存储运用者的数据，并将这些数据以树形目录的形式来安排办理，支撑运用者以观察者的人物指定自己重视哪些节点\数据的改变，当这些改变产生时，ZK会通知其观察者；为满意本篇目标所需，着重介绍以下几个要害特性：

• 数据安排：数据节点以树形目录(类似文件体系)安排办理，每一个节点中都会保存数据信息和节点信息。

ZooKeeper’s Hierarchical Namespace

• 集群形式：通常是由3、5个基数实例组成集群，当超越半数服务实例正常工作就能对外供给服务，既能防止单点毛病，又尽量高可用，每个服务实例都有一个数据备份，以完结数据大局一致

ZooKeeper Service

• 次序更新：更新恳求都会转由leader履行，来自同一客户端的更新将依照发送的次序被写入到ZK，处理写恳求创立Znode时，Znode称号后会被分配一个大局仅有的递加编号，能够经过次序号揣度恳求的次序，运用这个特功用够完结高档和谐服务

• 监听机制：给某个节点注册监听器，该节点一旦产生改变（例如更新或许删去），监听者就会收到一个Watch Event，能够感知到节点\数据的改变

• 暂时节点：session链接断开暂时节点就没了，不能创立子节点（很要害）

ZK的分布式锁正是根据以上特性来完结的，简略来说是：

• 暂时节点：用于支撑异常状况下的锁主动开释才能
• 次序节点：用于支撑公正锁获取锁和排队等候的才能
• 监听机制：用于支撑抢锁才能
• 集群形式：用于支撑锁服务的高可用

2. 加解锁的流程描绘

1. 创立一个永久节点作为锁节点（/lock2）

2. 企图加锁的客户端在指定锁称号节点（/lock2）下，创立暂时次序子节点

3. 获取锁节点（/lock2）下一切子节点

4. 对所获取的子节点按节点自增序号从小到大排序

5. 判别自己是不是榜首个子节点，若是，则获取锁

6. 若不是，则监听比该节点小的那个节点的删去事件（这种只监听前一个节点的方式防止了惊群效应）

7. 若是堵塞申请锁，则申请锁的操作可增加堵塞等候

8. 若监听事件生效（阐明前节点开释了，能够测验去获取锁），则回到第3步从头进行判别，直到获取到锁

9. 解锁时，将榜首个子节点删去开释

3. ZK分布式锁的才能

可能读者是单篇阅读，这儿引进榜首篇《分布式锁上-初探》中的一些内容，一个分布式锁应具有这样一些功用特色：

• 互斥性：在同一时刻，只要一个客户端能持有锁
• 安全性：防止死锁，假如某个客户端获得锁之后处理时刻超越最大约好时刻，或许持锁期间产生了毛病导致无法主动开释锁，其持有的锁也能够被其他机制正确开释，并确保后续其它客户端也能加锁，整个处理流程持续正常履行
• 可用性：也被称作容错性，分布式锁需求有高可用才能，防止单点毛病，当供给锁的服务节点毛病（宕机）时不影响服务运行，这儿有两种形式：一种是分布式锁服务本身具有集群形式，遇到毛病能主动切换恢复工作；另一种是客户端向多个独立的锁服务建议恳求，当某个锁服务毛病时仍然能够从其他锁服务读取到锁信息(Redlock)
• 可重入性：对同一个锁，加锁和解锁必须是同一个线程，即不能把其他线程程持有的锁给开释了
• 高效灵敏：加锁、解锁的速度要快；支撑堵塞和非堵塞；支撑公正锁和非公正锁

根据上文的内容，这儿简略总结一下ZK的才能矩阵（etcd 的状况请看《分布式锁中-根据etcd的完结好高雅》，Redis 锁的状况会在后续文章中弥补）：

关于功用不太高的一种说法

由于每次在创立锁和开释锁的过程中，都要动态创立、销毁暂时节点来完结锁功用。ZK中创立和删去节点只能经过Leader服务器来履行，然后Leader服务器还需求将数据同步到一切的Follower机器上，这样频频的网络通信，功用的短板是十分突出的。在高功用，高并发的场景下，不建议运用ZooKeeper的分布式锁。

由于ZooKeeper的高可用特性，在并发量不是太高的场景，也推荐运用ZK的分布式锁。

4. InterProcessMutex运用示例

Zookeeper 客户端框架 Curator 供给的 InterProcessMutex 是分布式锁的一种完结，acquire 办法堵塞|非堵塞获取锁，release 办法开释锁，另外还供给了可吊销、可重入功用。

4.1接口介绍

// 获取互斥锁
public void acquire() throws Exception;
// 在给定的时刻内获取互斥锁
public boolean acquire(long time, TimeUnit unit) throws Exception;
// 开释锁处理
public void release() throws Exception;
// 假如当前线程获取了互斥锁，则返回true
boolean isAcquiredInThisProcess();

4.2 pom依靠

<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.5.7</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>4.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.3.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-client</artifactId>
<version>4.3.0</version>
</dependency>

4.3示例

packagecom.atguigu.case3;
importorg.apache.curator.framework.CuratorFramework;
importorg.apache.curator.framework.CuratorFrameworkFactory;
importorg.apache.curator.framework.recipes.locks.InterProcessMutex;
importorg.apache.curator.retry.ExponentialBackoffRetry;
publicclassCuratorLockTest{
publicstaticvoidmain(String[]args){
//创立分布式锁1
InterProcessMutexlock1=newInterProcessMutex(getCuratorFramework(),"/locks");
//创立分布式锁2
InterProcessMutexlock2=newInterProcessMutex(getCuratorFramework(),"/locks");
newThread(newRunnable(){
@Override
publicvoidrun(){
try{
lock1.acquire();
System.out.println("线程1获取到锁");
lock1.acquire();
System.out.println("线程1再次获取到锁");
Thread.sleep(5*1000);
lock1.release();
System.out.println("线程1开释锁");
lock1.release();
System.out.println("线程1再次开释锁");
}catch(Exceptione){
e.printStackTrace();
}
}
}).start();
newThread(newRunnable(){
@Override
publicvoidrun(){
try{
lock2.acquire();
System.out.println("线程2获取到锁");
lock2.acquire();
System.out.println("线程2再次获取到锁");
Thread.sleep(5*1000);
lock2.release();
System.out.println("线程2开释锁");
lock2.release();
System.out.println("线程2再次开释锁");
}catch(Exceptione){
e.printStackTrace();
}
}
}).start();
}
privatestaticCuratorFrameworkgetCuratorFramework(){
ExponentialBackoffRetrypolicy=newExponentialBackoffRetry(3000,3);
CuratorFrameworkclient=CuratorFrameworkFactory.builder().connectString("xxx:2181,xxx:2181,xxx:2181")
.connectionTimeoutMs(2000)
.sessionTimeoutMs(2000)
.retryPolicy(policy).build();
//发动客户端
client.start();
System.out.println("zookeeper发动成功");
returnclient;
}
}

5. DIY一个阉割版的分布式锁

经过这个实例对照第2节内容来理解加解锁的流程，以及怎么防止惊群效应。

packagecom.rock.case2;
importorg.apache.zookeeper.*;
importorg.apache.zookeeper.data.Stat;
importjava.io.IOException;
importjava.util.List;
importjava.util.concurrent.CountDownLatch;
/**
*zk分布式锁v1版别:
*完结功用:
*1.防止了惊群效应
*缺失功用:
*1.超时操控
*2.读写锁
*3.重入操控
*/
publicclassDistributedLock{
privateStringconnectString;
privateintsessionTimeout;
privateZooKeeperzk;
privateCountDownLatchconnectLatch=newCountDownLatch(1);
privateCountDownLatchwaitLatch=newCountDownLatch(1);
privateStringwaitPath;
privateStringcurrentNode;
privateStringLOCK_ROOT_PATH;
privatestaticStringNODE_PREFIX="w";
publicDistributedLock(StringconnectString,intsessionTimeout,StringlockName){
//TODO:数据校验
this.connectString=connectString;
this.sessionTimeout=sessionTimeout;
this.LOCK_ROOT_PATH=lockName;
}
publicvoidinit()throwsIOException,KeeperException,InterruptedException{
//建联
zk=newZooKeeper(connectString,sessionTimeout,watchedEvent->{
//connectLatch连接上zk后开释
if(watchedEvent.getState()==Watcher.Event.KeeperState.SyncConnected){
connectLatch.countDown();
}
});
connectLatch.await();//等候zk正常连接后
//判别锁称号节点是否存在
Statstat=zk.exists(LOCK_ROOT_PATH,false);
if(stat==null){
//创立一下锁称号节点
try{
zk.create(LOCK_ROOT_PATH,LOCK_ROOT_PATH.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}catch(KeeperExceptione){
//并发创立冲突疏忽。
if(!e.code().name().equals("NODEEXISTS")){
throwe;
}
}
}
}
/**
*待弥补功用:
*1.超时设置
*2.读写区别
*3.重入操控
*/
publicvoidzklock()throwsKeeperException,InterruptedException{
if(!tryLock()){
waitLock();
zklock();
}
}
/**
*
*/
privatevoidwaitLock()throwsKeeperException,InterruptedException{
try{
zk.getData(waitPath,newWatcher(){
@Override
publicvoidprocess(WatchedEventwatchedEvent){
//waitLatch需求开释
if(watchedEvent.getType()==Watcher.Event.EventType.NodeDeleted&&watchedEvent.getPath().equals(waitPath)){
waitLatch.countDown();
}
}
},newStat());
//等候监听
waitLatch.await();
}catch(KeeperException.NoNodeExceptione){
//假如等候的节点现已被清除了,不等了,再测验去抢锁
return;
}
}
privatebooleantryLock()throwsKeeperException,InterruptedException{
currentNode=zk.create(LOCK_ROOT_PATH+"/"+NODE_PREFIX,null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
//判别创立的节点是否是最小的序号节点，假如是获取到锁；假如不是，监听他序号前一个节点
List<String>children=zk.getChildren(LOCK_ROOT_PATH,false);
//假如children 只要一个值，那就直接获取锁；假如有多个节点，需求判别，谁最小
if(children.size()==1){
returntrue;
}else{
StringthisNode=currentNode.substring(LOCK_ROOT_PATH.length()+1);
//经过w00000000获取该节点在children调集的方位
intindex=children.indexOf(thisNode);
if(index==0){
//自己便是榜首个节点
returntrue;
}
//需求监听他前一个节点改变
waitPath=LOCK_ROOT_PATH+"/"+children.get(index-1);
}
returnfalse;
}
//解锁
publicvoidunZkLock(){
//删去节点
try{
zk.delete(this.currentNode,-1);
}catch(InterruptedExceptione){
e.printStackTrace();
}catch(KeeperExceptione){
e.printStackTrace();
}
}
}

文末的弥补(请重视，莫错过)

假如这篇文章对您有所帮助，或许有所启示的话，帮助扫描下发二维码重视一下，重视大众号：架构染色，进行交流和学习。您的支撑是我坚持写作最大的动力。

求一键三连：点赞、保藏、转发。