分布式锁——zk实现

2024-05-10 11:38
文章标签 实现 分布式 zk

本文主要是介绍分布式锁——zk实现,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1 zk分布式锁原理

(参考:https://blog.csdn.net/xuefeng0707/article/details/80588855)

zk使用临时顺序节点+Watch机制实现分布式锁:

加锁:创建临时顺序节点,如果该节点是最小节点则获取到锁,否则对上一个临时节点注册监听

解锁:删除节点。

下面我们详细分析下加锁和解锁的过程,暂不考虑可重入性和锁超时,参考了curator源码。

1.1 加锁

过程1:创建临时顺序节点

举例子说明吧,在下面将要用到的例子中,我们会在“/lock/test/testZkLock/”目录下创建临时顺序子节点,子节点的名字不需要我们去命名,curator框架会自动命名为“lock-”,创建完临时顺序节点之后我们能看到这样的结构:

如果是多个进程或者线程竞争锁资源,就会在“/lock/test/testZkLock/”目录下依次创建临时顺序节点,在一次运行中我们启动3个线程,看到创建了3个临时节点:

过程2:竞争锁

在这个过程中会判断当前线程创建的临时顺序节点是不是序号最小的那个,如果是则获取到锁、继续执行后面的任务,否则的话对比当前序号小的节点注册监听。举例说明,假设在某一瞬间,某一共享资源下创建的临时节点是这样的:

lock_001
lock_002
lock_003

那么在加锁时,lock_001的线程可以获取到锁,lock_002在lock_001上注册Watch监听,然后wait()进入等待,同理lock_003在lock_002上注册监听。

原理图(https://blog.csdn.net/xuefeng0707/article/details/80588855):

curator源码:

 

1.2 解锁

不考虑可重入的话,解锁直接删除临时节点:

client.delete().guaranteed().forPath(ourPath);

当该临时节点被删除时,注册在该节点上的Watcher会监听到zk事件,然后执行process()方法:

notifyFromWatcher()其实就一行代码:

仍然举上面的例子,假设目前临时有序节点为:lock_001、lock_002、lock_003

当lock_001因为释放锁而被删除的时候,会触发zk通知机制,那么lock_002节点会收到事件通知,在收到通知的时候执行“notifyAll();”,通知当前JVM进程中所有处于wait状态的线程起来抢锁。

分布式环境下有多个进程,为什么只通知本进程的就可以?

假设节点2监听节点1是在进程A中进行的,节点3监听节点2是在进程B中进行的,那么当节点1删除的时候,通知到进程A,节点2就能收到,而节点2删除的时候会通知到进程B,节点3就能收到,意思是谁注册的监听就通知谁,这个过程是对的。

2 使用Curator实现分布式锁

Curator是Netfix公司开源的一套ZooKeeper客户端框架,对zk底层的连接、监听等进行了良好的封装,并且还提供了分布式锁API,因此我们不必自己实现上述复杂的理论,直接使用curator框架即可。

Curator分布式锁是一种可重入锁,实现了分布式的AQS,使用ConcurrentMap实现了一个类似ThreadLocal的功能,把线程(Thread.currentThread())作为key,锁作为value,在加锁时,如果在key中找到该线程,就对value加1,解锁时减1,减到0的时候删除临时节点、移除map中的该线程。

首先定义锁接口:

package utils.distributed.lock;public interface DistributedLock {/*** 阻塞式锁* @return*/void lock();/*** 非阻塞式锁* @return*/boolean tryLock();/*** 带超时时间的阻塞式锁* @param timeout* @return*/boolean tryLock(long timeout);/*** 解锁*/void unLock() throws Exception;/*** 释放资源*/void shutdown();
}

然后实现一个zk锁:

package utils.distributed.lock;import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;import java.util.Arrays;
import java.util.concurrent.TimeUnit;/*** zk分布式锁*/
public class ZKLock implements DistributedLock {private CuratorFramework client;private InterProcessMutex lock;public ZKLock(String host, String bizType, String lockKey) {client = CuratorFrameworkFactory.newClient(host,new ExponentialBackoffRetry(ZKLockConstant.BASE_SLEEP_TIME_MS, ZKLockConstant.MAX_RETRIES));client.start();String path = ZKLockConstant.ZK_SEPERATOR + StringUtils.join(Arrays.asList(ZKLockConstant.ZK_LOCK_BASE_PREFIX, bizType, lockKey), ZKLockConstant.ZK_SEPERATOR);lock = new InterProcessMutex(client, path);}public void lock() {try {lock.acquire();} catch (Exception e) {e.printStackTrace();}}public boolean tryLock() {return tryLock(0);}public boolean tryLock(long timeout) {try {return lock.acquire(timeout, TimeUnit.MILLISECONDS);} catch (Exception e) {e.printStackTrace();return false;}}public void unLock() throws Exception {lock.release();}public void shutdown() {client.close();}
}

 

3 测试

3.1 测试lock

package mytest.distributed.lock;import utils.distributed.lock.ZKLock;import java.util.concurrent.CountDownLatch;public class ZkLockTest {public static void main(String[] args) {String zkHost = "192.168.160.128:2181";String bizType = "test";String lockKey = "testZkLock";ZKLock zkLock = new ZKLock(zkHost, bizType, lockKey);// 启动3个线程模拟分布式锁竞争CountDownLatch conutDownLatch = new CountDownLatch(3);for (int i = 0; i < 3; i++) {new Thread(() -> {testZkLock(zkLock);conutDownLatch.countDown();}).start();}try {conutDownLatch.await();} catch (InterruptedException e) {e.printStackTrace();}zkLock.shutdown();}private static void testZkLock(ZKLock zkLock) {System.out.println("######## 开始加锁,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());zkLock.lock();try {System.out.println("######## 加锁成功,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}} finally {try {zkLock.unLock();System.out.println("######## 解锁成功,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());} catch (Exception e) {System.out.println("######## 解锁失败,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());}}}
}

打印结果:

全部成功了。

3.2 测试tryLock()

复用上面的main方法,将testZkLock换成testZkLockTryLock:

private static void testZkLockTryLock(ZKLock zkLock) {System.out.println("######## 开始加锁,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());if (!zkLock.tryLock()) {System.out.println("######## 加锁失败,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());return;}try {System.out.println("######## 加锁成功,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());} finally {try {zkLock.unLock();System.out.println("######## 解锁成功,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());} catch (Exception e) {System.out.println("######## 解锁失败,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());}}
}

打印结果:

 

 

因为只尝试一次,所以有失败的情况发生。

3.3 测试tryLock(timeout)

private static void testZkLockTryLockTimeOut(ZKLock zkLock) {System.out.println("######## 开始加锁,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());if (!zkLock.tryLock(3000)) {System.out.println("######## 加锁失败,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());return;}try {System.out.println("######## 加锁成功,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}} finally {try {zkLock.unLock();System.out.println("######## 解锁成功,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());} catch (Exception e) {System.out.println("######## 解锁失败,线程信息:" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());}}
}

打印结果:

线程18获取锁之后要持有5s才会释放,而线程16、17加锁等待时间只有3s,因此会超时。

4 分布式锁注解

实现一个基于注解的分布式锁:

4.1 定义分布式锁注解

package utils.distributed.annotation;import java.lang.annotation.*;/*** ZK分布式锁注解*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ZkLock {/*** zk地址* @return*/String zkHost();/*** 业务类型*/String bizType();/*** 锁名称** @return*/String lockKey();/*** 超时时间* @return*/long timeout();
}

4.2 定义分布式锁切面

package utils.distributed.advice;import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import utils.distributed.annotation.ZkLock;
import utils.distributed.lock.ZKLock;/*** @Description zk分布式锁切面* @Author lilong* @Date 2019-04-08 13:50*/
@Aspect
public class ZkLockAspectAdvice {@Around(value = "@annotation(utils.distributed.annotation.ZkLock) && @annotation(zkLock)")public Object process(ProceedingJoinPoint pjp, ZkLock zkLock) throws Throwable {ZKLock lock = new ZKLock(zkLock.zkHost(), zkLock.bizType(), zkLock.lockKey());boolean acquired = false;try {acquired = lock.tryLock(zkLock.timeout());if (acquired) {return pjp.proceed();} else {System.out.println("######## 加锁失败,线程信息:" + Thread.currentThread().getId() + "-"+ Thread.currentThread().getName());return null;}} finally {if (acquired) {lock.unLock();}}}
}

4.3 在applicationContext中增加切面配置

<!-- zk分布式锁注解 -->
<bean id="zkLockAspectAdvice" class="utils.distributed.advice.ZkLockAspectAdvice"/>
<bean id="zkLock" class="utils.distributed.lock.ZKLock" destroy-method="shutdown"/>

4.4 使用注解

@Override
@ZkLock(zkHost = "192.168.160.128:2181", bizType = "test", lockKey = "queryKeyValue", timeout = 3000)
public KeyValueJsonPO queryKeyValue(String bizType, String key) {return keyValueJsonPOMapper.queryKeyValue(bizType, key);
}

 

这篇关于分布式锁——zk实现的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


原文地址:
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.chinasem.cn/article/976342

相关文章

Python实现批量提取BLF文件时间戳

《Python实现批量提取BLF文件时间戳》BLF(BinaryLoggingFormat)作为Vector公司推出的CAN总线数据记录格式,被广泛用于存储车辆通信数据,本文将使用Python轻松提取... 目录一、为什么需要批量处理 BLF 文件二、核心代码解析:从文件遍历到数据导出1. 环境准备与依赖库

linux下shell脚本启动jar包实现过程

《linux下shell脚本启动jar包实现过程》确保APP_NAME和LOG_FILE位于目录内,首次启动前需手动创建log文件夹,否则报错,此为个人经验,供参考,欢迎支持脚本之家... 目录linux下shell脚本启动jar包样例1样例2总结linux下shell脚本启动jar包样例1#!/bin

go动态限制并发数量的实现示例

《go动态限制并发数量的实现示例》本文主要介绍了Go并发控制方法,通过带缓冲通道和第三方库实现并发数量限制,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面... 目录带有缓冲大小的通道使用第三方库其他控制并发的方法因为go从语言层面支持并发,所以面试百分百会问到

Go语言并发之通知退出机制的实现

《Go语言并发之通知退出机制的实现》本文主要介绍了Go语言并发之通知退出机制的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录1、通知退出机制1.1 进程/main函数退出1.2 通过channel退出1.3 通过cont

Python实现PDF按页分割的技术指南

《Python实现PDF按页分割的技术指南》PDF文件处理是日常工作中的常见需求,特别是当我们需要将大型PDF文档拆分为多个部分时,下面我们就来看看如何使用Python创建一个灵活的PDF分割工具吧... 目录需求分析技术方案工具选择安装依赖完整代码实现使用说明基本用法示例命令输出示例技术亮点实际应用场景扩

java如何实现高并发场景下三级缓存的数据一致性

《java如何实现高并发场景下三级缓存的数据一致性》这篇文章主要为大家详细介绍了java如何实现高并发场景下三级缓存的数据一致性,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 下面代码是一个使用Java和Redisson实现的三级缓存服务,主要功能包括:1.缓存结构:本地缓存:使

如何在Java Spring实现异步执行(详细篇)

《如何在JavaSpring实现异步执行(详细篇)》Spring框架通过@Async、Executor等实现异步执行,提升系统性能与响应速度,支持自定义线程池管理并发,本文给大家介绍如何在Sprin... 目录前言1. 使用 @Async 实现异步执行1.1 启用异步执行支持1.2 创建异步方法1.3 调用

Spring Boot配置和使用两个数据源的实现步骤

《SpringBoot配置和使用两个数据源的实现步骤》本文详解SpringBoot配置双数据源方法,包含配置文件设置、Bean创建、事务管理器配置及@Qualifier注解使用,强调主数据源标记、代... 目录Spring Boot配置和使用两个数据源技术背景实现步骤1. 配置数据源信息2. 创建数据源Be

在MySQL中实现冷热数据分离的方法及使用场景底层原理解析

《在MySQL中实现冷热数据分离的方法及使用场景底层原理解析》MySQL冷热数据分离通过分表/分区策略、数据归档和索引优化,将频繁访问的热数据与冷数据分开存储,提升查询效率并降低存储成本,适用于高并发... 目录实现冷热数据分离1. 分表策略2. 使用分区表3. 数据归档与迁移在mysql中实现冷热数据分

linux批量替换文件内容的实现方式

《linux批量替换文件内容的实现方式》本文总结了Linux中批量替换文件内容的几种方法,包括使用sed替换文件夹内所有文件、单个文件内容及逐行字符串,强调使用反引号和绝对路径,并分享个人经验供参考... 目录一、linux批量替换文件内容 二、替换文件内所有匹配的字符串 三、替换每一行中全部str1为st