Apache Curator 分布式锁的介绍,以及案例

server/2024/10/20 1:26:55/

可重入锁(InterProcessMutex):这种锁允许同一个客户端多次获取同一把锁而不会被阻塞,类似于Java中的ReentrantLock。它通过在Zookeeper的指定路径下创建临时序列节点来实现锁的功能。如果获取锁失败,当前线程会监听前一个节点的变动情况并等待,直到被唤醒或超时

package com.zz.lock;import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;public class CuratorReentrantLockExample {private final String lockPath = "/curator/lock"; // 锁的Zookeeper路径private CuratorFramework client; // Curator客户端private InterProcessMutex mutex; // 可重入锁// 初始化Curator客户端和可重入锁public void init() {// 设置Zookeeper服务地址String connectString = "192.168.200.130:2181";// 设置重试策略ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(3000, 3);// 创建Curator客户端client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);client.start();// 创建可重入锁mutex = new InterProcessMutex(client, lockPath);}// 执行业务逻辑,使用可重入锁public void executeBusinessLogic() {try {// 获取锁mutex.acquire();// 模拟业务逻辑System.out.println("当前线程获得锁,开始执行业务逻辑。");// 模拟重入逻辑reentrantLock();// 模拟业务逻辑System.out.println("当前线程完成业务逻辑执行。");} catch (Exception e) {e.printStackTrace();} finally {// 确保释放锁if (mutex.isAcquiredInThisProcess()) {try {mutex.release();} catch (Exception e) {e.printStackTrace();}}}}// 模拟可重入逻辑public void reentrantLock() {try {// 再次获取同一把锁mutex.acquire();System.out.println("当前线程重入成功,再次获得同一把锁。");// 模拟一些操作...} catch (Exception e) {e.printStackTrace();} finally {// 释放锁if (mutex.isAcquiredInThisProcess()) {try {mutex.release();} catch (Exception e) {e.printStackTrace();}}}}// 程序入口public static void main(String[] args) {CuratorReentrantLockExample example = new CuratorReentrantLockExample();example.init();// 执行业务逻辑example.executeBusinessLogic();}
}

不可重入锁(InterProcessSemaphoreMutex):与可重入锁类似,但不允许同一个线程在持有锁的情况下再次获取该锁。这种锁很容易导致死锁,使用时需要特别注意

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;public class CuratorReentrantLockExample {private final String lockPath = "/curator/lock"; // 锁的Zookeeper路径private CuratorFramework client; // Curator客户端private InterProcessMutex mutex; // 可重入锁// 初始化Curator客户端和可重入锁public void init() {// 设置Zookeeper服务地址String connectString = "127.0.0.1:2181";// 设置重试策略ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(3000, 3);// 创建Curator客户端client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);client.start();// 创建可重入锁mutex = new InterProcessMutex(client, lockPath);}// 执行业务逻辑,使用可重入锁public void executeBusinessLogic() {try {// 获取锁mutex.acquire();// 模拟业务逻辑System.out.println("当前线程获得锁,开始执行业务逻辑。");// 模拟重入逻辑reentrantLock();// 模拟业务逻辑System.out.println("当前线程完成业务逻辑执行。");} catch (Exception e) {e.printStackTrace();} finally {// 确保释放锁if (mutex.isAcquiredInThisProcess()) {try {mutex.release();} catch (Exception e) {e.printStackTrace();}}}}// 模拟可重入逻辑public void reentrantLock() {try {// 再次获取同一把锁mutex.acquire();System.out.println("当前线程重入成功,再次获得同一把锁。");// 模拟一些操作...} catch (Exception e) {e.printStackTrace();} finally {// 释放锁if (mutex.isAcquiredInThisProcess()) {try {mutex.release();} catch (Exception e) {e.printStackTrace();}}}}// 程序入口public static void main(String[] args) {CuratorReentrantLockExample example = new CuratorReentrantLockExample();example.init();// 执行业务逻辑example.executeBusinessLogic();}
}

读写锁(InterProcessReadWriteLock):提供一对相关的锁,读锁可以被多个读操作共享,而写锁则独占。一个拥有写锁的线程可以获取读锁,但读锁不能升级为写锁。这种锁是公平的,保证用户按请求顺序获取锁。(读写锁在逻辑上有点像数据库的事务)

