Redis 分布式锁

一、分布式锁

1. 介绍

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。

在之前的单进程环境下，我们保证数据的安全性是通过使用 sychronized 同步锁来实现的，而 synchronized 是 JVM 级别的，也就是说它只对单进程起作用。
而在分布式或集群模式下，我们的系统的有多个进程的，synchronized 只能锁住各自进程内的线程，它不能保证多进程环境下的安全性，所以这个时候我们需要有另外一个锁，可以作用于多个进程之间并且互斥，这就是我们要说的分布式锁。

Redis 本身对多进程是可见的，可以利用 setnx 命令实现多进程之间的互斥，setex 设置过期时间到期自动释放，避免死锁提高安全性，利用 Redis 集群保证高可用和高并发的特性！

2. 基本操作

① 获取锁

//确保只能有一个线程获取到锁
setnx lock thread1

② 设置过期时间

//添加过期时间，到期自动删除，避免死锁
expire lock 20

为了避免在获取锁和设置过期时间之间服务器出现什么问题，这两个步骤可以合并到一起保证原子性。

set lock thread1 ex 20 nx

成功返回 ok，失败返回 nil！

③ 释放锁

//删除即可
del lock

3. 初级分布式锁

@Component
public class ILockImpl implements ILock {//name是锁的名称，不同的业务需要有不同的锁private String name;private StringRedisTemplate stringRedisTemplate;public void setName(String name) {this.name = name;}public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "LOCK:";@Overridepublic boolean tryLock(long timeoutSec) {//获取线程标识long threadId = Thread.currentThread().getId();//获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(name + KEY_PREFIX, threadId + "", timeoutSec, TimeUnit.SECONDS);//返回结果会自动拆箱，有空指针的风险，可以使用Boolean.TRUE.equals方法做一个比较return Boolean.TRUE.equals(success);}@Overridepublic void unLock() {//释放锁stringRedisTemplate.delete(KEY_PREFIX + name);}
}

4. 误释放别人的锁

这是什么情况呢？比如说现在有一个线程 A 正占用着锁，但是 A 出现了点阻塞，以至于 A 的业务还没有执行完，锁就过期被释放了。释放了之后，必然会有其它的线程乘虚而入，比如 B 抢到了这个分布式锁，开始执行自己的业务，此刻 A 线程突然醒过来了，A 执行完就去 delete 释放锁，其实它自己的锁早就过期了，此时 delete 是把 B 的锁给释放了，然而 B 业务还没执行完。既然锁又空了，那么这个时候肯定又会有另一个线程乘虚而入，我们叫它 C 线程，C 线程拿到锁也开始执行自己的业务，这不就是 B 和 C 一起执行吗？多个线程可以同时访问公共资源，这又如何保证安全问题。

针对这个问题，我们可以在获取锁的时候顺便存入线程的标识，不同进程之间的线程 ID 可能重复，可以用 UUID拼接，在释放锁时要做一个判断，判断一下这个锁标识是不是自己的，是就释放，不是就什么都不做，避免误删别人的锁。

@Component
public class ILockImpl implements ILock {//name是锁的名称，不同的业务需要有不同的锁private String name;private StringRedisTemplate stringRedisTemplate;public void setName(String name) {this.name = name;}public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "LOCK:";private static final String ID_PREFIX = UUID.randomUUID().toString() + "-";@Overridepublic boolean tryLock(long timeoutSec) {//获取线程标识String threadId = ID_PREFIX + Thread.currentThread().getId();//获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(name + KEY_PREFIX, threadId, timeoutSec, TimeUnit.SECONDS);//返回结果会自动拆箱，有空指针的风险，可以使用Boolean.TRUE.equals方法做一个比较return Boolean.TRUE.equals(success);}@Overridepublic void unLock() {//获取线程标识String threadId = ID_PREFIX + Thread.currentThread().getId();//获取锁中的标识String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);if (threadId.equals(id)) {//释放锁stringRedisTemplate.delete(KEY_PREFIX + name);}}
}

二、Lua 脚本

特殊情况，判断完锁标识是自己的之后，正准备要释放锁，这时候突然就阻塞了，就是在这个离谱的地方阻塞，然后超时，释放锁，这让下一个线程乘虚而入，很巧它也在这个特殊位置上阻塞，上面误释放别人锁的问题再次发生了。虽然这是一种特殊中的特殊情况，但是也得考虑在内。

让判断锁标识和释放锁的动作同时发生，来保证原子性！

基于 Redis 的分布式锁实现思路：
① 利用 set nx ex 获取锁，并设置过期时间，保存线程标识；
② 锁释放前，先判断锁中标识是否与当前标识一致，如果一致则释放锁，不一致则什么都不做。

1. Lua 基本用法

Redis 提供了 Lua 脚本功能，在一个脚本中编写多条 Redis 命令，确保多条命令执行时的原子性。

下面是 Redis 提供的调用函数，语法如下：

执行脚本：

//0代表脚本需要的KEY类型的参数个数为0EVAL "return redis.call('set', 'name', 'zhanlaoshi')" 0

如果脚本中的 key、value 不想写死，可以作为参数传递。key 类型参数会放入 KEYS 数组，其它参数会放入 ARGV 数组，在脚本中可以从 KEYS 和 ARGV 数组中获取这些参数。

EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name zxe

注意 Lua 语言中的数组下标是从 1 开始的！

2. Lua 改造分布式锁

@Component
public class ILockImpl implements ILock {//name是锁的名称，不同的业务需要有不同的锁private String name;private StringRedisTemplate stringRedisTemplate;public void setName(String name) {this.name = name;}public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "LOCK:";private static final String ID_PREFIX = UUID.randomUUID().toString() + "-";private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;static {UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);}@Overridepublic boolean tryLock(long timeoutSec) {//获取线程标识String threadId = ID_PREFIX + Thread.currentThread().getId();//获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(name + KEY_PREFIX, threadId, timeoutSec, TimeUnit.SECONDS);//返回结果会自动拆箱，有空指针的风险，可以使用Boolean.TRUE.equals方法做一个比较return Boolean.TRUE.equals(success);}@Overridepublic void unLock() {//调用lua脚本stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name),ID_PREFIX + Thread.currentThread().getId());}
}

使用 Lua 脚本改进后，判断和释放是在脚本中执行的，能够保证原子性！