手动撸一个 Redis 分布式锁

01 问题引入其实通过 Redis 实现分布式锁经常会有面试官会问很多同学都知道用 SetNx() 去获取锁解决并发问题。SetNx() 是什么我简单解答一下。Redis SetnxSET if Not eXists 命令在指定的 key 不存在时为 key 设置指定的值。对于下面 2 种问题你知道如何解决么如果获取锁的机器挂掉如何处理当锁超时时A、B 两个线程同时获取锁可能导致锁被同时获取如何解决这个就是我们实现 Redis 分布式锁时需要重点解决的 2 个问题。02 理论知识刚才说过通过 SetNx() 去获取锁可以解决并发问题。当获取到锁处理完业务逻辑后会将锁释放。但当机器宕机或者重启时没有执行 Del() 删除锁操作会导致锁一直没有释放。所以我们还需要记录锁的超时时间判断锁是否超时。这里我们通过 GetKey() 获取锁的超时时间 A通过和当前时间比较判断锁是否超时。如果锁未超时直接返回如果锁超时重新设置锁的超时时间成功获取锁。还有其它问题么当然因为在并发场景下会存在 A、B 两个线程同时执行 SetNx()导致两个线程同时获取到锁。那如何解决呢将 SetNx() 用 GetSet() 替换。GetSet() 是什么我简单解答一下。Redis Getset 命令用于设置指定 key 的值并返回 key 的旧值。这里不太好理解我举个例子。假如 A、B 两个线程A 先执行B 后执行对于线程 A 和 B通过 GetKey 获取的超时时间都是 T1 100对于线程 A将超时时间 Ta 200 通过 GetSet() 设置返回 T2 100此时满足条件 “T1 T2”获取锁成功对于线程 B将超时时间 Tb 201 通过 GetSet() 设置由于锁超时时间已经被 A 重新设置所以返回 T2 200此时不满足条件 “T1 T2”获取锁失败。可能有同学会继续问之前设置的超时是 Ta 200现在变成了 Tb 201延长或缩短了锁的超时时间不会有问题么其实在现实并发场景中能走到这一步基本是“同时”进来的两者的时间差非常小可以忽略此影响。03 代码实战这里给出 Go 代码注释都写得非常详细即使你不会 Go读注释也能读懂。// 获取分布式锁需要考虑以下情况 // 1. 机器A获取到锁但是在未释放锁之前机器挂掉或者重启会导致其它机器全部hang住这时需要根据锁的超时时间判断该锁是否需要重置 // 2. 当锁超时时需要考虑两台机器同时去获取该锁需要通过GETSET方法让先执行该方法的机器获取锁另外一台继续等待。 func GetDistributeLock(key string, expireTime int64) bool { currentTime : time.Now().Unix() expires : currentTime expireTime redisAlias : jointly // 1.获取锁并将value值设置为锁的超时时间 redisRet, err : redis.SetNx(redisAlias, key, expires) if nil err utils.MustInt64(1) redisRet { // 成功获取到锁 return true } // 2.当获取到锁的机器突然重启挂掉时就需要判断锁的超时时间如果锁超时新的机器可以重新获取锁 // 2.1 获取锁的超时时间 currentLockTime, err : redis.GetKey(redisAlias, key) if err ! nil { return false } // 2.2 当锁的超时时间大于等于当前时间证明锁未超时直接返回 if utils.MustInt64(currentLockTime) currentTime { return false } // 2.3 将最新的超时时间更新到锁的value值并返回旧的锁的超时时间 oldLockTime, err : redis.GetSet(redisAlias, key, expires) if err ! nil { return false } // 2.4 当锁的两个旧的超时时间相等时证明之前没有其它机器进行GetSet操作成功获取锁 // 说明这里存在并发情况如果有A和B同时竞争A会先GetSet当B再去GetSet时oldLockTime就等于A设置的超时时间 if utils.MustString(oldLockTime) currentLockTime { return true } return false }删除锁逻辑// 删除分布式锁 // return bool true-删除成功false-删除失败 func DelDistributeLock(key string) bool { redisAlias : jointly redisRet : redis.Del(redisAlias, key) if redisRet ! nil { return false } return true }业务逻辑func DoProcess(processId int) { fmt.Printf(启动第%d个线程\n, processId) redisKey : redis_lock_key for { // 获取分布式锁 isGetLock : GetDistributeLock(redisKey, 10) if isGetLock { fmt.Printf(Get Redis Key Success, id:%d\n, processId) time.Sleep(time.Second * 3) // 删除分布式锁 DelDistributeLock(redisKey) } else { // 如果未获取到该锁为了避免redis负载过高先睡一会 time.Sleep(time.Second * 1) } } }最后起个 10 个多线程去执行这个 DoProcess()func main() { // 初始化资源 var group string group var name string name var host string // 初始化资源 host http://ip:port _, err : xrpc.NewXRpcDefault(group, name, host) if err ! nil { panic(fmt.Sprintf(initRpc when init rpc failed, err:%v, err)) } redis.SetRedis(louzai, redis_louzai) // 开启10个线程去抢Redis分布式锁 for i : 0; i 9; i { go DoProcess(i) } // 避免子线程退出主线程睡一会 time.Sleep(time.Second * 100) return }程序跑了100 s我们可以看到每次都只有 1 个线程获取到锁分别是 2、1、5、9、3执行结果如下启动第0个线程 启动第6个线程 启动第9个线程 启动第4个线程 启动第5个线程 启动第2个线程 启动第1个线程 启动第8个线程 启动第7个线程 启动第3个线程 Get Redis Key Success, id:2 Get Redis Key Success, id:2 Get Redis Key Success, id:1 Get Redis Key Success, id:5 Get Redis Key Success, id:5 Get Redis Key Success, id:5 Get Redis Key Success, id:5 Get Redis Key Success, id:5 Get Redis Key Success, id:5 Get Redis Key Success, id:5 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:9 Get Redis Key Success, id:3 Get Redis Key Success, id:3 Get Redis Key Success, id:3 Get Redis Key Success, id:3 Get Redis Key Success, id:3