基于Redis实现分布式锁

目录

一.分布式锁介绍

1.为什么需要分布式锁

 2.分布式锁应该具备那些条件

基于Redis实现分布式锁

1.为什么要给锁设置一个过期时间

2.如何实现锁的优雅续期

一.分布式锁介绍

1.为什么需要分布式锁

在多线程的环境下,如果多个线程同时访问共享资源(例如商品库存,外卖订单), 会发生数据竞争,可能会导致出现脏数据或者系统问题,威胁到程序的正常运行.

举个例子，假设现在有 100 个用户参与某个限时秒杀活动，每位用户限购 1 件商品，且商品的数量只有 3 个。如果不对共享资源进行互斥访问，就可能出现以下情况：

• 线程 1、2、3 等多个线程同时进入抢购方法，每一个线程对应一个用户。
• 线程 1 查询用户已经抢购的数量，发现当前用户尚未抢购且商品库存还有 1 个，因此认为可以继续执行抢购流程。
• 线程 2 也执行查询用户已经抢购的数量，发现当前用户尚未抢购且商品库存还有 1 个，因此认为可以继续执行抢购流程。
• 线程 1 继续执行，将库存数量减少 1 个，然后返回成功。
• 线程 2 继续执行，将库存数量减少 1 个，然后返回成功。
• 此时就发生了超卖问题，导致商品被多卖了一份

为了保证共享资源被安全的访问,我们需要使用互斥操作对共享资源进行保护,即同一时刻只允许一个线程访问共享资源,其他线程需要等待当前线程释放后才能访问,这样可以避免数据竞争和脏数据问题,保证程序的正确性和稳定性.

如何实现共享资源的互斥访问呢?锁是一个比较通用的方案,更准确点来说是一个悲观锁.

悲观锁总是假设最坏的情况,认为共享资源每次被访问的时候就会出现问题(比如数据被修改),所以每次在获取资源操作的时候都会上锁,这样其他线程想拿到这个资源就会阻塞直到锁被上一个持有者释放,也就是说,共享资源每次只给一个线程使用,其他线程,用完后再把资源转让给其他线程.

对于多线程来说,在Java中,我们通常使用ReentranLock类,synchronized关键字这类JDK自带的本地锁来控制一个JVM进程内的多个线程对本地共享资源的访问. 

下面是对本地锁画的一张图

从图中可以看出，这些线程访问共享资源是互斥的，同一时刻只有一个线程可以获取到本地锁访问共享资源。

分布式系统下，不同的服务/客户端通常运行在独立的 JVM 进程上。如果多个 JVM 进程共享同一份资源的话，使用本地锁就没办法实现资源的互斥访问了。于是，分布式锁 就诞生了。

举个例子：系统的订单服务一共部署了 3 份，都对外提供服务。用户下订单之前需要检查库存，为了防止超卖，这里需要加锁以实现对检查库存操作的同步访问。由于订单服务位于不同的 JVM 进程中，本地锁在这种情况下就没办法正常工作了。我们需要用到分布式锁，这样的话，即使多个线程不在同一个 JVM 进程中也能获取到同一把锁，进而实现共享资源的互斥访问。

 下面是分布式锁画的一张示意图.

 2.分布式锁应该具备那些条件

一个最基本的分布式锁需要满足:

互斥:任意一个时刻,锁只能被一个线程持有

高可用:锁服务是高可用的,当一个锁服务出现问题能够自动切换到另一个锁服务.并且,即使客户端的释放锁的代码逻辑出现问题,锁最终一定还是会被释放,不会影响其他线程对共享资源的访问.这一般是通过超时机制实现的.

可重入:一个节点获取了锁之后,还可以再次获取锁

高性能:获取和释放锁的操作应该快速完成,并且不应该对整个系统的性能造成过大影响.

非阻塞:如果获取不到锁.不能无限期等待,避免对系统正常运行造成影响.

基于Redis实现分布式锁

无论是本地锁还是分布式锁,核心都在于互斥

 在 Redis 中， SETNX 命令是可以帮助我们实现互斥。SETNX 即 SET if Not eXists (对应 Java 中的 setIfAbsent 方法)，如果 key 不存在的话，才会设置 key 的值。如果 key 已经存在， SETNX 啥也不做。

SETNX lockKey uniqueValue
(integer) 1
SETNX lockKey uniqueValue
(integer) 0

释放锁的话,直接通过DEL命令删除对应的key即可.

