ZooKeeper 实战(五) Curator实现分布式锁

本文主要是介绍ZooKeeper 实战(五) Curator实现分布式锁，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

ZooKeeper 实战(五) Curator实现分布式锁
- 1.简介
- - 1.1.分布式锁概念
  - 1.2.Curator 分布式锁的实现方式
  - 1.3.分布式锁接口
- 2.准备工作
- 3.分布式可重入锁
- - 3.1.锁对象
  - 3.2.非重入式抢占锁
  - - 测试代码
    - 输出日志
  - 3.3.重入式抢占锁
  - - 测试代码
    - 输出日志
- 4.分布式非可重入锁
- - 4.1.锁对象
  - 4.2.重入式抢占锁
  - - 测试代码
    - 输出日志
- 5.分布式可重入读写锁
- - 5.1.锁对象
  - 5.2.读锁和写锁的竞争
  - - 测试代码
    - 输出日志
- 6.共享信号量
- - 6.1.锁对象
  - 6.2.信号量抢占
  - - 测试代码
    - 输出日志
- 7.多共享锁
- - 7.1.锁对象
  - 7.2.获取共享锁
  - - 测试代码
    - 输出日志

ZooKeeper 实战(五) Curator实现分布式锁

1.简介

1.1.分布式锁概念

分布式锁是一种用于实现分布式系统中的同步机制的技术。它允许在多个进程或线程之间实现互斥访问共享资源，以避免并发访问时的数据不一致问题。分布式锁的主要目的是在分布式系统中提供类似于全局锁的效果，以确保在任何时刻只有一个进程或线程可以访问特定的资源。

zookeeper基于临时有序节点实现分布式锁。每个客户端对某个临界资源加锁时，在zookeeper上的与该临界资源对应的指定节点的目录下，生成一个唯一的临时有序节点。判断是否获取锁的方式很简单，只需要判断临时有序节点中序号最小的那个是否由自身创建。当释放锁的时候，只需将这个临时有序节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。

1.2.Curator 分布式锁的实现方式

curator-recipes中实现的锁有五种：

Shared Reentrant Lock 分布式可重入锁
Shared Lock 分布式非可重入锁
Shared Reentrant Read Write Lock 可重入读写锁
Shared Semaphore 共享信号量
Multi Shared Lock 多共享锁

1.3.分布式锁接口

org.apache.curator.framework.recipes.locks.InterProcessLock 该接口为分布式锁的行为规范，定义了获取锁和释放锁方法。

public interface InterProcessLock
{/*** 阻塞获取锁（如果未获取到锁，则一直阻塞）*/public void acquire() throws Exception;/*** 非阻塞获取锁** @param time 超时时间* @param unit 时间单位* @return 如果在超时时间内获取到锁则返回true，否则返回false* @throws Exception ZK errors, connection interruptions*/public boolean acquire(long time, TimeUnit unit) throws Exception;/*** 释放锁。每次获取锁之后必须调用该方法释放锁（acquire和release成对出现）*/public void release() throws Exception;/*** 判断该线程是否获取到锁* @return true/false*/boolean isAcquiredInThisProcess();
}

2.准备工作

首先，需要读者掌握 Ideal 同一项目启动多个Service的能力，详细教程可参考博主另一篇博客，博主创建了两个启动实例，一个端口号为8888，另一个9981，此处随意只要不与其他服务的端口号冲突即可。

在这里插入图片描述

另外，在启动项目之前，请根据先前所写的教程启动zookeeper的单机服务器，参考ZooKeeper 实战(一) 超详细的单机与集群部署教程，并创建一个存储数据节点：路径/ahao/data，数据内容40。（可参照如下操作，由于博主已经创建过，所以重新设置了一遍）

在这里插入图片描述

3.分布式可重入锁

可重入锁是一种自我保护的锁，允许同一进程或线程多次获得相同的锁，而不会造成死锁。

3.1.锁对象

类路径：org.apache.curator.framework.recipes.locks.InterProcessMutex

公开构造方法如下：

     /*** @param client 当前客户端实例* @param path   锁节点路径*/public InterProcessMutex(CuratorFramework client, String path)/*** @param client 当前客户端实例* @param path   锁节点路径* @param driver 锁驱动实例（工具类，只要提供两个方法：#createsTheLock 创建锁节点，#getsTheLock 获取当前锁）*/public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver)

