分布式锁实现方案-202509120200

分布式锁实现方案

一、概述

锁常常用于多线程并发的场景下保证数据的一致性，例如防止超卖、一人一单等场景需求 。

通过加锁可以解决在单机情况下安全问题，但是在集群模式下就不行了。集群模式，即部署了多

个服务器、并配置了负载均衡后，原来加的锁会失效，具体原因如下：

由于现在我们部署了多个tomcat，每个tomcat都有一个属于自己的jvm，那么假设在服务器

A的tomcat内部，有两个线程，这两个线程由于使用的是同一份代码，那么他们的锁对象是同一

个，是可以实现互斥的；但是如果现在是服务器B的tomcat内部，又有两个线程，但是他们的锁

对象写的虽然和服务器A一样，但是锁对象却不是同一个，所以线程3和线程4可以实现互斥，但

是却无法和线程1和线程2实现互斥。这就是集群环境下，syn锁失效的原因，在这种情况下，我

们就需要使用分布式锁来解决这个问题。

在分布式系统中，当不同进程或线程一起访问共享资源时，会造成资源争抢，如果不加以控制

的话，就会引发程序错乱。而分布式锁它采用了一种互斥机制来防止线程或进程间相互干扰，从

而保证了数据的一致性

常见的分布式锁实现方案

基于 Redis 实现分布式锁

基于 Zookeeper 实现分布式锁

基于Redisson实现分布式锁，Redission是一个基于Redis实现的Java分布式对象存储和缓存

框架。它提供了丰富的分布式数据结构和服务。例如：分布式锁、分布式队列、分布式Rate

Limiter等。

二、Redis实现分布式锁

1.最简单的版本：setnx key value

基于setnx命令的特性，我们就可以实现一个最简单的分布式锁了。我们通过向Redis发

送 setnx 命令，然后判断Redis返回的结果是否为1，结果是1就表示setnx成功了，那本次就

获得锁了，可以继续执行业务逻辑；如果结果是0，则表示setnx失败了，那本次就没有获取

到锁，可以通过循环的方式一直尝试获取锁，直到其他客户端释放了锁（delete掉key）后，

就可以正常执行setnx命令获取到锁。执行流程如下图所示：

存在的问题：死锁。这种方式虽然实现了分布式锁的功能，但有一个很明显的问题：没有给

key设置过期时间，万一程序在发送delete命令释放锁之前宕机了，那么这个key就会永久的存储

在Redis中了，其他客户端也永远获取不到这把锁了

解决方案：设置key的过期时间，SET key value [expiration EX seconds|PX milliseconds] [N

X|XX]

EX: 设置超时时间，单位是秒

PX: 设置超时时间，单位是毫秒

NX: IF NOT EXIST 的缩写，只有 KEY不存在的前提下 才会设置值

XX: IF EXIST 的缩写，只有在 KEY存在的前提下 才会设置值

存在的问题：超时失效(下面可以用Redisson来实现锁的自动续期)

存在的问题：锁误删

解决方案：要解决上面的问题，最省事的做法就是把锁的过期时间设置长一点，要远大于业

务处理时间，但这样就会严重影响系统的性能，假如一台服务器在释放锁之前宕机了，而锁的超

时时间设置了一个小时，那么在这一个小时内，其他线程访问这个服务时就一直阻塞在那里。所

以，一般不推荐使用这种方式。

另一种解决方法就是在set key value ex seconds nx时，把value设置成一个唯一值，每个线

程的value都不一样，在删除key之前，先通过get key命令得到value，然后判断value是否是自己

线程生成的，如果是，则删除掉key释放锁，如果不是，则不删除key。

当业务处理还没结束的时候，key自动过期了，也可以正常释放自己的锁，不影响其他线

程。

二次升级后的方案看起来似乎已经没什么问题了，但其实不然。仔细分析流程后我们发现，

判断锁是否属于当前线程和释放锁两个步骤并不是原子操作。正常来说，如果线程1通过get操作

从Redis中得到的value是123，那么就会执行删除锁的操作，但假如在执行删除锁的动作之前，