DEL lockKey
(integer) 1

 为了防止误删除到其他的锁,这里建议使用Lua脚本通过key对应的value(唯一值来判断).

选用Lua脚本是为了保证解锁操作的原子性,因为Redis在执行Lua脚本时,可以以原子性的方式执行,从而保证了锁释放操作的原子性.

// 释放锁时，先比较锁对应的 value 值是否相等，避免锁的误释放
if redis.call("get",KEYS[1]) == ARGV[1] thenreturn redis.call("del",KEYS[1])
elsereturn 0
end

 

 当前实现这种分布式锁还有一些问题比如释放锁的逻辑突然挂掉,可能导致锁无法释放,进而造成共享资源无法再被其他线程/进程访问.

1.为什么要给锁设置一个过期时间

为了所释放失败,所以在设置key的时候设置一个过期时间

127.0.0.1:6379> SET lockKey uniqueValue EX 3 NX
OK

 注:一定要保证设置指定key和过期时间是一个原子操作,不然会出现锁无法被释放的问题

假如先设置值,在设置过期时间,在两次操作之间可能出现服务中断或者是故障,这将导致锁无法释放.

 

这样确实可以解决问题，不过，这种解决办法同样存在漏洞：如果操作共享资源的时间大于过期时间，就会出现锁提前过期的问题，进而导致分布式锁直接失效。如果锁的超时时间设置过长，又会影响到性能。

你或许在想：如果操作共享资源的操作还未完成，锁过期时间能够自己续期就好了！

2.如何实现锁的优雅续期

对于 Java 开发的小伙伴来说，已经有了现成的解决方案：Redisson 。其他语言的解决方案，可以在 Redis 官方文档中找到，地址：Distributed Locks with Redis | Docs 。

Redisson 中的分布式锁自带自动续期机制，使用起来非常简单，原理也比较简单，其提供了一个专门用来监控和续期锁的 Watch Dog（ 看门狗），如果操作共享资源的线程还未执行完成的话，Watch Dog 会不断地延长锁的过期时间，进而保证锁不会因为超时而被释放。

 看门狗名字的由来于 getLockWatchdogTimeout() 方法，这个方法返回的是看门狗给锁续期的过期时间，默认为 30 秒

//默认 30秒，支持修改
private long lockWatchdogTimeout = 30 * 1000;public Config setLockWatchdogTimeout(long lockWatchdogTimeout) {this.lockWatchdogTimeout = lockWatchdogTimeout;return this;
}
public long getLockWatchdogTimeout() {return lockWatchdogTimeout;
}

private void renewExpiration() {//......Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {//......// 异步续期，基于 Lua 脚本CompletionStage<Boolean> future = renewExpirationAsync(threadId);future.whenComplete((res, e) -> {if (e != null) {// 无法续期log.error("Can't update lock " + getRawName() + " expiration", e);EXPIRATION_RENEWAL_MAP.remove(getEntryName());return;}if (res) {// 递归调用实现续期renewExpiration();} else {// 取消续期cancelExpirationRenewal(null);}});}// 延迟 internalLockLeaseTime/3（默认 10s，也就是 30/3） 再调用}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);}

 默认情况下，每过 10 秒，看门狗就会执行续期操作，将锁的超时时间设置为 30 秒。看门狗续期前也会先判断是否需要执行续期操作，需要才会执行续期，否则取消续期操作。

Watch Dog 通过调用 renewExpirationAsync() 方法实现锁的异步续期：

protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,// 判断是否为持锁线程，如果是就执行续期操作，就锁的过期时间设置为 30s（默认）"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return 1; " +"end; " +"return 0;",Collections.singletonList(getRawName()),internalLockLeaseTime, getLockName(threadId));
}

 可以看出， renewExpirationAsync 方法其实是调用 Lua 脚本实现的续期，这样做主要是为了保证续期操作的原子性

我这里以 Redisson 的分布式可重入锁 RLock 为例来说明如何使用 Redisson 实现分布式锁：

// 1.获取指定的分布式锁对象
RLock lock = redisson.getLock("lock");
// 2.拿锁且不设置锁超时时间，具备 Watch Dog 自动续期机制
lock.lock();
// 3.执行业务
...
// 4.释放锁
lock.unlock();

只有未指定锁超时时间，才会使用到 Watch Dog 自动续期机制。

// 手动给锁设置过期时间，不具备 Watch Dog 自动续期机制
lock.lock(10, TimeUnit.SECONDS);

如果使用 Redis 来实现分布式锁的话，还是比较推荐直接基于 Redisson 来做的。