3.2.非重入式抢占锁

由于非重入式抢占锁的场景对于5种分布式锁实现方式均适用，并且测试场景均一样，所以本节介绍完非重入式抢占锁的测试场景之后，对应的另4种实现方式的该测试场景将不在赘述。

测试场景：有两台服务实例 C1，C2。C1和C2个有两个线程（线程池调度）并发执行，对同一共享资源（/ahao/data中的数据）进行减1 操作。

测试代码

/*** @Name: CuratorDemoApplication* @Description:* @Author: ahao* @Date: 2024/1/10 3:29 PM*/
@Slf4j
@SpringBootApplication
public class CuratorDemoApplication implements ApplicationRunner{@Autowiredprivate CuratorFramework client;public static void main(String[] args) {SpringApplication.run(CuratorDemoApplication.class,args);}// 存储数据节点的路径final String dataPath = "/ahao/data";@Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");// 锁节点路径String lockPath = "/ahao/lock";TimeUnit.SECONDS.sleep(3);// 创建可重入锁InterProcessMutex mutex = new InterProcessMutex(client,lockPath);// 创建一个线程池ExecutorService executorService = Executors.newFixedThreadPool(2);// 保证并发执行，当前时间的秒针部分为0则结束循环int seconds = LocalDateTime.now().getSecond();while (seconds != 0){seconds = LocalDateTime.now().getSecond();}// 提交两个任务for (int i = 0; i < 2; i++) {executorService.submit(() -> {while (share(mutex)) {try {// 睡眠0.5秒TimeUnit.MILLISECONDS.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});}}/*** 用来模拟临界资源的方法*/public boolean share(final InterProcessMutex mutex){boolean b = true;try {// 获取锁if (mutex.acquire(3, TimeUnit.SECONDS)) {// 获取数据节点中的值byte[] bytes = client.getData().forPath(dataPath);String s = new String(bytes);Integer integer = Integer.valueOf(s);if(integer > 0){// 设置新值client.setData().forPath(dataPath,String.valueOf(integer-1).getBytes(StandardCharsets.UTF_8));log.info("当前值：{}",integer);b = true;}else {log.info("任务已完成。。。。");b = false;}}} catch (Exception e) {throw new RuntimeException(e);} finally {try {// 释放锁mutex.release();return b;} catch (Exception e) {log.error("释放锁失败");throw new RuntimeException(e);}}}
}

输出日志

从下方日志可见，没有出现重复的数值，保证了分布式系统中实现互斥访问共享资源，避免并发访问时的数据不一致问题。

实例c1：

2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：40
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：39
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：36
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：35
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：32
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：31
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：28
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：27
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：24
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：23
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：20
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：19
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：16
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：15
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：12
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：11
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：8
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：7
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：4
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：3
2024-01-15 INFO 65348 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。
2024-01-15 INFO 65348 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。

实例c2：

2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：38
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：37
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：34
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：33
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：30
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：29
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：26
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：25
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：22
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：21
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：18
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：17
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：14
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：13
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：10
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：9
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：6
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：5
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 当前值：2
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 当前值：1
2024-01-15 INFO 65387 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。
2024-01-15 INFO 65387 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。

3.3.重入式抢占锁

测试场景：有两台服务实例 C1，C2。C1和C2个有两个线程（线程池调度）并发执行，对同一共享资源（/ahao/data中的数据）进行连续两次的减1 操作，并且第一次操作成功后才能进行第二次操作。

测试代码

具体的减1操作抽离成方法，以便接下来的测试。

		@Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String lockPath = "/ahao/lock";TimeUnit.SECONDS.sleep(3);// 创建可重入锁InterProcessMutex mutex = new InterProcessMutex(client,lockPath);// 创建一个线程池ExecutorService executorService = Executors.newFixedThreadPool(2);// 保证并发执行，当前时间的秒针部分为0则结束循环int seconds = LocalDateTime.now().getSecond();while (seconds != 0){seconds = LocalDateTime.now().getSecond();}// 提交两个任务for (int i = 0; i < 2; i++) {executorService.submit(() -> {// 循环执行while (share(mutex,1)) {try {// 睡眠0.5秒TimeUnit.MILLISECONDS.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});}}/*** 用来模拟临界资源的方法*/public boolean share(final InterProcessLock mutex, int n){boolean b = true;try {// 获取锁if (mutex.acquire(3, TimeUnit.SECONDS)) {// 减1操作b = doLock(n);// 最多执行三次if (b && n < 3){b = share(mutex,n+1);}}} catch (Exception e) {throw new RuntimeException(e);} finally {try {mutex.release();return b;} catch (Exception e) {log.error("释放锁失败");throw new RuntimeException(e);}}}// 减1操作private boolean doLock(int n) throws Exception {// 获取数据节点中的值byte[] bytes = client.getData().forPath(dataPath);String s = new String(bytes);Integer integer = Integer.valueOf(s);// 判断是否为0if(integer > 0){// 设置新值client.setData().forPath(dataPath,String.valueOf(integer-1).getBytes(StandardCharsets.UTF_8));log.info("第{}次加锁当前值：{}",n,integer);return true;}else {log.info("任务已完成。。。。");return false;}}