系统卡顿了几秒钟，恰好在这几秒钟内，key自动过期了，线程2就顺利获取到锁开始执行自己的

逻辑了，此时，线程1卡顿恢复了，开始继续执行删除锁的动作，那么此时删除的还是线程2的

锁。

解决方案：

Lua脚本：Redis提供了Lua脚本的支持。Lua脚本是一种轻量小巧的脚本语言，它支持原子性操作，

Redis会将整个Lua脚本作为一个整体执行，中间不会被其他请求插入，因此Redis执行Lua脚本是一个原

子操作。

使用 Redis 的  SETNX  命令（SET if Not eXists）尝试设置一个键值对。

仅当  KEYS[1] （锁名称）不存在时，才将其值设置为  ARGV[1] （通常是客户端唯一标

识）。

返回  1  表示设置成功（获得锁），0  表示键已存在（锁被占用）。

判断  SETNX  是否成功（返回值为  1 ）。

若获得锁，立即为  KEYS[1]  设置过期时间  ARGV[2] （单位：秒）。

作用：防止死锁（如客户端崩溃后锁无法释放）。

返回  1 ，表示成功获得锁。

返回  0 ，表示获取锁失败（锁已被其他客户端持有）

在上面的流程中，我们把get key value、判断value是否属于当前线程、删除锁这三步写到Lu

a脚本中，使它们变成一个整体交个Redis执行，改造后流程如下：

if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +

" redis.call('expire',KEYS[1],ARGV[2]) return 1 else

return 0 end"

这样改造之后，就解决了释放锁时取值、判断值、删除锁等多个步骤无法保证原子操作的问

题了。至此，使用Redis+Lua脚本实现简单的分布式锁可行的方案。

Lua脚本实现可重入锁：

既然Lua脚本可以在释放锁时使用，那肯定也能在加锁时使用，而且一般情况下，推荐使用L

ua脚本，因为在使用上面set key value ex seconds nx命令加锁时，并不能做到重入锁的效果，也

就是当一个线程获取到锁后，在没有释放这把锁之前，当前线程自己也无法再获得这把锁，这显

然会影响系统的性能。使用Lua脚本就可以解决这个问题，我们可以在Lua脚本中先判断锁（key）

是否存在，如果存在则再判断持有这把锁的线程是否是当前线程，如果不是则加锁失败，否则当

前线程再次持有这把锁，并把锁的重入次数+1。在释放锁时，也是先判断持有锁的线程是否是当

前线程，如果是则将锁的重入次数-1，直至重入次数减至0，即可删除该锁（key）。

实际项目开发中，基本上不需要开发者去写这些分布式锁的实现逻辑，已经有较为成熟且流

行的第三方工具-Redisson。

二、Redisson实现分布式锁

1. Redisson实现Redis分布式锁的底层原理

现在某个客户端要加锁。如果该客户端面对的是一个redis cluster集群，他首先会根据hash

节点选择一台机器。这里注意，仅仅只是选择一台机器！这点很关键！紧接着，就会发送一段lua

脚本到redis上，那段lua脚本如下所示：

KEYS[1]代表的是你加锁的那个key，比如说：RLock lock = redisson.getLock("LockName ");

这里你自己设置了加锁的那个锁key就是“LockName ”。

@Override

public boolean tryLock() {

return get(tryLockAsync());

}

RFuture tryLockInnerAsync(long waitTime, long leaseTime,

TimeUnit unit, long threadId, RedisStrictCommand command) {

return evalWriteAsync(getRawName(), LongCodec.INSTANCE,

command,

"if (redis.call('exists', KEYS[1]) == 0) then " +

"redis.call('hincrby', KEYS[1], ARGV[2], 1); "

+

"redis.call('pexpire', KEYS[1], ARGV[1]); " +

"return nil; " +

"end; " +

"if (redis.call('hexists', KEYS[1], ARGV[2])

== 1) then " +

"redis.call('hincrby', KEYS[1], ARGV[2], 1); "

+

"redis.call('pexpire', KEYS[1], ARGV[1]); " +

"return nil; " +

"end; " +

"return redis.call('pttl', KEYS[1]);",

Collections.singletonList(getRawName()),

unit.toMillis(leaseTime), getLockName(threadId));

}