package com.zz.lock;import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;public class CuratorReadWriteLockExample {private final String lockPath = "/curator/read-write-lock"; // 锁的Zookeeper路径private CuratorFramework client; // Curator客户端private InterProcessReadWriteLock lock; // 读写锁// 初始化Curator客户端和读写锁public void init() {// 设置Zookeeper服务地址String connectString = "192.168.200.130:2181";// 设置重试策略ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(3000, 3);// 创建Curator客户端client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);client.start();// 创建读写锁lock = new InterProcessReadWriteLock(client, lockPath);}// 执行读操作public void executeReadOperation() throws Exception {try {// 获取读锁lock.readLock().acquire();// 模拟读操作System.out.println("读操作开始,线程安全地读取数据。");// 模拟读操作延迟Thread.sleep(3000);System.out.println("读操作结束。");} catch (Exception e) {e.printStackTrace();} finally {// 释放读锁if (lock.readLock().isAcquiredInThisProcess()) {lock.readLock().release();}}}// 执行写操作public void executeWriteOperation() throws Exception {try {// 获取写锁lock.writeLock().acquire();// 模拟写操作System.out.println("写操作开始,线程独占资源进行写入。");// 模拟写操作延迟Thread.sleep(3000);System.out.println("写操作结束,更新了数据。");} catch (Exception e) {e.printStackTrace();} finally {// 释放写锁if (lock.writeLock().isAcquiredInThisProcess()) {lock.writeLock().release();}}}// 程序入口public static void main(String[] args) {CuratorReadWriteLockExample example = new CuratorReadWriteLockExample();example.init();// 启动多个读操作线程for (int i = 0; i < 5; i++) {new Thread(() -> {try {example.executeReadOperation();} catch (Exception e) {e.printStackTrace();}}).start();}// 启动写操作线程new Thread(() -> {try {example.executeWriteOperation();} catch (Exception e) {e.printStackTrace();}}).start();}
}