输出日志

由此可见，通过日志可以发现，每次获取到锁的线程都是连续执行三次并且重复获取锁并释放。

实例c1：

2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：34
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：33
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：32
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：31
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：30
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：29
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：22
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：21
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：20
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：19
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：18
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：17
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：10
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：9
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：8
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：7
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：6
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：5
2024-01-16 INFO 91120 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。
2024-01-16 INFO 91120 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。

实例c2：

2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：40
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：39
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：38
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：37
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：36
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：35
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：28
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：27
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：26
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：25
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：24
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：23
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：16
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：15
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：14
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：13
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：12
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：11
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：4
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第2次加锁当前值：3
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 第3次加锁当前值：2
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 第1次加锁当前值：1
2024-01-16 INFO 91115 --- [pool-3-thread-2] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。
2024-01-16 INFO 91115 --- [pool-3-thread-1] com.ahao.demo.CuratorDemoApplication     : 任务已完成。。。。

4.分布式非可重入锁

与org.apache.curator.framework.recipes.locks.InterProcessMutex 类似，只是不可重入。

4.1.锁对象

类路径：InterProcessSemaphoreMutex

公开构造方法如下：

 		/*** @param client 当前客户端实例* @param path   锁节点路径*/		public InterProcessSemaphoreMutex(CuratorFramework client, String path);

4.2.重入式抢占锁

测试场景：有一台服务实例 C1。C1中的main线程进行连续两次的加锁操作。

测试代码

/*** @Name: CuratorDemoApplication* @Description:* @Author: ahao* @Date: 2024/1/10 3:29 PM*/
@Slf4j
@SpringBootApplication
public class CuratorDemoApplication implements ApplicationRunner{@Autowiredprivate CuratorFramework client;public static void main(String[] args) {SpringApplication.run(CuratorDemoApplication.class,args);}String dataPath = "/ahao/data";@Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String lockPath = "/ahao/lock";TimeUnit.SECONDS.sleep(3);// 创建非可重入锁InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(client, lockPath);// 调用测试方法share(mutex);}/*** 用来模拟临界资源的方法*/public void share(final InterProcessLock mutex){try {// 获取锁if (mutex.acquire(5, TimeUnit.SECONDS)) {log.info("第一次加锁成功");}// 再次获取锁if (!mutex.acquire(5, TimeUnit.SECONDS)) {log.info("第二次加锁失败了");}} catch (Exception e) {throw new RuntimeException(e);} finally {try {// 获取多少次锁就要释放多少次mutex.release();mutex.release();} catch (Exception e) {log.error("释放锁失败：{}",e.getStackTrace());throw new RuntimeException(e);}}}}

输出日志

本次测试仅需要启动一个实例c1即可，用于测试分布式非可重入锁多次加锁的场景。根据以下输出日志可知，第一次加锁成功，在第二次加锁时超时失败了，导致之后在第二次释放锁时出现异常。

2024-01-16 INFO 45025 --- [           main] com.ahao.demo.CuratorDemoApplication     : 第一次加锁成功
2024-01-16 INFO 45025 --- [           main] com.ahao.demo.CuratorDemoApplication     : 第二次加锁失败了
2024-01-16 ERROR 45025 --- [           main] com.ahao.demo.CuratorDemoApplication     : 释放锁失败：org.apache.curator.shaded.com.google.common.base.Preconditions.checkState(Preconditions.java:444)
2024-01-16 INFO 45025 --- [           main] ConditionEvaluationReportLoggingListener : Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
2024-01-16 ERROR 45025 --- [           main] o.s.boot.SpringApplication               : Application run failedjava.lang.IllegalStateException: Failed to execute ApplicationRunnerat org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:762)at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:749)at org.springframework.boot.SpringApplication.run(SpringApplication.java:314)at org.springframework.boot.SpringApplication.run(SpringApplication.java:1303)at org.springframework.boot.SpringApplication.run(SpringApplication.java:1292)at com.ahao.demo.CuratorDemoApplication.main(CuratorDemoApplication.java:41)
Caused by: java.lang.RuntimeException: java.lang.IllegalStateException: Not acquiredat com.ahao.demo.CuratorDemoApplication.share(CuratorDemoApplication.java:83)at com.ahao.demo.CuratorDemoApplication.run(CuratorDemoApplication.java:56)at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:759)... 5 common frames omitted
Caused by: java.lang.IllegalStateException: Not acquiredat org.apache.curator.shaded.com.google.common.base.Preconditions.checkState(Preconditions.java:444)at org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex.release(InterProcessSemaphoreMutex.java:68)at com.ahao.demo.CuratorDemoApplication.share(CuratorDemoApplication.java:80)... 7 common frames omitted