ARGV[2]代表的是加锁的客户端的ID，类似于下面这样：8743c9c0-0795-4907-87fd-6c719a

6b4586:1

ARGV[1]代表的就是锁key的默认生存时间，默认30秒通过这个命令设置一个hash数据结

构，这行命令执行后，会出现一个类似下面的数据结构：

hset LockName 8743c9c0-0795-4907-87fd-6c719a6b4586:1 1

检查锁(KEYS[1])是否存在

如果不存在(==0)，表示可以获取锁

使用hincrby 命令在哈希表中为客户端标识(ARGV[2])设置值1(表示第一次获取锁)

设置锁的过期时间(ARGV[1]毫秒)

返回nil表示成功获取锁

上述就代表8743c9c0-0795-4907-87fd-6c719a6b4586:1 这个客户端对LockName

这个锁key完成了加锁。

接着会执行pexpire myLock 30000 命令，设置LockName这个锁key的生存时间是30

秒。

2、锁互斥机制

这个时候，如果客户端2来尝试加锁，执行了同样的一段lua脚本，会咋样呢？很简单，

第一个if判断会执行exists myLock，发现myLock这个锁key已经存在了。

接着第二个if判断，判断一下，myLock锁key的hash数据结构中，是否包含客户端2的I

D，但是明显不是的，因为那里包含的是客户端1的ID。

所以，客户端2会获取到pttl myLock返回的一个数字，这个数字代表了myLock这个锁ke

y的剩余生存时间。比如还剩15000毫秒的生存时间。

此时客户端2会进入一个while循环，不停的尝试加锁（类似于自旋锁）。

3、可重入加锁机制

第一个if判断肯定不成立，exists myLock会显示锁key已经存在了。

第二个if判断会成立，因为myLock的hash数据结构中包含的那个ID，就是客户端1的那个I

D，也就是8743c9c0-0795-4907-87fd-6c719a6b4586:1

此时就会执行可重入加锁的逻辑，他会用：

incrby myLock 8743c9c0-0795-4907-87fd-6c71a6b4586:1 1

通过这个命令，对客户端1的加锁次数，累加1。

此时myLock数据结构变为下面这样（myLock的hash数据结构中的那个客户端ID，就对应着

加锁的次数）：

LockName:

{

"8743c9c0-0795-4907-87fd-6c719a6b4586 :1"1

}

如果前面两个条件都不满足，表示锁被其他客户端持有

返回锁的剩余生存时间(毫秒)，客户端可以根据这个值决定是否等待

4.Watch Dog自动延期机制

客户端1加锁的锁key默认生存时间才30秒，如果超过了30秒，客户端1还想一直持有这把

锁，怎么办呢？

简单！只要客户端1一旦加锁成功，就会启动一个watch dog看门狗，他是一个后台线程，会

每隔10秒检查一下，如果客户端1还持有锁key，那么就会不断的延长锁key的生存时间。

watch dog自动延期机制核心方法为：

LockName:

{

"8743c9c0-0795-4907-87fd-6c719a6b4586 :1"2

}

// 获取锁入口

void lock(long leaseTime, TimeUnit unit);

//leaseTime未设置时，默认为-1，触发WatchDog初始化，

