浅谈Java(SpringBoot)基于zookeeper的分布式锁实现
作者:LJY_SUPER 发布时间:2023-11-16 08:14:56
通过zookeeper实现分布式锁
1、创建zookeeper的client
首先通过CuratorFrameworkFactory创建一个连接zookeeper的连接CuratorFramework client
public class CuratorFactoryBean implements FactoryBean<CuratorFramework>, InitializingBean, DisposableBean {
private static final Logger LOGGER = LoggerFactory.getLogger(ContractFileInfoController.class);
private String connectionString;
private int sessionTimeoutMs;
private int connectionTimeoutMs;
private RetryPolicy retryPolicy;
private CuratorFramework client;
public CuratorFactoryBean(String connectionString) {
this(connectionString, 500, 500);
}
public CuratorFactoryBean(String connectionString, int sessionTimeoutMs, int connectionTimeoutMs) {
this.connectionString = connectionString;
this.sessionTimeoutMs = sessionTimeoutMs;
this.connectionTimeoutMs = connectionTimeoutMs;
}
@Override
public void destroy() throws Exception {
LOGGER.info("Closing curator framework...");
this.client.close();
LOGGER.info("Closed curator framework.");
}
@Override
public CuratorFramework getObject() throws Exception {
return this.client;
}
@Override
public Class<?> getObjectType() {
return this.client != null ? this.client.getClass() : CuratorFramework.class;
}
@Override
public boolean isSingleton() {
return true;
}
@Override
public void afterPropertiesSet() throws Exception {
if (StringUtils.isEmpty(this.connectionString)) {
throw new IllegalStateException("connectionString can not be empty.");
} else {
if (this.retryPolicy == null) {
this.retryPolicy = new ExponentialBackoffRetry(1000, 2147483647, 180000);
}
this.client = CuratorFrameworkFactory.newClient(this.connectionString, this.sessionTimeoutMs, this.connectionTimeoutMs, this.retryPolicy);
this.client.start();
this.client.blockUntilConnected(30, TimeUnit.MILLISECONDS);
}
}
public void setConnectionString(String connectionString) {
this.connectionString = connectionString;
}
public void setSessionTimeoutMs(int sessionTimeoutMs) {
this.sessionTimeoutMs = sessionTimeoutMs;
}
public void setConnectionTimeoutMs(int connectionTimeoutMs) {
this.connectionTimeoutMs = connectionTimeoutMs;
}
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
public void setClient(CuratorFramework client) {
this.client = client;
}
}
2、封装分布式锁
根据CuratorFramework创建InterProcessMutex(分布式可重入排它锁)对一行数据进行上锁
public InterProcessMutex(CuratorFramework client, String path) {
this(client, path, new StandardLockInternalsDriver());
}
使用 acquire方法
1、acquire() :入参为空,调用该方法后,会一直堵塞,直到抢夺到锁资源,或者zookeeper连接中断后,上抛异常。
2、acquire(long time, TimeUnit unit):入参传入超时时间、单位,抢夺时,如果出现堵塞,会在超过该时间后,返回false。
public void acquire() throws Exception {
if (!this.internalLock(-1L, (TimeUnit)null)) {
throw new IOException("Lost connection while trying to acquire lock: " + this.basePath);
}
}
public boolean acquire(long time, TimeUnit unit) throws Exception {
return this.internalLock(time, unit);
}
释放锁 mutex.release();
public void release() throws Exception {
Thread currentThread = Thread.currentThread();
InterProcessMutex.LockData lockData = (InterProcessMutex.LockData)this.threadData.get(currentThread);
if (lockData == null) {
throw new IllegalMonitorStateException("You do not own the lock: " + this.basePath);
} else {
int newLockCount = lockData.lockCount.decrementAndGet();
if (newLockCount <= 0) {
if (newLockCount < 0) {
throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + this.basePath);
} else {
try {
this.internals.releaseLock(lockData.lockPath);
} finally {
this.threadData.remove(currentThread);
}
}
}
}
}
封装后的DLock代码
1、调用InterProcessMutex processMutex = dLock.mutex(path);
2、手动释放锁processMutex.release();
3、需要手动删除路径dLock.del(path);
推荐 使用:
都是 函数式编程
在业务代码执行完毕后 会释放锁和删除path
1、这个有返回结果
public T mutex(String path, ZkLockCallback zkLockCallback, long time, TimeUnit timeUnit)
2、这个无返回结果
public void mutex(String path, ZkVoidCallBack zkLockCallback, long time, TimeUnit timeUnit)
public class DLock {
private final Logger logger;
private static final long TIMEOUT_D = 100L;
private static final String ROOT_PATH_D = "/dLock";
private String lockRootPath;
private CuratorFramework client;
public DLock(CuratorFramework client) {
this("/dLock", client);
}
public DLock(String lockRootPath, CuratorFramework client) {
this.logger = LoggerFactory.getLogger(DLock.class);
this.lockRootPath = lockRootPath;
this.client = client;
}
public InterProcessMutex mutex(String path) {
if (!StringUtils.startsWith(path, "/")) {
path = Constant.keyBuilder(new Object[]{"/", path});
}
return new InterProcessMutex(this.client, Constant.keyBuilder(new Object[]{this.lockRootPath, "", path}));
}
public <T> T mutex(String path, ZkLockCallback<T> zkLockCallback) throws ZkLockException {
return this.mutex(path, zkLockCallback, 100L, TimeUnit.MILLISECONDS);
}
public <T> T mutex(String path, ZkLockCallback<T> zkLockCallback, long time, TimeUnit timeUnit) throws ZkLockException {
String finalPath = this.getLockPath(path);
InterProcessMutex mutex = new InterProcessMutex(this.client, finalPath);
try {
if (!mutex.acquire(time, timeUnit)) {
throw new ZkLockException("acquire zk lock return false");
}
} catch (Exception var13) {
throw new ZkLockException("acquire zk lock failed.", var13);
}
T var8;
try {
var8 = zkLockCallback.doInLock();
} finally {
this.releaseLock(finalPath, mutex);
}
return var8;
}
private void releaseLock(String finalPath, InterProcessMutex mutex) {
try {
mutex.release();
this.logger.info("delete zk node path:{}", finalPath);
this.deleteInternal(finalPath);
} catch (Exception var4) {
this.logger.error("dlock", "release lock failed, path:{}", finalPath, var4);
// LogUtil.error(this.logger, "dlock", "release lock failed, path:{}", new Object[]{finalPath, var4});
}
}
public void mutex(String path, ZkVoidCallBack zkLockCallback, long time, TimeUnit timeUnit) throws ZkLockException {
String finalPath = this.getLockPath(path);
InterProcessMutex mutex = new InterProcessMutex(this.client, finalPath);
try {
if (!mutex.acquire(time, timeUnit)) {
throw new ZkLockException("acquire zk lock return false");
}
} catch (Exception var13) {
throw new ZkLockException("acquire zk lock failed.", var13);
}
try {
zkLockCallback.response();
} finally {
this.releaseLock(finalPath, mutex);
}
}
public String getLockPath(String customPath) {
if (!StringUtils.startsWith(customPath, "/")) {
customPath = Constant.keyBuilder(new Object[]{"/", customPath});
}
String finalPath = Constant.keyBuilder(new Object[]{this.lockRootPath, "", customPath});
return finalPath;
}
private void deleteInternal(String finalPath) {
try {
((ErrorListenerPathable)this.client.delete().inBackground()).forPath(finalPath);
} catch (Exception var3) {
this.logger.info("delete zk node path:{} failed", finalPath);
}
}
public void del(String customPath) {
String lockPath = "";
try {
lockPath = this.getLockPath(customPath);
((ErrorListenerPathable)this.client.delete().inBackground()).forPath(lockPath);
} catch (Exception var4) {
this.logger.info("delete zk node path:{} failed", lockPath);
}
}
}
@FunctionalInterface
public interface ZkLockCallback<T> {
T doInLock();
}
@FunctionalInterface
public interface ZkVoidCallBack {
void response();
}
public class ZkLockException extends Exception {
public ZkLockException() {
}
public ZkLockException(String message) {
super(message);
}
public ZkLockException(String message, Throwable cause) {
super(message, cause);
}
}
配置CuratorConfig
@Configuration
public class CuratorConfig {
@Value("${zk.connectionString}")
private String connectionString;
@Value("${zk.sessionTimeoutMs:500}")
private int sessionTimeoutMs;
@Value("${zk.connectionTimeoutMs:500}")
private int connectionTimeoutMs;
@Value("${zk.dLockRoot:/dLock}")
private String dLockRoot;
@Bean
public CuratorFactoryBean curatorFactoryBean() {
return new CuratorFactoryBean(connectionString, sessionTimeoutMs, connectionTimeoutMs);
}
@Bean
@Autowired
public DLock dLock(CuratorFramework client) {
return new DLock(dLockRoot, client);
}
}
测试代码
@RestController
@RequestMapping("/dLock")
public class LockController {
@Autowired
private DLock dLock;
@RequestMapping("/lock")
public Map testDLock(String no){
final String path = Constant.keyBuilder("/test/no/", no);
Long mutex=0l;
try {
System.out.println("在拿锁:"+path+System.currentTimeMillis());
mutex = dLock.mutex(path, () -> {
try {
System.out.println("拿到锁了" + System.currentTimeMillis());
Thread.sleep(10000);
System.out.println("操作完成了" + System.currentTimeMillis());
} finally {
return System.currentTimeMillis();
}
}, 1000, TimeUnit.MILLISECONDS);
} catch (ZkLockException e) {
System.out.println("拿不到锁呀"+System.currentTimeMillis());
}
return Collections.singletonMap("ret",mutex);
}
@RequestMapping("/dlock")
public Map testDLock1(String no){
final String path = Constant.keyBuilder("/test/no/", no);
Long mutex=0l;
try {
System.out.println("在拿锁:"+path+System.currentTimeMillis());
InterProcessMutex processMutex = dLock.mutex(path);
processMutex.acquire();
System.out.println("拿到锁了" + System.currentTimeMillis());
Thread.sleep(10000);
processMutex.release();
System.out.println("操作完成了" + System.currentTimeMillis());
} catch (ZkLockException e) {
System.out.println("拿不到锁呀"+System.currentTimeMillis());
e.printStackTrace();
}catch (Exception e){
e.printStackTrace();
}
return Collections.singletonMap("ret",mutex);
}
@RequestMapping("/del")
public Map delDLock(String no){
final String path = Constant.keyBuilder("/test/no/", no);
dLock.del(path);
return Collections.singletonMap("ret",1);
}
}
以上所述是小编给大家介绍的Java(SpringBoot)基于zookeeper的分布式锁实现详解整合,希望对大家有所帮助.
来源:https://blog.csdn.net/LJY_SUPER/article/details/87807091
猜你喜欢
- LocalDateTime 是 Java 8 中日期时间 API 提供的一个类,在日期和时间的表示上提供了更加丰富和灵活的支持。LocalD
- ActiveMQ是Apache的一个开源项目,它是一个功能强劲的开源消息总线,也是一个中间件产品,它是JMS的一个实现。在介绍ActiveM
- 概述日常工作中,我们经常会有发送 HTTP 网络请求的需求,概括下我们常见的发送 HTTP 请求的需求内容:可以发送基本的 GET/POST
- 在讲述这个模式之前,我们先看一个案例:游戏回档游戏的某个场景,一游戏角色有生命力、攻击力、防御力等数据,在打Boss前和后会不一样,我们允许
- 背景:本人不是Java开发人员,经过四年多的历练,可以说是一枚BI攻城师了吧,最近粗糙的写了一个Portal来集成cognos报表,下面就入
- 我们在j2ee当中,连接数据库的时候经常会用到properties配置文件,我们原来在eclipse或者myeclipse当中会在src文件
- Tomcat 如何实现WebSocketWebSocket协议属于HTML5标准,越来越多浏览器已经原生支持WebSocket,它能让客户端
- 1、原理事务的概念想必大家都很清楚,其ACID特性在开发过程中占有重要的地位。同时在并发过程中会出现一些一致性问题,为了解决一致性问题,也出
- 单行文本的输入存在严重的缺陷,也不适合实际的运用,本节通过一个无功能的记事本来介绍可以进行多行输入的JTextArea:JTextArea(
- SpringCloudStream配置以下配置摘自《SpringCloud微服务实战》,配置主要包括两大部分:Stream配置(基础配置、通
- 需求: 给定一个URL地址, 例如: http://www.cncounter.com/tools/shorturl.php, 解析对应的I
- 前言 因为自己在做的一个小软件里面需要用到从A-Z排序的ListView,所以自然而然的想到了微信的联系人,我想要的就是那样的效果。本来没
- 使用foreach循环的坑我们首先看一段MyBatis中使用foreach循环的sql:SELECT * FROM table where
- 什么是队列结构一种线性结构,具有特殊的运算法则【只能在一端(队头)删除,在另一端(队尾)插入】。分类:顺序队列结构链式队列结构基本操作:入队
- 在C程序代码中我们可以利用操作系统提供的互斥锁来实现同步块的互斥访问及线程的阻塞及唤醒等工作。然而在Java中除了提供LockAPI外还在语
- 用法在java中经常会遇到需要对数据进行类型转换的场景,String类型的数据转为Int类型属于比较常见的场景,主要有两种转换方法:1. 使
- 在windows环境下,我们通常在IDE如VS的工程中开发C++项目,对于生成和使用静态库(*.lib)与动态库(*.dll)可能都已经比较
- 前言本文给你提供在Spring Boot 应用程序中编写好的单元测试的机制,并且深入技术细节。我们将带你学习如何以可测试的方式创建Sprin
- 目录题目及要求:提示:原创代码:代码思路:题目及要求:给定一个字符串 s ,请你找出其中不含有重复字符的 最长子串 的长度。提示:0 <
- Java常用类库Math类Math包含用于执行基本数字运算的方法,例如基本指数,对数,平方根和三角函数一、Field SummaryModi