5.分布式可重入读写锁

读写锁顾名思义，包含两把锁：读锁和写锁。当写锁未生效（未被获取）时，读锁能够被多个线程获取使用。但是写锁只能被一个线程获取持有。只有当写锁释放时，读锁才能被持有。可重入表示一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁，读锁可以重入读锁。这也意味着写锁可以降级成读锁，比如请求写锁 —>读锁 —->释放写锁。从读锁升级成写锁是不行的。可重入读写锁是“公平的”，每个实例将按请求的顺序获取锁。

5.1.锁对象

类路径：org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock

公开构造方法如下：

    /*** @param client 当前客户端实例* @param path   锁节点路径*/		public InterProcessReadWriteLock(CuratorFramework client, String basePath)/***  @param client 当前客户端实例* @param path   锁节点路径* @param lockData 存储在锁节点的数据内容*/public InterProcessReadWriteLock(CuratorFramework client, String basePath, byte[] lockData)

5.2.读锁和写锁的竞争

测试场景：有两台服务实例 C1，C2。C1和C2个有3个线程（2个读线程和1个写线程）并发执行，其中写线程进行加1操作并重复执行5次，每次都加写锁，而读线程进行查询操作并重复执行10次每次都加读锁。

测试代码

@Slf4j
@SpringBootApplication
public class CuratorDemoApplication implements ApplicationRunner {@Autowiredprivate CuratorFramework client;public static void main(String[] args) {SpringApplication.run(CuratorDemoApplication.class, args);}String dataPath = "/ahao/data";@Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String lockPath = "/ahao/lock";TimeUnit.SECONDS.sleep(3);InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, lockPath);// 获取读锁InterProcessMutex readLock = readWriteLock.readLock();// 获取写锁InterProcessMutex writeLock = readWriteLock.writeLock();// 保证并发执行，当前时间的秒针部分为30的整数倍则结束循环int seconds = LocalDateTime.now().getSecond();while (seconds/30 != 0){seconds = LocalDateTime.now().getSecond();}for (int j = 0; j < 2; j++) {// 读线程new Thread(() -> {for (int i = 0; i < 10; i++) {try {// 加锁if (readLock.acquire(3, TimeUnit.SECONDS)) {doLock(false);}} catch (Exception e) {throw new RuntimeException(e);} finally {// 释放锁try {readLock.release();} catch (Exception e) {throw new RuntimeException(e);}}}}, "读线程"+j).start();}// 写线程new Thread(() -> {for (int i = 0; i < 5; i++) {try {// 加锁if (writeLock.acquire(3, TimeUnit.SECONDS)) {doLock(true);}} catch (Exception e) {throw new RuntimeException(e);} finally {// 释放锁try {writeLock.release();} catch (Exception e) {throw new RuntimeException(e);}}}}, "写线程").start();}/*** 加操作* @param isAdd true 表示加1，false 表示不加，查询数据* @return* @throws Exception*/public void doLock(boolean isAdd) throws Exception {// 获取数据节点中的值byte[] bytes = client.getData().forPath(dataPath);Integer integer = Integer.valueOf(new String(bytes));if (isAdd) {// 设置新值client.setData().forPath(dataPath, String.valueOf(integer + 1).getBytes(StandardCharsets.UTF_8));log.info("加1操作后：{}", integer);} else {log.info("查询数据：{}", integer);}}   
}