public void lock() {

try {

lock(-1, null, false);

} // ...

// scheduleExpirationRenewal()启动定时任务：

protected void scheduleExpirationRenewal(long threadId) {

// 创建一个新的续期记录对象

ExpirationEntry entry = new ExpirationEntry();

// 尝试将续期记录存入全局MAP（如果不存在）

ExpirationEntry oldEntry =

(ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName()

, entry);

if (oldEntry != null) {

// 若已有续期任务：只需添加当前线程ID（可重入场景）

oldEntry.addThreadId(threadId);

} else {

// 首次创建续期任务：添加当前线程ID

entry.addThreadId(threadId);

try {

// 启动续期定时任务

this.renewExpiration();

} finally {

// 异常处理：线程中断时取消续期

if (Thread.currentThread().isInterrupted()) {

this.cancelExpirationRenewal(threadId);

}

}

}

}

private void renewExpiration() {

// 获取当前锁的续期记录

ExpirationEntry ee =

(ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());

if (ee != null) {

// 创建定时任务（在1/3锁有效期后触发）

Timeout task =

this.commandExecutor.getConnectionManager().newTimeout(new TimerTask()

{

public void run(Timeout timeout) throws Exception {

// 获取最新续期记录

ExpirationEntry ent =

(ExpirationEntry)RedissonBaseLock.EXPIRATION_RENEWAL_MAP.get(RedissonBa

seLock.this.getEntryName());

if (ent != null) {

// 获取第一个持有线程ID（即使可重入也只需续期一次）

Long threadId = ent.getFirstThreadId();

if (threadId != null) {

// 异步续期操作 执行如下Lua脚本

CompletionStage future =

RedissonBaseLock.this.renewExpirationAsync(threadId);

future.whenComplete((res, e) -> {

if (e != null) {

// 异常：续期失败（日志记录+清理）

RedissonBaseLock.log.error("Can't

update lock " + RedissonBaseLock.this.getRawName() + " expiration", e);

RedissonBaseLock.EXPIRATION_RENEWAL_MAP.remove(RedissonBaseLock.this.ge

tEntryName());

} else {

renewExpirationAsync() 通过Lua脚本延长锁的过期时间：

每间隔30/3s检查锁是否存在，若存在则重置过期时间为30秒（internalLockLeaseTim

e ）

if (res) {

// 续期成功：递归调用下一次续期

RedissonBaseLock.this.renewExpiration();

} else {

// 续期失败：锁已释放/不存在

RedissonBaseLock.this.cancelExpirationRenewal((Long)null);

}

}

});

}

}

}

}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);

// 将定时任务绑定到续期记录

ee.setTimeout(task);

}

}

// RedissonBaseLock.java

protected CompletionStage renewExpirationAsync(long

threadId) {

// 执行异步的Lua脚本操作

// KEYS[1]：锁在Redis中的键名（如"my_lock"）

//处理Long类型的编解码器

// 指定返回类型为布尔值

return this.evalWriteAsync(this.getRawName(),

LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,

"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) "

+// 检查锁是否存在且持有者匹配

"then redis.call('pexpire', KEYS[1], " +// 续期

操作

"ARGV[1]); return 1; " +

"end; return 0;",

Collections.singletonList(this.getRawName()),

this.internalLockLeaseTime,//

ARGV[1]：新的锁过期时间（毫秒）

this.getLockName(threadId));ARGV[2]：//锁持有者标识（UUID:threadId）

}

WatchDog的生命周期：

启动时机

锁获取成功：未指定leaseTime 时触发。

锁重入：同一线程重复获取锁时不会重复启动。

停止条件

锁释放：调用unlock() 时清理续期任务。

节点宕机：客户端进程终止，定时任务自动取消

5.解锁

如果执行lock.unlock()，就可以释放分布式锁，此时的业务逻辑也是非常简单的。其实

说白了，就是每次都对myLock数据结构中的那个加锁次数减1。如果发现加锁次数是0了，

说明这个客户端已经不再持有锁了，此时就会用del myLock命令，从redis里删除这个key。

然后呢，另外的客户端2就可以尝试完成加锁了。一般我们在生产系统中，可以用Redisson

框架提供的这个类库来基于redis进行分布式锁的加锁与释放锁。

protected RFuture unlockInnerAsync(long threadId) {

return evalWriteAsync( // 执行异步的Lua脚本操作

getRawName(), // KEYS[1]：锁在Redis中的键名

LongCodec.INSTANCE, // 编解码器（处理Long类型）

RedisCommands.EVAL_BOOLEAN, // 返回类型为布尔值

"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +

// [1] 验证锁持有者

"return nil;" +

// [2] 非法操作处理

"end; " +

"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);

" + // [3] 减少重入计数

"if (counter > 0) then " +

// [4] 判断是否完全释放

"redis.call('pexpire', KEYS[1], ARGV[2]); " +

// [5] 更新锁过期时间

"return 0; " +

// [6] 部分解锁（重入）

"else " +

"redis.call('del', KEYS[1]); " +

// [7] 完全释放锁

"redis.call('publish', KEYS[2], ARGV[1]); " +

// [8] 发布解锁通知

"return 1; " +

// [9] 完全解锁成功

"end; " +

"return nil;",

// [10] 兜底返回

三、DxGlobalLock注解

Arrays.asList(getRawName(), getChannelName()), // KEYS数组

LockPubSub.UNLOCK_MESSAGE, // ARGV[1]：解锁消息标识

internalLockLeaseTime, // ARGV[2]：锁过期时间（毫秒）

getLockName(threadId) // ARGV[3]：锁持有者标识

);

}

@Around(value = "pointcut()") //环绕通知，方法的前后添加分布式锁的逻辑

public Object doGlobalLock(ProceedingJoinPoint joinPoint) throws

Throwable {

//获取当前执行的方法名

String methodName = joinPoint.getSignature().getName();

//方法名为keyPair直接放行

if (methodName.equals("keyPair")) {

return joinPoint.proceed();

}

//方法的执行结果变量；

Object result;

// 获取方法签名信息

Class[] par = ((MethodSignature)

joinPoint.getSignature()).getParameterTypes();

Class targetClass = joinPoint.getTarget().getClass();

// 反射获取目标方法

Method objMethod = ClassUtil.getDeclaredMethod(targetClass,

methodName, par);

// 检查方法上的@DxGlobalLock注解

DxGlobalLock annotation =

AnnotationUtils.findAnnotation(objMethod, DxGlobalLock.class);

if (annotation == null) {

//无注解直接放行

return joinPoint.proceed();

}

//标记是否为写锁

boolean iswrite = false;

//锁名

String name = null;

try {

//锁构建的参数计数器

int keyCount = 0;

//获取锁名

name = annotation.name();

//构建锁key

StringBuilder key = new StringBuilder(annotation.key());

//处理参数参与构建锁的情况

if (!annotation.noParams()) {

if (key.toString().isEmpty()) {

// 获取方法入参 查找带有@DxGlobalLockParam注解的参数

boolean hasParams = false;

Annotation[][] parameterAnnotations =

objMethod.getParameterAnnotations();

//循环查找 考虑其父类

while (!hasParams) {

for (Annotation[] parameterAnnotation :

parameterAnnotations) {

for (Annotation an : parameterAnnotation) {

if

(an.annotationType().isAssignableFrom(DxGlobalLockParam.class)) {

hasParams = true;

}

}

}

if (!hasParams) {

try {

targetClass =

targetClass.getSuperclass();

objMethod =

targetClass.getMethod(methodName, par);

} catch (NoSuchMethodException exception) {

log.error("方法入参没有DxGlobalLockParam

注解,或者将该注解写在了接口上{}", methodName);

throw new RuntimeException("方法入参没有

DxGlobalLockParam注解,或者将该注解写在了接口上");

}

parameterAnnotations =

objMethod.getParameterAnnotations();

}

}

//处理参数

for (int i = 0; i < parameterAnnotations.length;

i++) {

//

for (Annotation an : parameterAnnotations[i]) {

if

(an.annotationType().isAssignableFrom(DxGlobalLockParam.class)) {

keyCount++;

// 获取入参

DxGlobalLockParam globalLockParam =

(DxGlobalLockParam) an;

Object params = joinPoint.getArgs()[i];

if (params == null) {

continue;

}

Class paramsClass =

params.getClass();

boolean wrapClass =

DxGlobalLockUtils.isWrapClass(paramsClass);

if (wrapClass) {

key.append(params);

} else {

String filedValue =

globalLockParam.value();

Object okey = null;

if (filedValue.length() > 1) {

//是否包含. 包含点代表取属性类的属性

if (filedValue.contains(".")) {

String[] split =

filedValue.split("\\.");

var targetObject = params;

for (int k = 0; k <

split.length; k++) {

// 反射获取

Field field =

paramsClass.getDeclaredField(split[k]);

field.setAccessible(true);

if (k == (split.length

okey =

field.get(targetObject);

} else {

targetObject =

field.get(targetObject);

paramsClass =

targetObject.getClass();

}

}

} else {

// 反射获取

Field field = null;

while (field == null) {

try {

field =

paramsClass.getDeclaredField(filedValue);

} catch

(NoSuchFieldException noSuchFieldException) {

paramsClass =

paramsClass.getSuperclass();

if

(paramsClass.getName().equals(Object.class.getName())) {

log.error("没有

找到类要锁住的属性{}", filedValue);

throw new

RuntimeException("没有找到类要锁住的属性");

}

}

}

field.setAccessible(true);

okey = field.get(params);

}

if (okey != null) {

if

(DxGlobalLockUtils.isWrapClass(okey.getClass())) {

key.append(okey);

} else {

key.append(JSON.toJSONString(okey));

}

}

} else {

// 获取反射注解

Field[] fields =

paramsClass.getDeclaredFields();

for (Field field : fields) {

Annotation annotation2 =

field.getAnnotation(DxGlobalLockKey.class);

if (annotation2 != null) {

field.setAccessible(true);

Object o =

field.get(params);

if (o != null) {

if

(DxGlobalLockUtils.isWrapClass(o.getClass())) {

key.append(o);

} else {

key.append(JSON.toJSONString(o));

}

}

}

}

}

}

}

}

}

}

}

// 构建最终的锁名称

if (StrUtil.isEmpty(name)) {

name = targetClass.getName() +

joinPoint.getSignature().getName();

}

String newKey = key.toString();

if (keyCount > 1) {

// 多参数排序（保证参数顺序不同生成相同key）

newKey = DxGlobalLockUtils.sort(newKey);

}

name = DxGlobalLockUtils.buildLockValue(name, newKey);

if (annotation.isWrite()) {

iswrite = true;

if

(dxDistributedReadWriteLock.getWriteLock().tryLock(name,

annotation.waitTimeoutMillisecond(), annotation.timeoutMillisecond(),

TimeUnit.MILLISECONDS)) {

result = joinPoint.proceed();

} else {

// log.warn("等待写锁超时{}---{}", name, methodName);

throw new TaskBusyException("任务处理中，请稍后重试~"

+ name);

}

} else {

if

(dxDistributedReadWriteLock.getReadLock().tryLock(name,

annotation.waitTimeoutMillisecond(), annotation.timeoutMillisecond(),

TimeUnit.MILLISECONDS)) {

result = joinPoint.proceed();

} else {

// log.warn("等待读锁超时{}---{}", name, methodName);

throw new TaskBusyException("任务处理中，请稍后重试~"

+ name);

}

}

四、RedisLock类中的tryLock（）

} catch (Exception e) {

if (!(e instanceof TaskBusyException)) {

log.error(e.getMessage(), e);

}

throw e;

} finally {

if (name != null) {

if (iswrite) {

if

(dxDistributedReadWriteLock.getWriteLock().isLocked(name) &&

dxDistributedReadWriteLock.getWriteLock().isHeldByCurrentThread(name))

{

dxDistributedReadWriteLock.getWriteLock().unlock(name);

}

} else {

if

(dxDistributedReadWriteLock.getReadLock().isLocked(name) &&

dxDistributedReadWriteLock.getReadLock().isHeldByCurrentThread(name)) {

dxDistributedReadWriteLock.getReadLock().unlock(name);

}

}

}

@Override

public boolean tryLock(long waitTime, long leaseTime, TimeUnit

unit) throws InterruptedException {

long time = unit.toMillis(waitTime);

long current = System.currentTimeMillis();

long threadId = Thread.currentThread().getId();

// 1.尝试获取锁

Long ttl = tryAcquire(leaseTime, unit, threadId);

// lock acquired

if (ttl == null) {

return true;

}

// 申请锁的耗时如果大于等于最大等待时间，则申请锁失败.

time -= System.currentTimeMillis() - current;

if (time <= 0) {

acquireFailed(threadId);

return false;

}