联锁(InterProcessMultiLock):这是一个锁的容器,可以同时获取多个锁。如果获取过程中任何一个锁请求失败,已获取的所有锁都会被释放。这在需要同时持有多个锁执行操作的场景中非常有用

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMultiLock;
import java.util.Arrays;
import java.util.List;public class InterProcessMultiLockExample {private CuratorFramework client;private List<String> lockPaths = Arrays.asList("/lock1", "/lock2", "/lock3");private List<InterProcessLock> locks = lockPaths.stream().map(path -> new InterProcessMutex(client, path)).collect(Collectors.toList());private InterProcessMultiLock multiLock;public void init() {String connectString = "127.0.0.1:2181";RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 3);client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);client.start();multiLock = new InterProcessMultiLock(locks);}public void executeProtectedOperation() {try {multiLock.acquire();// 所有锁都已获取,执行你的业务逻辑System.out.println("All locks acquired, performing business logic.");// 业务逻辑...} catch (Exception e) {e.printStackTrace();} finally {// 确保释放所有锁multiLock.release();}}public static void main(String[] args) {InterProcessMultiLockExample example = new InterProcessMultiLockExample();example.init();example.executeProtectedOperation();}
}

信号量(InterProcessSemaphoreV2):Curator提供了一种信号量实现,可以控制同时访问某个资源的线程数量。通过acquire方法请求获取信号量,使用完成后通过returnAll方法释放

信号量(Semaphore)确实可以起到限流的作用。在分布式系统中,信号量是一种常用的限流工具,它通过控制同时访问某个资源或执行某个操作的线程数量来实现限流。以下是信号量实现限流的几个关键点:1. **资源限制**:信号量通过一个计数器来限制可用资源的数量。例如,如果你有10个停车位,你可以设置信号量的初始值为10。2. **请求处理**:当一个线程需要访问资源时,它首先尝试从信号量中获取一个“许可”(lease)。如果信号量的计数器大于0,该线程成功获取一个许可,然后继续执行。否则,线程将被阻塞,直到其他线程释放资源。3. **释放资源**:线程完成资源访问后,必须释放它获取的许可,通过将许可返还给信号量来实现。这会将信号量的计数器增加1,允许其他等待的线程获取许可。4. **公平性**:信号量通常是公平的,意味着线程将按照它们请求许可的顺序来获得它们。这有助于避免某些线程长时间等待访问资源。5. **跨JVM共享**:在分布式系统中,不同的进程可能在不同的JVM中运行。Apache Curator 提供的 `InterProcessSemaphoreV2` 允许跨JVM共享信号量状态,因此所有相关进程都能协调地访问共享资源。6. **自动资源回收**:如果持有信号量许可的线程或进程崩溃,Curator 会自动释放该许可,确保资源不会被永久占用,其他线程可以继续获取该资源。通过这种方式,信号量可以有效地控制对共享资源的并发访问,防止系统过载,从而实现限流。这在许多场景下都非常有用,比如数据库连接池、线程池、外部服务调用等。
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.curator.framework.recipes.locks.Lease;
import org.apache.curator.retry.ExponentialBackoffRetry;public class InterProcessSemaphoreV2Demo {private static final String PATH = "/semaphore/path";private static CuratorFramework client;public static void main(String[] args) throws Exception {// 初始化Curator客户端client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(3000, 2));client.start();// 创建InterProcessSemaphoreV2实例,设置最大租约数为5InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, PATH, 5);// 线程示例,模拟同时请求信号量的多个线程for (int i = 0; i < 10; i++) {new Thread(new SemaphoreTask(semaphore)).start();}// 等待一段时间,让线程执行Thread.sleep(10000);// 关闭客户端连接client.close();}static class SemaphoreTask implements Runnable {private final InterProcessSemaphoreV2 semaphore;public SemaphoreTask(InterProcessSemaphoreV2 semaphore) {this.semaphore = semaphore;}@Overridepublic void run() {try {Lease lease = semaphore.acquire();System.out.println(Thread.currentThread().getName() + " acquired a lease.");// 模拟业务逻辑处理Thread.sleep(3000);// 释放信号量租约semaphore.returnLease(lease);System.out.println(Thread.currentThread().getName() + " returned a lease.");} catch (Exception e) {e.printStackTrace();}}}
}

 

分布式锁的实现原理:Curator的分布式锁通常是基于Zookeeper的临时顺序节点来实现的。当多个客户端尝试获取锁时,Zookeeper会为它们创建顺序节点,并让它们按照节点的序号依次尝试获取锁。未获取到锁的客户端会监听前一个序号的节点,一旦前一个节点释放锁,监听的客户端就会尝试获取锁


http://www.ppmy.cn/server/99969.html

相关文章

单元测试详解

&#x1f345; 点击文末小卡片 &#xff0c;免费获取软件测试全套资料&#xff0c;资料在手&#xff0c;涨薪更快 单元测试是很常见的技术的名词&#xff0c;但背后的逻辑和原理你是否清楚&#xff0c;让我们一起review一下。 1、单元测试是什么&#xff1f; 单测是单元测试,…

Python管理myql、搭建frp服务上线mysql数据库、Python生成可执行文件及python模块发布web服务

Day 21 复习 [rootpython ~]# vim test03.py a3 b4 print(ab) print(a**2b**2) [rootpython ~]# python3 test03.py 7 25 # 逐行调试 [rootpython ~]# python3 -m pdb test03.py > /root/test03.py(1)<module>() -> a3 (Pdb) n > /root/test03.p…

LeetCode --- 136双周赛

题目列表 3238. 求出胜利玩家的数目 3239. 最少翻转次数使二进制矩阵回文 I 3240. 最少翻转次数使二进制矩阵回文 II 3241. 标记所有节点需要的时间 一、求出胜利玩家的数量 这题可以直接模拟统计符合条件的玩家数量&#xff0c;代码如下 class Solution { public:int wi…

【应用开发】关于RS232串口如何进行数据传输

【应用开发】关于RS232串口如何进行数据传输 1.背景2. 传输1.背景 RS-232C 全双工通信利用 TX 和 RX 引脚在独立的线路上同时传输和接收模拟电压信号,从而实现高效的双向数据通信。 这种全双工模式使得 RS-232C 在需要双向实时通信的应用中表现得非常出色。 在编程时上位机发…

变量的注意或许需要调试

输入一个自然数N&#xff08;1<N<9&#xff09;&#xff0c;从小到大输出用1~N组成的所有排列&#xff0c;也就说全排列。例如输入3则输出 123 132 213 231 312 321 输入格式: 输入一个自然数N&#xff08;1<N<9&#xff09; 输出格式: N的全排列&#xff0c;每行一…

OpenCV 与多视图几何

欢迎访问我的博客首页。 OpenCV 与多视图几何 1. 工具函数2. 本质矩阵3. 单应矩阵4. 基础矩阵5. 参考 1. 工具函数 像素坐标、归一化坐标和相机坐标之间的相互转化。 #include <iostream> #include <opencv2/opencv.hpp>// 相机坐标系到像素坐标系的变换。 cv::Po…

前端工程化项目 用npm拉git项目的时候是在是太慢了怎么办

最近在家拉git项目发现npm i之后,开始下得挺快&#xff0c;过会就卡着不动了&#xff0c;大概几分钟后才下好。这对一个有强迫症的码农来说是不能容忍的。 只能退出去 重新下载 其实我们只要换一下国内的下载镜像源就好了 npm config set registry https://registry.npmmirror…

基于STM32开发的智能电能监测系统

目录 引言环境准备工作 硬件准备软件安装与配置系统设计 系统架构硬件连接代码实现 初始化代码控制代码应用场景 家庭电能监测工业用电管理常见问题及解决方案 常见问题解决方案结论 1. 引言 智能电能监测系统通过实时采集电流、电压等电力参数&#xff0c;计算电能消耗&…