输出日志

可以观察到，读线程所查询的数据存在重复数据，说明了在同一时刻可以加多个读锁，而写线程不会出现重复数据，只能有一个线程可以获取到写锁。

实例c1：

2024-01-16  INFO 64462 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：40
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：41
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：42
2024-01-16  INFO 64462 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：42
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：43
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：44
2024-01-16  INFO 64462 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：44
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：45
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：46
2024-01-16  INFO 64462 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：46
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：47
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：48
2024-01-16  INFO 64462 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：48
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：49
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64462 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50

实例c2：

2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：40
2024-01-16  INFO 64453 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：41
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：42
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：42
2024-01-16  INFO 64453 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：43
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：44
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：44
2024-01-16  INFO 64453 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：45
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：46
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：46
2024-01-16  INFO 64453 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：47
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：48
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：48
2024-01-16  INFO 64453 --- [  写线程] com.ahao.demo.CuratorDemoApplication     : 加1操作后：49
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程1] com.ahao.demo.CuratorDemoApplication     : 查询数据：50
2024-01-16  INFO 64453 --- [ 读线程0] com.ahao.demo.CuratorDemoApplication     : 查询数据：50

6.共享信号量

对 JUC 熟悉的读者应该了解Semaphore（信号量）。Semaphore是一种用于实现同步的对象，通常用于限制对共享资源的并发访问。Semaphore可以用来控制进入临界区的线程数量，通过使用Semaphore，可以防止过多线程同时访问共享资源，从而避免出现资源竞争和死锁等问题。

而Curator中的Semaphore和JUC中的Semaphore，如出一辙，只是一个应用于分布式场景，一个应用于进程（服务器）内部。

6.1.锁对象

类路径：org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2

公开构造方法如下：

    /*** @param client 当前客户端实例* @param path   锁节点路径* @param maxLeases 允许线程进入临界区的最大数量（许可证数量）*/public InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases);/*** @param client 当前客户端实例* @param path   锁节点路径* @param count  用于监听许可证数量*/public InterProcessSemaphoreV2(CuratorFramework client, String path, SharedCountReader count);

6.2.信号量抢占

测试场景：有两台服务实例 C1，C2。C1和C2个有3个线程并发执行。但是有2个许可证，每个线程一次只能获取一个许可证，获取成功则执行4s耗时操作（睡眠4s），然后释放锁。

测试代码

    @Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String lockPath = "/ahao/lock";TimeUnit.SECONDS.sleep(3);// 创建共享信号量InterProcessSemaphoreV2 semaphoreV2 = new InterProcessSemaphoreV2(client,lockPath,2);// 保证并发执行，当前时间的秒针部分为30的整数倍则结束循环int seconds = LocalDateTime.now().getSecond();while (seconds/30 != 0){seconds = LocalDateTime.now().getSecond();}// 创建线程争夺许可证for (int i = 0; i < 3; i++) {new Thread(() -> {Lease acquire = null;try {acquire = semaphoreV2.acquire(5, TimeUnit.SECONDS);if (acquire != null){log.info("抢到许可证，参与竞争的节点：{}",semaphoreV2.getParticipantNodes());// 睡眠4秒TimeUnit.SECONDS.sleep(4);}else {log.info("抢到许可证失败");}} catch (Exception e) {throw new RuntimeException(e);}finally {if (acquire != null){semaphoreV2.returnLease(acquire);}}}, "线程"+i).start();}}

输出日志

实例c1：

2024-01-16  INFO 4319 --- [ 线程2] com.ahao.demo.CuratorDemoApplication     : 抢到许可证
2024-01-16  INFO 4319 --- [ 线程1] com.ahao.demo.CuratorDemoApplication     : 抢到许可证
2024-01-16  INFO 4319 --- [ 线程0] com.ahao.demo.CuratorDemoApplication     : 抢到许可证失败

实例c2：

2024-01-16  INFO 4307 --- [ 线程0] com.ahao.demo.CuratorDemoApplication     : 抢到许可证
2024-01-16  INFO 4307 --- [ 线程2] com.ahao.demo.CuratorDemoApplication     : 抢到许可证
2024-01-16  INFO 4307 --- [ 线程1] com.ahao.demo.CuratorDemoApplication     : 抢到许可证失败

