Redis分布式锁

一、什么是分布式锁

为了防止分布式系统中的多个进程之间相互干扰，我们需要一种分布式协调技术来对这些进程进行调度。而这个分布式协调技术的核心就是来实现这个分布式锁，控制分布式系统有序的去对共享资源进行操作，通过互斥来保持一致性。

二、设计分布式锁需要考虑的问题

• 在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行
• 高可用、高性能的获取锁与释放锁
• 具备可重入特性（可理解为重新进入，由多于一个任务并发使用，而不必担心数据错误）
• 具备锁失效机制，防止死锁
• 具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败

三、Redis分布式锁

1. 加锁

最简单的方法是使用 setnx 命令。key 是锁的唯一标识，按业务来决定命名。

> setnx lock:codehole true
复制代码

2. 解锁

有加锁就得有解锁。当得到锁的线程执行完任务，需要释放锁，以便其他线程可以进入。释放锁的最简单方式是执行 del 指令

> del lock:codehole
复制代码

3. 设置超时时间

锁超时是什么意思呢？如果一个得到锁的线程在执行任务的过程中挂掉，来不及显式地释放锁，这块资源将会永远被锁住（死锁），别的线程再也别想进来。所以，setnx 的 key 必须设置一个超时时间，以保证即使没有被显式释放，这把锁也要在一定时间后自动释放。setnx 不支持超时参数，所以需要额外的指令

> setnx lock : codehole true
OK
> expire lock:codehole 5
do something critical
> del lock : codehole
(integer) 1
复制代码

4. 存在问题 & 解决办法

• setnx 和 expire 的非原子性

这里有个问题，加锁了之后如果机器宕机那么这个锁就不会得到释放所以会加入过期时间，加入过期时间需要和setNx同一个原子操作，在Redis2.8之前我们需要使用Lua脚本达到我们的目的，但是redis2.8之后redis支持nx和ex操作是同一原子操作。

> set lock:codehole true ex 5 nx
OK
do something eritical ...
> del lock:codehole
复制代码

• 超时问题

假如某线程成功得到了锁，并且设置的超时时间是 30 秒。
如果某些原因导致线程 A 执行的很慢很慢，过了 30 秒都没执行完，这时候锁过期自动释放，线程 B 得到了锁。
随后，线程 A 执行完了任务，线程 A 接着执行 del 指令来释放锁。但这时候线程 B 还没执行完，线程A实际上 删除的是线程 B 加的锁。

tag = random. next int()
if redis.set(key, tag, nx=True, ex=S}:
do_ something ()
redis.delifequals(key, tag)  ＃假想的 delifequals 指令
复制代码

这时候可以加一个随机数，释放时先匹配随机数是否一致，在删除Key。是匹配 value 和删除 key 不是个原子操作，
这就需要使用 Lua 脚本保证多个指令的原子性。

虽然我们避免了线程 A 误删掉 key 的情况，但是同一时间有 A，B 两个线程在访问代码块，仍然是不完美的。怎么办呢？我们可以让获得锁的线程开启一个守护线程，用来给快要过期的锁“续航”。

四、 Redisson分布式锁

一般使用都是Redisson分布式锁

• 可重入锁
• 支持单点模式、主从模式、哨兵模式、集群模式
• 加锁和解锁的执行层面采用Lua脚本，保证原子性

加锁Lua脚本

-- 若锁不存在：则新增锁，并设置锁重入计数为1、设置锁过期时间
if (redis.call('exists', KEYS[1]) == 0) then
    redis.call('hset', KEYS[1], ARGV[2], 1);
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
end;
 
-- 若锁存在，且唯一标识也匹配：则表明当前加锁请求为锁重入请求，故锁重入计数+1，并再次设置锁过期时间
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]);
复制代码

解锁Lua脚本

-- 若锁不存在：则直接广播解锁消息，并返回1
if (redis.call('exists', KEYS[1]) == 0) then
    redis.call('publish', KEYS[2], ARGV[1]);
    return 1; 
end;
 
-- 若锁存在，但唯一标识不匹配：则表明锁被其他线程占用，当前线程不允许解锁其他线程持有的锁
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
    return nil;
end; 
 
-- 若锁存在，且唯一标识匹配：则先将锁重入计数减1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); 
if (counter > 0) then 
    -- 锁重入计数减1后还大于0：表明当前线程持有的锁还有重入，不能进行锁删除操作，但可以友好地帮忙设置下过期时期
    redis.call('pexpire', KEYS[1], ARGV[2]); 
    return 0; 
else 
    -- 锁重入计数已为0：间接表明锁已释放了。直接删除掉锁，并广播解锁消息，去唤醒那些争抢过锁但还处于阻塞中的线程
    redis.call('del', KEYS[1]); 
    redis.call('publish', KEYS[2], ARGV[1]); 
    return 1;
end;
 
return nil;
复制代码

问题：缺陷在哨兵模式或者主从模式下，如果 master 实例宕机的时候，可能导致多个客户端同时完成加锁。解决方案，使用Redlock算法, 参考redis.cn/topics/dist…