Redis分布式锁的实现方案
redis 分布式锁SetNx + Expire + Del
Redis版本如果低于2.6.12，SetNX和Expire要分开使用：
ok, err:= redis.SetNx(key, 1)
// 出错，加锁失败
if err != nil {
return
}
// 锁被其他任务占用
if !ok {
return
}
// 设置过期时间
_, err = redis.Expire(key, 60 * time.Second)
if err != nil {
return
}
// 加锁成功，处理任务
// 处理业务完毕，释放锁
_, err = redis.Del(key)
if err != nil {
return
}
此方式故障点：
- redis.Expire执行失败或者未执行
- ****redis.Del**执行失败或者未执行
- ****redis.Expire**设置的key（锁）过期了，但是任务还没执行完成
故障点1所导致的问题： redis.Expire执行失败或者未执行，锁永远不会过期。如果此时任务处理过程中panic或者其他原因导致进程终止，也就是redis.Del没有执行，此时由于锁永远不会过期，此任务后续将无法继续执行，因为所有任务都无法获取锁了。
故障点2导致的问题： redis.Del执行失败或者未执行，由于之前设置了锁的过期时间，所以锁过期后，其他任务可以抢占锁。此时任务没有较大的影响。
故障点3导致的问题： 锁过期了，但是任务还没执行完毕。此时其他任务将会抢占锁，将会出现任务执行了2次的情况。两个相同的任务同时执行，也会导致其他问题。例如，第一个任务执行完毕，删除了锁，会导致第二个任务的锁也被删除，此时任务将不可控。
使用 Set + Del ，解决 redis.Expire 未执行问题
Redis版本如果高于2.6.12，可以使用Set命令直接设置过期时间，相当于SetNx + Expire：
SET key value [NX | XX] [GET] [EX seconds | PX milliseconds | EXAT unix-time-seconds | PXAT unix-time-milliseconds | KEEPTTL]
程序可以改写为：
ok, err:= redis.Set(key, 1, "NX", "EX", 60 * time.Second)
// 出错，加锁失败
if err != nil {
return
}
// 锁被其他任务占用
if !ok {
return
}
// 加锁成功，处理任务
// 处理业务完毕，释放锁
_, err = redis.Del(key)
if err != nil {
return
}
改造后，此时能解决redis.Expire执行失败问题。
解决 锁过期 导致 删除了别的任务加的锁 的问题
此问题是锁过期带来的额外问题，其实只要解决了锁过期问题，就不会出现此问题，但是我们还是要简单聊下此问题怎么解决。
解决此问题的入手点很简单，只要保证谁加的锁，谁释放即可。
我们可以在加锁的时候，将Redis的value设置为一个随机数，删除锁时先获取锁的值，如果值等于随机数，那么就可以执行删除锁操作。
改造代码如下：
value := rand.Int()
ok, err:= redis.Set(key, value, "NX", "EX", 60 * time.Second)
// 出错，加锁失败
if err != nil {
return
}
// 锁被其他任务占用
if !ok {
return
}
// 加锁成功，处理任务
// 处理业务完毕，释放锁
{
// 先获取锁的值
v, err := redis.Get(key).Result()
if err != nil {
return
}
// 如果值和设置的时候的value相同，则删除锁
if v == value {
_, err = redis.Del(key)
if err != nil {
return
}
}
}
注意，如上代码示例，只使用业务代码层面来判断删除锁逻辑，如果redis.Get之后，刚好锁过期了，此时锁被别的任务抢占走，此时再执行了redis.Del错误的删除了锁。
解决方案是使用LUA脚本，将redis.Get和redis.Del放到LUA脚本中：
value := rand.Int()
ok, err:= redis.Set(key, value, "NX", "EX", 60 * time.Second)
// 出错，加锁失败
if err != nil {
return
}
// 锁被其他任务占用
if !ok {
return
}
// 加锁成功，处理任务
// 处理业务完毕，释放锁
var delScript = redis.NewScript(1, `
local val = redis.call("GET", KEYS[1])
if val == ARGV[1] then
return redis.call("DEL", KEYS[1])
elseif val == false then
return -1
else
return 0
end
`)
_, err = delScript.Run(ctx, rdb, []string{key}, value).Result()
if err != nil {
// TODO error log
}
使用延时器解决 锁过期 问题
任务没执行完，但是锁过期，会导致锁被其他任务抢占引发一系列故障。
我们可以在加锁成功后，启动一个协程，每30s续约一次锁的过期时间。
程序可以改写为：
rdb := redis.NewClient(&redis.Options{})
ctx := context.Background()
key := "redis_key_lock"
// lock
value := rand.Int()
ok, err:= redis.Set(key, value, "NX", "EX", 60 * time.Second)
if err != nil {
// TODO error log
return
}
if !ok {
return
}
renewLock := func(ctx context.Context) {
ticker := time.NewTicker(30 * time.Second)
for {
select {
case <-ctx.Done():
return
case <-ticker.C: // 续约锁
_, err := rdb.Expire(ctx, key, 60*time.Second).Result()
if err != nil {
// TODO error log
}
continue
}
}
}
renewLockCtx, cancel := context.WithCancel(ctx)
go renewLock(renewLockCtx)
// do something
// unlock
cancel()
delScript := redis.NewScript(`
local val = redis.call("GET", KEYS[1])
if val == ARGV[1] then
return redis.call("DEL", KEYS[1])
elseif val == false then
return -1
else
return 0
end
`)
_, err = delScript.Run(ctx, rdb, []string{key}, value).Result()
if err != nil {
// TODO error log
}
此时已经达到了比较完善的Redis分布式锁逻辑，但是此时引入了新的问题：
- 续约函数中的redis.Expire如果一直不执行（比如机器网络故障），或者续约协程挂掉了，任然可能会导致锁提前释放。
此问题实际是Redis单点故障引入的问题，解决办法是使用Redis多实例集群分布式锁。目前比较常用的方案是RedLock分布式锁算法。
Redis主备模式故障分析
以上分析都是以Redis单点模式进行的。如果在Master-Slave模式下的Redis部署，会出现额外问题：
- 任务获取锁后，Master执行redis.Set后，主节点命令尚未同步到Slave节点，此时Master节点挂掉，Slave此时切换成为Master，此时Redis不存在锁。其他任务就可以获得锁。
此问题的解决方案是：不要使用Master-Slave架构的Redis实现分布式锁😂。
RedLock分布式锁算法
为了解决Redis的单点故障，以及Master-Slave模式出现的问题，Salvatore设计了一种基于多个Redis节点的分布式锁算法。
算法的核心思想是：采用N个Redis节点，其中N%2==1，每个节点不设置主备。如果任务要上锁，则依次对这N个Redis节点进行加锁，加锁成功的节点个数为L，若L >= (N/2+1) 则认为加锁成功。即若大多数节点加锁成功，则认为总体加锁成功。
RedLock的具体算法，在此不再具体分析，可以参见Redis官方文档了解算法细节和具体代码实现。需要注意的是，RedLock算法不是一个完美的算法，比如没有考虑机器时钟同步问题。具体可以参见文章：