7.多共享锁

表示将多个锁合并为一个锁。在获取多共享锁时，必须获取其内部所有的锁，才算获取成功，否则释放所有已获取的锁。同样调用释放锁方法时，会释放所有的锁。

7.1.锁对象

类路径：org.apache.curator.framework.recipes.locks.InterProcessMultiLock

公开构造方法如下：

		/*** @param client 当前客户端实例* @param path   多个锁节点路径*/public InterProcessMultiLock(CuratorFramework client, List<String> paths);/*** @param locks 多个锁对象*/public InterProcessMultiLock(List<InterProcessLock> locks);

7.2.获取共享锁

测试场景：有一台服务实例 C1，启动3个线程并发执行，抢占同一个共享锁。

测试代码

@Overridepublic void run(ApplicationArguments args) throws Exception {log.info("。。。。。。。。。。。。。容器初始化完毕。。。。。。。。。。。。。。");String lockPath = "/ahao/lock";String lockPath2 = "/ahao/lock2";TimeUnit.SECONDS.sleep(3);// 创建锁1InterProcessMutex mutex = new InterProcessMutex(client,lockPath);// 创建锁2InterProcessMutex mutex2 = new InterProcessMutex(client,lockPath2);// 创建共享锁InterProcessMultiLock multiLock = new InterProcessMultiLock(List.of(mutex,mutex2));for (int i = 0; i < 3; i++) {new Thread(()->{try {if (multiLock.acquire(5, TimeUnit.SECONDS)) {log.info("获取到锁");TimeUnit.SECONDS.sleep(3);}else {log.info("获取失败");}} catch (Exception e) {throw new RuntimeException(e);} finally {try {multiLock.release();} catch (Exception e) {throw new RuntimeException(e);}}},"线程"+i).start();}}

输出日志

可见在第3个线程获取锁时，由于没有获取到/ahao/lock2对应的锁对象导致的超时。

2024-01-16  INFO 14352 --- [ 线程1] com.ahao.demo.CuratorDemoApplication     : 获取到锁
2024-01-16  INFO 14352 --- [ 线程2] com.ahao.demo.CuratorDemoApplication     : 获取到锁
2024-01-16  INFO 14352 --- [ 线程0] com.ahao.demo.CuratorDemoApplication     : 获取失败
Exception in thread "线程0" java.lang.RuntimeException: java.lang.Exception: java.lang.IllegalMonitorStateException: You do not own the lock: /ahao/lock2at com.ahao.demo.CuratorDemoApplication.lambda$run$0(CuratorDemoApplication.java:70)at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.Exception: java.lang.IllegalMonitorStateException: You do not own the lock: /ahao/lock2at org.apache.curator.framework.recipes.locks.InterProcessMultiLock.release(InterProcessMultiLock.java:169)at com.ahao.demo.CuratorDemoApplication.lambda$run$0(CuratorDemoApplication.java:68)... 1 more
Caused by: java.lang.IllegalMonitorStateException: You do not own the lock: /ahao/lock2at org.apache.curator.framework.recipes.locks.InterProcessMutex.release(InterProcessMutex.java:140)at org.apache.curator.framework.recipes.locks.InterProcessMultiLock.release(InterProcessMultiLock.java:158)... 2 more

这篇关于ZooKeeper 实战(五) Curator实现分布式锁的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

ZooKeeper 实战(五) Curator实现分布式锁

文章目录

ZooKeeper 实战(五) Curator实现分布式锁

1.简介

1.1.分布式锁概念

1.2.Curator 分布式锁的实现方式

1.3.分布式锁接口

2.准备工作

3.分布式可重入锁

3.1.锁对象

3.2.非重入式抢占锁

测试代码

输出日志

3.3.重入式抢占锁

测试代码

输出日志

4.分布式非可重入锁

4.1.锁对象

4.2.重入式抢占锁

测试代码

输出日志

5.分布式可重入读写锁

5.1.锁对象

5.2.读锁和写锁的竞争

测试代码

输出日志

6.共享信号量

6.1.锁对象

6.2.信号量抢占

测试代码

输出日志

7.多共享锁

7.1.锁对象

7.2.获取共享锁

测试代码

输出日志

相关文章