Java手写分布式锁的实现
part1 前言
随着互联网业务的发展，原本单机部署的系统演化成如今的分布式集群系统后，由于分布式系统多线程，多进程并且分布在不同的机器上，这会使原本的单机锁失效，而且单纯的java api并不能提供分布式锁的能力，为了解决这个问题就需要一种跨jvm的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题。
本文将从实现本地锁所产生的问题入手，从而介绍分布式锁主流的实现方案，重点实现基于redis的分布式锁。
本地锁会出现的问题（此篇幅代码图片过多，因此放在最后）
part2 分布式锁的实现
1 分布式锁主要的实现有：
基于数据库实现分布式锁
基于缓存（redis等,本文基于redis实现手写分布式锁 ，因为这样可以更好的理解分布式锁的原理及实现，当然也可以使用redisson）
基于zookeeper
2 每种分布式锁的解决方案都有各自的优缺点
性能角度：redis > zk > mysql
安全角度：zk > redis == mysql
难易程度：zk > redis > mysql
3 分布式锁要具备的特点：
独占排他互斥
可以通过 setnx （redis命令：执行多次，只有一次能够成功）
set key value ex 3 nx
防死锁发生
请求获取到锁之后，服务器挂掉了，导致锁无法释放：给lock锁添加过期时间
可以通过redis命令 expire
或者通过 set key value ex 3 nx
保证原子性
redis是单线程的，接受或者执行指令遵循one-by-one原则。只要指令之间不被插入其他指令即可保证原子性，lua脚本批量发送多个指令给redis服务器，lua脚本也可以实现一些业务逻辑，redis集成了lua脚本，可以直接使用eval指令执行lua脚本。
获取锁和设置过期时间之间
判断和删除之间：lua脚本
防误删：
uuid给每个线程的锁添加唯一标识
自动续期
可重入：hash数据结构 + lua脚本
集群情况下，可能导致锁失效：redlock算法（redis特有的）
一个请求从主中获取到锁，从还没来得及同步数据，主就挂掉了，从就升级为新主，新的请求就可以从新主中获取锁
4 基于redis分布式锁的基本实现
我们可以借助于redis中的命令setnx(key, value)，key不存在就新增，存在就什么都不做。同时有多个客户端发送setnx命令，只有一个客户端可以成功，返回1（true）；其他的客户端返回0（false），流程图如下图所示：
多个客户端同时尝试获取锁（setnx）
获取成功，执行业务逻辑，执行完成释放锁（del）
其他客户端等待重试
代码实现：
public void testlock() { // 1. 从redis中获取锁,setnx boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, 111); if (lock) { // 查询redis中的num值 string value = this.redistemplate.opsforvalue().get(num); // 没有该值return if (stringutils.isblank(value)){ return ; } // 有值就转成成int int num = integer.parseint(value); // 把redis中的num值+1 this.redistemplate.opsforvalue().set(num, string.valueof(++num)); // 2. 释放锁 del this.redistemplate.delete(lock); } else { // 3. 每隔1秒钟回调一次，再次尝试获取锁 try { thread.sleep(1000); testlock(); } catch (interruptedexception e) { e.printstacktrace(); } }}
那么以上代码是否可以解决全部问题呢？ 显示是不能的，我们假设setnx刚好获取到锁，业务逻辑出现异常，导致锁无法释放，怎么办呢？
5 优化分布式锁_设置过期时间
设置过期有俩种方式可以选择：
通过expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）
在set时指定过期时间（推荐）
代码实现优化就是在设置锁的时候设置过期时间:
public void testlock() { // 1. 从redis中获取锁,setnx boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, 111,3, timeunit.minutes); if (lock) { //与之前相同代码略过 ... }}
那么还会不会存在问题呢?
场景：如果业务逻辑的执行时间是7s。执行流程如下:
index1业务逻辑没执行完，3秒后锁被自动释放。
index2获取到锁，执行业务逻辑，3秒后锁被自动释放。
index3获取到锁，执行业务逻辑
index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。
最终等于没锁的情况。
解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁
6 优化分布式锁_防止误删除
public void testlock() { // 1. 从redis中获取锁,setnx string uuid = uuid.randomuuid().tostring(); boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, uuid,3, timeunit.minutes); if (lock) { //与之前相同代码略过 ... // 2. 释放锁 del if (stringutils.equals(redistemplate.opsforvalue().get(lock),uuid)){ this.redistemplate.delete(lock); } }}
场景：
index1执行删除时，查询到的lock值确实和uuid相等
index1执行删除前，lock刚好过期时间已到，被redis自动释放
index2获取了lock
index1执行删除，此时会把index2的lock删除
问题：缺乏原子性
7 优化分布式锁_lua脚本保证删除的原子性
首先我们先简单介绍一下lua脚本的基本知识（lua脚本是c语言）
定义变量：
全局变量：a = 11
局部变量：local b = 22
redis不允许lua脚本创建全局变量，只能声明局部变量
流程控制:
if(exp) then业务逻辑elseif(exp) then业务逻辑else业务逻辑end
redis中执行lua脚本:
eval script numkeys keys[] args[] ： eval指令的输出不是lua脚本的打印而是lua脚本的返回值
script：lua脚本字符串，定义动态变量：keys[1] argv[1]
numkeys：key数组的元素个数
keys：keys数组
args：argv数组
redis集群执行lua脚本可能会报错：如果所有keys不在同一个分片上，lua脚本就会报错：解决方案是:
keys只传一个
可以使用cluster keyslot bb{xx}
删除lua脚本：
if redis.call('get', keys[1]) == argv[1] then return redis.call('del', keys[1]) else return 0 endpublic void testlock() { // 1. 从redis中获取锁,setnx string uuid = uuid.randomuuid().tostring(); boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, uuid, 3, timeunit.seconds); if (lock) { //与之前相同代码略过 ... // 2. 释放锁 del string script = if redis.call('get', keys[1]) == argv[1] then return redis.call('del', keys[1]) else return 0 end; this.redistemplate.execute(new defaultredisscript(script), arrays.aslist(lock), uuid); } else { // 3. 每隔1秒钟回调一次，再次尝试获取锁 try { thread.sleep(1000); testlock(); } catch (interruptedexception e) { e.printstacktrace(); } }}
8 优化分布式锁_可以重入
上述加锁命令使用了 setnx ，一旦键存在就无法再设置成功，这就导致后续同一线程内继续加锁，将会加锁失败。当一个线程执行一段代码成功获取锁之后，继续执行时，又遇到加锁的子任务代码，可重入性就保证线程能继续执行，而不可重入就是需要等待锁释放之后，再次获取锁成功，才能继续往下执行。
可重入锁最大特性就是计数，计算加锁的次数。所以当可重入锁需要在分布式环境实现时，我们也就需要统计加锁次数。
我们基于redis hash 实现方案 ：
redis 提供了 hash （哈希表）这种可以存储键值对数据结构。所以我们可以使用 redis hash 存储的锁的重入次数，然后利用 lua 脚本判断逻辑。
加锁
if (redis.call('exists', keys[1]) == 0 or redis.call('hexists', keys[1], argv[1]) == 1) then redis.call('hincrby', keys[1], argv[1], 1); redis.call('expire', keys[1], argv[2]); return 1;else return 0;end
假设值为：keys:[lock], argv[uuid, expire]
如果锁不存在或者这是自己的锁，就通过hincrby（不存在新增，存在就加1）获取锁或者锁次数加1。 代码实例如下：
private boolean trylock(string lockname, string uuid, long expire){ string script = if (redis.call('exists', keys[1]) == 0 or redis.call('hexists', keys[1], argv[1]) == 1) + then + redis.call('hincrby', keys[1], argv[1], 1); + redis.call('expire', keys[1], argv[2]); + return 1; + else + return 0; + end; if (!this.redistemplate.execute(new defaultredisscript(script, boolean.class), arrays.aslist(lockname), uuid, expire.tostring())){ try { // 没有获取到锁，重试 thread.sleep(200); trylock(lockname, uuid, expire); } catch (interruptedexception e) { e.printstacktrace(); } } // 获取到锁，返回true return true;}
解锁
-- 判断 hash set 可重入 key 的值是否等于 0-- 如果为 nil 代表 自己的锁已不存在，在尝试解其他线程的锁，解锁失败-- 如果为 0 代表 可重入次数被减 1-- 如果为 1 代表 该可重入 key 解锁成功if (redis.call('hexists', keys[1], argv[1]) == 0) then return nil;end;-- 小于等于 0 代表可以解锁if (redis.call('hincrby', keys[1], argv[1], -1) > 0) then return 0;else redis.call('del', keys[1]); return 1;end;
这里之所以没有跟加锁一样使用 boolean ,这是因为解锁 lua 脚本中，三个返回值含义如下：
1 代表解锁成功，锁被释放
0 代表可重入次数被减 1
null 代表其他线程尝试解锁，解锁失败
如果返回值使用 boolean，spring-data-redis 进行类型转换时将会把 null 转为 false，这就会影响我们逻辑判断，所以返回类型只好使用 long。
private void unlock(string lockname, string uuid){ string script = if (redis.call('hexists', keys[1], argv[1]) == 0) then + return nil; + end; + if (redis.call('hincrby', keys[1], argv[1], -1) > 0) then + return 0; + else + redis.call('del', keys[1]); + return 1; + end;; // 这里之所以没有跟加锁一样使用 boolean ,这是因为解锁 lua 脚本中，三个返回值含义如下： // 1 代表解锁成功，锁被释放 // 0 代表可重入次数被减 1 // null 代表其他线程尝试解锁，解锁失败 long result = this.redistemplate.execute(new defaultredisscript(script, long.class), lists.newarraylist(lockname), uuid); // 如果未返回值，代表尝试解其他线程的锁 if (result == null) { throw new illegalmonitorstateexception(attempt to unlock lock, not locked by lockname: + lockname + with request: + uuid); }}
使用
public void testlock() { // 加锁 string uuid = uuid.randomuuid().tostring(); boolean lock = this.trylock(lock, uuid, 300l); if (lock) { // 读取redis中的num值 string numstring = this.redistemplate.opsforvalue().get(num); if (stringutils.isblank(numstring)) { return; } // ++操作 integer num = integer.parseint(numstring); num++; // 放入redis this.redistemplate.opsforvalue().set(num, string.valueof(num)); // 测试可重入性 this.testsublock(uuid); // 释放锁 this.unlock(lock, uuid); }}// 测试可重入性private void testsublock(string uuid){ // 加锁 boolean lock = this.trylock(lock, uuid, 300l); if (lock) { system.out.println(分布式可重入锁。。。); this.unlock(lock, uuid); }}
9 优化分布式锁_自动续期
a线程超时时间设为10s（为了解决死锁问题），但代码执行时间可能需要30s，然后redis服务端10s后将锁删除。 此时，b线程恰好申请锁，redis服务端不存在该锁，可以申请，也执行了代码。
那么问题来了， a、b线程都同时获取到锁并执行业务逻辑，这与分布式锁最基本的性质相违背：在任意一个时刻，只有一个客户端持有锁（即独享排他）。
锁延期方法：开启子线程执行延期
/** * 锁延期 * 线程等待超时时间的2/3时间后,执行锁延时代码,直到业务逻辑执行完毕,因此在此过程中,其他线程无法获取到锁,保证了线程安全性 * @param lockname * @param expire 单位：毫秒 */private void renewtime(string lockname, string uuid, long expire){ string script = if(redis.call('hexists', keys[1], argv[1]) == 1) then redis.call('expire', keys[1], argv[2]); return 1; else return 0; end; new thread(() -> { while (this.redistemplate.execute(new defaultredisscript(script, boolean.class), lists.newarraylist(lockname), uuid, expire.tostring())){ try { // 到达过期时间的2/3时间，自动续期 thread.sleep(expire / 3); } catch (interruptedexception e) { e.printstacktrace(); } } }).start();}
获取锁成功后，调用延期方法给锁 定时延期：
private boolean trylock(string lockname, string uuid, long expire){ string script = if (redis.call('exists', keys[1]) == 0 or redis.call('hexists', keys[1], argv[1]) == 1) + then + redis.call('hincrby', keys[1], argv[1], 1); + redis.call('expire', keys[1], argv[2]); + return 1; + else + return 0; + end; if (!this.redistemplate.execute(new defaultredisscript(script, boolean.class), arrays.aslist(lockname), uuid, expire.tostring())){ try { // 没有获取到锁，重试 thread.sleep(200); trylock(lockname, uuid, expire); } catch (interruptedexception e) { e.printstacktrace(); } } // 锁续期 this.renewtime(lockname, uuid, expire * 1000); // 获取到锁，返回true return true;}
10 优化分布式锁_redlock算法
redis集群状态下的问题：
客户端a从master获取到锁
在master将锁同步到slave之前，master宕掉了。
slave节点被晋级为master节点
客户端b取得了同一个资源被客户端a已经获取到的另外一个锁。安全失效 解决集群下锁失效，参照redis官方网站针对redlock文档：redis.io/topics/dist… [1]
11 本地锁会出现的问题
我们知道java中有synchronized、lock锁、读写锁readwritelock，众所周知这些锁都是本地锁。
提到锁就不得不提juc:java.util.concurrent包，又称concurrent包。jdk1.5提供，为多线程高并发编程而提供的包,但此文章的场景是分布式场景，后续会出juc的文章。
简单的介绍一下synchronized及lock锁
synchronized是一个关键字，lock是一个接口，reentrantlock是实现了lock接口的一个类
reentrantlock：悲观的独占的互斥的排他的可公平可不公平的可重入锁
synchronized：悲观的独占的互斥的排他的非公平的可重入锁
准备
redis、ab工具(压测)
不使用任何锁的情况下
我们首先创建一个测试方法，testnolock
@getmapping(/test)public void testnolock(){ string count = (string) this.redistemplate.opsforvalue().get(count); if (count == null){ //没有值直接返回 return; } // 有值就转成成int int number = integer.parseint(count); // 把redis中的num值+1 this.redistemplate.opsforvalue().set(count, string.valueof(++number));}
测试之前的查看值为1
@getmapping(/getcount)public string getcount(){ string count = string.valueof(this.redistemplate.opsforvalue().get(count)); return count; //1}
接下来使用ab压力测试工具
// ab -n（一次发送的请求数） -c（请求的并发数） 访问路径ab -n100 -c50 http://127.0.0.1:8080/test/test
再次查询结果为6，说明问题很大
使用本地锁
public synchronized void testnolock(){ string count = string.valueof(this.redistemplate.opsforvalue().get(count)); if (null.equals(count)){ //没有值直接返回 return; } // 有值就转成成int int number = integer.parseint(count); // 把redis中的num值+1 this.redistemplate.opsforvalue().set(count, string.valueof(++number));}
再次使用ab压力测试工具
ab -n100 -c50 http://127.0.0.1:8080/test/test
此次结果为106，说明结果是正确的，看样子结果是非常完美的，但是真的很完美吗？
使用集群+本地锁
我们只需要在idea中在启动俩个服务，修改端口号，三个运行实例的名称是相同的，并且网关的配置就是通过服务名在负载均衡，所以我们只需要访问网关，网关就会给我们做负载均衡了。
再次使用ab压力测试工具（将count重置为1）
ab -n100 -c50 http://127.0.0.1:8080/test/test
此次的结果为58！！！
到此我们可以知道，本地锁是有局限性的。
Dialog IC解决方案帮助智能家居更互联和更安全
负线性稳压器具有0.8 µV RMS噪声和74dB电源抑制比
交换机和AI有什么关系？
可穿戴技术与功能
基于迪文T5L0 ASIC的摄像头方案
Java手写分布式锁的实现
全球产业格局巨变，德系与美系背道而驰
九阳电磁炉故障代码 (e0,e1,e2,e7,e8)
Microvast开发适用于多种车型的软包电池，可在20分钟充满电
PC彻底不行了？苹果微软不约而同瞄准了高端市场
尼得科传动技术推出使用了精密控制用减速机“FLEXWAVE”的超扁平执行器
镍电池极片自动化控制要点
并行端口接口为低压系统供电
测量万用表电阻挡接地问题
温控电路的设计与调试
苹果手机电池门升温_苹果手机电池爆炸引关注_苹果电池爆炸原因竟是“降频门”？
单片2.4GHz无线收发芯片nRF24E1的应用
谷歌无人机快递Wing进军芬兰将送餐馆外卖
中国联通举行综合改革合作协议签约仪式 实现国有资产保值增值
自主全向仓鼠球机器人可飞越障碍物
随着互联网业务的发展，原本单机部署的系统演化成如今的分布式集群系统后，由于分布式系统多线程，多进程并且分布在不同的机器上，这会使原本的单机锁失效，而且单纯的java api并不能提供分布式锁的能力，为了解决这个问题就需要一种跨jvm的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题。
本文将从实现本地锁所产生的问题入手，从而介绍分布式锁主流的实现方案，重点实现基于redis的分布式锁。
本地锁会出现的问题（此篇幅代码图片过多，因此放在最后）
part2 分布式锁的实现
1 分布式锁主要的实现有：
基于数据库实现分布式锁
基于缓存（redis等,本文基于redis实现手写分布式锁 ，因为这样可以更好的理解分布式锁的原理及实现，当然也可以使用redisson）
基于zookeeper
2 每种分布式锁的解决方案都有各自的优缺点
性能角度：redis > zk > mysql
安全角度：zk > redis == mysql
难易程度：zk > redis > mysql
3 分布式锁要具备的特点：
独占排他互斥
可以通过 setnx （redis命令：执行多次，只有一次能够成功）
set key value ex 3 nx
防死锁发生
请求获取到锁之后，服务器挂掉了，导致锁无法释放：给lock锁添加过期时间
可以通过redis命令 expire
或者通过 set key value ex 3 nx
保证原子性
redis是单线程的，接受或者执行指令遵循one-by-one原则。只要指令之间不被插入其他指令即可保证原子性，lua脚本批量发送多个指令给redis服务器，lua脚本也可以实现一些业务逻辑，redis集成了lua脚本，可以直接使用eval指令执行lua脚本。
获取锁和设置过期时间之间
判断和删除之间：lua脚本
防误删：
uuid给每个线程的锁添加唯一标识
自动续期
可重入：hash数据结构 + lua脚本
集群情况下，可能导致锁失效：redlock算法（redis特有的）
一个请求从主中获取到锁，从还没来得及同步数据，主就挂掉了，从就升级为新主，新的请求就可以从新主中获取锁
4 基于redis分布式锁的基本实现
我们可以借助于redis中的命令setnx(key, value)，key不存在就新增，存在就什么都不做。同时有多个客户端发送setnx命令，只有一个客户端可以成功，返回1（true）；其他的客户端返回0（false），流程图如下图所示：
多个客户端同时尝试获取锁（setnx）
获取成功，执行业务逻辑，执行完成释放锁（del）
其他客户端等待重试
代码实现：
public void testlock() { // 1. 从redis中获取锁,setnx boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, 111); if (lock) { // 查询redis中的num值 string value = this.redistemplate.opsforvalue().get(num); // 没有该值return if (stringutils.isblank(value)){ return ; } // 有值就转成成int int num = integer.parseint(value); // 把redis中的num值+1 this.redistemplate.opsforvalue().set(num, string.valueof(++num)); // 2. 释放锁 del this.redistemplate.delete(lock); } else { // 3. 每隔1秒钟回调一次，再次尝试获取锁 try { thread.sleep(1000); testlock(); } catch (interruptedexception e) { e.printstacktrace(); } }}
那么以上代码是否可以解决全部问题呢？ 显示是不能的，我们假设setnx刚好获取到锁，业务逻辑出现异常，导致锁无法释放，怎么办呢？
5 优化分布式锁_设置过期时间
设置过期有俩种方式可以选择：
通过expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）
在set时指定过期时间（推荐）
代码实现优化就是在设置锁的时候设置过期时间:
public void testlock() { // 1. 从redis中获取锁,setnx boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, 111,3, timeunit.minutes); if (lock) { //与之前相同代码略过 ... }}
那么还会不会存在问题呢?
场景：如果业务逻辑的执行时间是7s。执行流程如下:
index1业务逻辑没执行完，3秒后锁被自动释放。
index2获取到锁，执行业务逻辑，3秒后锁被自动释放。
index3获取到锁，执行业务逻辑
index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。
最终等于没锁的情况。
解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁
6 优化分布式锁_防止误删除
public void testlock() { // 1. 从redis中获取锁,setnx string uuid = uuid.randomuuid().tostring(); boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, uuid,3, timeunit.minutes); if (lock) { //与之前相同代码略过 ... // 2. 释放锁 del if (stringutils.equals(redistemplate.opsforvalue().get(lock),uuid)){ this.redistemplate.delete(lock); } }}
场景：
index1执行删除时，查询到的lock值确实和uuid相等
index1执行删除前，lock刚好过期时间已到，被redis自动释放
index2获取了lock
index1执行删除，此时会把index2的lock删除
问题：缺乏原子性
7 优化分布式锁_lua脚本保证删除的原子性
首先我们先简单介绍一下lua脚本的基本知识（lua脚本是c语言）
定义变量：
全局变量：a = 11
局部变量：local b = 22
redis不允许lua脚本创建全局变量，只能声明局部变量
流程控制:
if(exp) then业务逻辑elseif(exp) then业务逻辑else业务逻辑end
redis中执行lua脚本:
eval script numkeys keys[] args[] ： eval指令的输出不是lua脚本的打印而是lua脚本的返回值
script：lua脚本字符串，定义动态变量：keys[1] argv[1]
numkeys：key数组的元素个数
keys：keys数组
args：argv数组
redis集群执行lua脚本可能会报错：如果所有keys不在同一个分片上，lua脚本就会报错：解决方案是:
keys只传一个
可以使用cluster keyslot bb{xx}
删除lua脚本：
if redis.call('get', keys[1]) == argv[1] then return redis.call('del', keys[1]) else return 0 endpublic void testlock() { // 1. 从redis中获取锁,setnx string uuid = uuid.randomuuid().tostring(); boolean lock = this.redistemplate.opsforvalue().setifabsent(lock, uuid, 3, timeunit.seconds); if (lock) { //与之前相同代码略过 ... // 2. 释放锁 del string script = if redis.call('get', keys[1]) == argv[1] then return redis.call('del', keys[1]) else return 0 end; this.redistemplate.execute(new defaultredisscript(script), arrays.aslist(lock), uuid); } else { // 3. 每隔1秒钟回调一次，再次尝试获取锁 try { thread.sleep(1000); testlock(); } catch (interruptedexception e) { e.printstacktrace(); } }}
8 优化分布式锁_可以重入
上述加锁命令使用了 setnx ，一旦键存在就无法再设置成功，这就导致后续同一线程内继续加锁，将会加锁失败。当一个线程执行一段代码成功获取锁之后，继续执行时，又遇到加锁的子任务代码，可重入性就保证线程能继续执行，而不可重入就是需要等待锁释放之后，再次获取锁成功，才能继续往下执行。
可重入锁最大特性就是计数，计算加锁的次数。所以当可重入锁需要在分布式环境实现时，我们也就需要统计加锁次数。
我们基于redis hash 实现方案 ：
redis 提供了 hash （哈希表）这种可以存储键值对数据结构。所以我们可以使用 redis hash 存储的锁的重入次数，然后利用 lua 脚本判断逻辑。
加锁
if (redis.call('exists', keys[1]) == 0 or redis.call('hexists', keys[1], argv[1]) == 1) then redis.call('hincrby', keys[1], argv[1], 1); redis.call('expire', keys[1], argv[2]); return 1;else return 0;end
假设值为：keys:[lock], argv[uuid, expire]
如果锁不存在或者这是自己的锁，就通过hincrby（不存在新增，存在就加1）获取锁或者锁次数加1。 代码实例如下：
private boolean trylock(string lockname, string uuid, long expire){ string script = if (redis.call('exists', keys[1]) == 0 or redis.call('hexists', keys[1], argv[1]) == 1) + then + redis.call('hincrby', keys[1], argv[1], 1); + redis.call('expire', keys[1], argv[2]); + return 1; + else + return 0; + end; if (!this.redistemplate.execute(new defaultredisscript(script, boolean.class), arrays.aslist(lockname), uuid, expire.tostring())){ try { // 没有获取到锁，重试 thread.sleep(200); trylock(lockname, uuid, expire); } catch (interruptedexception e) { e.printstacktrace(); } } // 获取到锁，返回true return true;}
解锁
-- 判断 hash set 可重入 key 的值是否等于 0-- 如果为 nil 代表 自己的锁已不存在，在尝试解其他线程的锁，解锁失败-- 如果为 0 代表 可重入次数被减 1-- 如果为 1 代表 该可重入 key 解锁成功if (redis.call('hexists', keys[1], argv[1]) == 0) then return nil;end;-- 小于等于 0 代表可以解锁if (redis.call('hincrby', keys[1], argv[1], -1) > 0) then return 0;else redis.call('del', keys[1]); return 1;end;
这里之所以没有跟加锁一样使用 boolean ,这是因为解锁 lua 脚本中，三个返回值含义如下：
1 代表解锁成功，锁被释放
0 代表可重入次数被减 1
null 代表其他线程尝试解锁，解锁失败
如果返回值使用 boolean，spring-data-redis 进行类型转换时将会把 null 转为 false，这就会影响我们逻辑判断，所以返回类型只好使用 long。
private void unlock(string lockname, string uuid){ string script = if (redis.call('hexists', keys[1], argv[1]) == 0) then + return nil; + end; + if (redis.call('hincrby', keys[1], argv[1], -1) > 0) then + return 0; + else + redis.call('del', keys[1]); + return 1; + end;; // 这里之所以没有跟加锁一样使用 boolean ,这是因为解锁 lua 脚本中，三个返回值含义如下： // 1 代表解锁成功，锁被释放 // 0 代表可重入次数被减 1 // null 代表其他线程尝试解锁，解锁失败 long result = this.redistemplate.execute(new defaultredisscript(script, long.class), lists.newarraylist(lockname), uuid); // 如果未返回值，代表尝试解其他线程的锁 if (result == null) { throw new illegalmonitorstateexception(attempt to unlock lock, not locked by lockname: + lockname + with request: + uuid); }}
使用
public void testlock() { // 加锁 string uuid = uuid.randomuuid().tostring(); boolean lock = this.trylock(lock, uuid, 300l); if (lock) { // 读取redis中的num值 string numstring = this.redistemplate.opsforvalue().get(num); if (stringutils.isblank(numstring)) { return; } // ++操作 integer num = integer.parseint(numstring); num++; // 放入redis this.redistemplate.opsforvalue().set(num, string.valueof(num)); // 测试可重入性 this.testsublock(uuid); // 释放锁 this.unlock(lock, uuid); }}// 测试可重入性private void testsublock(string uuid){ // 加锁 boolean lock = this.trylock(lock, uuid, 300l); if (lock) { system.out.println(分布式可重入锁。。。); this.unlock(lock, uuid); }}
9 优化分布式锁_自动续期
a线程超时时间设为10s（为了解决死锁问题），但代码执行时间可能需要30s，然后redis服务端10s后将锁删除。 此时，b线程恰好申请锁，redis服务端不存在该锁，可以申请，也执行了代码。
那么问题来了， a、b线程都同时获取到锁并执行业务逻辑，这与分布式锁最基本的性质相违背：在任意一个时刻，只有一个客户端持有锁（即独享排他）。
锁延期方法：开启子线程执行延期
/** * 锁延期 * 线程等待超时时间的2/3时间后,执行锁延时代码,直到业务逻辑执行完毕,因此在此过程中,其他线程无法获取到锁,保证了线程安全性 * @param lockname * @param expire 单位：毫秒 */private void renewtime(string lockname, string uuid, long expire){ string script = if(redis.call('hexists', keys[1], argv[1]) == 1) then redis.call('expire', keys[1], argv[2]); return 1; else return 0; end; new thread(() -> { while (this.redistemplate.execute(new defaultredisscript(script, boolean.class), lists.newarraylist(lockname), uuid, expire.tostring())){ try { // 到达过期时间的2/3时间，自动续期 thread.sleep(expire / 3); } catch (interruptedexception e) { e.printstacktrace(); } } }).start();}
获取锁成功后，调用延期方法给锁 定时延期：
private boolean trylock(string lockname, string uuid, long expire){ string script = if (redis.call('exists', keys[1]) == 0 or redis.call('hexists', keys[1], argv[1]) == 1) + then + redis.call('hincrby', keys[1], argv[1], 1); + redis.call('expire', keys[1], argv[2]); + return 1; + else + return 0; + end; if (!this.redistemplate.execute(new defaultredisscript(script, boolean.class), arrays.aslist(lockname), uuid, expire.tostring())){ try { // 没有获取到锁，重试 thread.sleep(200); trylock(lockname, uuid, expire); } catch (interruptedexception e) { e.printstacktrace(); } } // 锁续期 this.renewtime(lockname, uuid, expire * 1000); // 获取到锁，返回true return true;}
10 优化分布式锁_redlock算法
redis集群状态下的问题：
客户端a从master获取到锁
在master将锁同步到slave之前，master宕掉了。
slave节点被晋级为master节点
客户端b取得了同一个资源被客户端a已经获取到的另外一个锁。安全失效 解决集群下锁失效，参照redis官方网站针对redlock文档：redis.io/topics/dist… [1]
11 本地锁会出现的问题
我们知道java中有synchronized、lock锁、读写锁readwritelock，众所周知这些锁都是本地锁。
提到锁就不得不提juc:java.util.concurrent包，又称concurrent包。jdk1.5提供，为多线程高并发编程而提供的包,但此文章的场景是分布式场景，后续会出juc的文章。
简单的介绍一下synchronized及lock锁
synchronized是一个关键字，lock是一个接口，reentrantlock是实现了lock接口的一个类
reentrantlock：悲观的独占的互斥的排他的可公平可不公平的可重入锁
synchronized：悲观的独占的互斥的排他的非公平的可重入锁
准备
redis、ab工具(压测)
不使用任何锁的情况下
我们首先创建一个测试方法，testnolock
@getmapping(/test)public void testnolock(){ string count = (string) this.redistemplate.opsforvalue().get(count); if (count == null){ //没有值直接返回 return; } // 有值就转成成int int number = integer.parseint(count); // 把redis中的num值+1 this.redistemplate.opsforvalue().set(count, string.valueof(++number));}
测试之前的查看值为1
@getmapping(/getcount)public string getcount(){ string count = string.valueof(this.redistemplate.opsforvalue().get(count)); return count; //1}
接下来使用ab压力测试工具
// ab -n（一次发送的请求数） -c（请求的并发数） 访问路径ab -n100 -c50 http://127.0.0.1:8080/test/test
再次查询结果为6，说明问题很大
使用本地锁
public synchronized void testnolock(){ string count = string.valueof(this.redistemplate.opsforvalue().get(count)); if (null.equals(count)){ //没有值直接返回 return; } // 有值就转成成int int number = integer.parseint(count); // 把redis中的num值+1 this.redistemplate.opsforvalue().set(count, string.valueof(++number));}
再次使用ab压力测试工具
ab -n100 -c50 http://127.0.0.1:8080/test/test
此次结果为106，说明结果是正确的，看样子结果是非常完美的，但是真的很完美吗？
使用集群+本地锁
我们只需要在idea中在启动俩个服务，修改端口号，三个运行实例的名称是相同的，并且网关的配置就是通过服务名在负载均衡，所以我们只需要访问网关，网关就会给我们做负载均衡了。
再次使用ab压力测试工具（将count重置为1）
ab -n100 -c50 http://127.0.0.1:8080/test/test
此次的结果为58！！！
到此我们可以知道，本地锁是有局限性的。
Dialog IC解决方案帮助智能家居更互联和更安全
负线性稳压器具有0.8 µV RMS噪声和74dB电源抑制比
交换机和AI有什么关系？
可穿戴技术与功能
基于迪文T5L0 ASIC的摄像头方案
Java手写分布式锁的实现
全球产业格局巨变，德系与美系背道而驰
九阳电磁炉故障代码 (e0,e1,e2,e7,e8)
Microvast开发适用于多种车型的软包电池，可在20分钟充满电
PC彻底不行了？苹果微软不约而同瞄准了高端市场
尼得科传动技术推出使用了精密控制用减速机“FLEXWAVE”的超扁平执行器
镍电池极片自动化控制要点
并行端口接口为低压系统供电
测量万用表电阻挡接地问题
温控电路的设计与调试
苹果手机电池门升温_苹果手机电池爆炸引关注_苹果电池爆炸原因竟是“降频门”？
单片2.4GHz无线收发芯片nRF24E1的应用
谷歌无人机快递Wing进军芬兰将送餐馆外卖
中国联通举行综合改革合作协议签约仪式 实现国有资产保值增值
自主全向仓鼠球机器人可飞越障碍物