为什么须要分布式锁
- 用户下单
锁住 uid，避免反复下单。
- 库存扣减
锁住库存，避免超卖。
- 余额扣减
锁住账户，避免并发操作。
分布式系统中共享同一个资源时往往须要分布式锁来保障变更资源一致性。
分布式锁须要具备个性
- 排他性
锁的根本个性，并且只能被第一个持有者持有。
- 防死锁
高并发场景下临界资源一旦产生死锁十分难以排查，通常能够通过设置超时工夫到期主动开释锁来躲避。
- 可重入
锁持有者反对可重入，避免锁持有者再次重入时锁被超时开释。
- 高性能高可用
锁是代码运行的要害前置节点，一旦不可用则业务间接就报故障了。高并发场景下，高性能高可用是根本要求。
实现 Redis 锁应先把握哪些知识点
- set 命令
SET key value [EX seconds] [PX milliseconds] [NX|XX]
EXsecond ：设置键的过期工夫为 second 秒。 SET key value EX second 成果等同于 SETEX key second value 。PXmillisecond ：设置键的过期工夫为 millisecond 毫秒。 SET key value PX millisecond 成果等同于 PSETEX key millisecond value 。NX：只在键不存在时，才对键进行设置操作。 SET key value NX 成果等同于 SETNX key value 。XX：只在键曾经存在时，才对键进行设置操作。
- Redis.lua 脚本
应用 redis lua 脚本能将一系列命令操作封装成 pipline 实现整体操作的原子性。
go-zero 分布式锁 RedisLock 源码剖析
core/stores/redis/redislock.go
- 加锁流程
-- KEYS[1]: 锁key-- ARGV[1]: 锁value,随机字符串-- ARGV[2]: 过期工夫-- 判断锁key持有的value是否等于传入的value-- 如果相等阐明是再次获取锁并更新获取工夫，避免重入时过期-- 这里阐明是“可重入锁”if redis.call("GET", KEYS[1]) == ARGV[1] then -- 设置 redis.call("SET", KEYS[1], ARGV[1], "PX", ARGV[2]) return "OK"else -- 锁key.value不等于传入的value则阐明是第一次获取锁 -- SET key value NX PX timeout : 当key不存在时才设置key的值 -- 设置胜利会主动返回“OK”，设置失败返回“NULL Bulk Reply” -- 为什么这里要加“NX”呢，因为须要避免把他人的锁给笼罩了 return redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2])end- 解锁流程
-- 开释锁-- 不能够开释他人的锁if redis.call("GET", KEYS[1]) == ARGV[1] then -- 执行胜利返回“1” return redis.call("DEL", KEYS[1])else return 0end- 源码解析
package redisimport ( "math/rand" "strconv" "sync/atomic" "time" red "github.com/go-redis/redis" "github.com/tal-tech/go-zero/core/logx")const ( letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" lockCommand = `if redis.call("GET", KEYS[1]) == ARGV[1] then redis.call("SET", KEYS[1], ARGV[1], "PX", ARGV[2]) return "OK"else return redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2])end` delCommand = `if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1])else return 0end` randomLen = 16 // 默认超时工夫，避免死锁 tolerance = 500 // milliseconds millisPerSecond = 1000)// A RedisLock is a redis lock.type RedisLock struct { // redis客户端 store *Redis // 超时工夫 seconds uint32 // 锁key key string // 锁value，避免锁被他人获取到 id string}func init() { rand.Seed(time.Now().UnixNano())}// NewRedisLock returns a RedisLock.func NewRedisLock(store *Redis, key string) *RedisLock { return &RedisLock{ store: store, key: key, // 获取锁时，锁的值通过随机字符串生成 // 实际上go-zero提供更加高效的随机字符串生成形式 // 见core/stringx/random.go：Randn id: randomStr(randomLen), }}// Acquire acquires the lock.// 加锁func (rl *RedisLock) Acquire() (bool, error) { // 获取过期工夫 seconds := atomic.LoadUint32(&rl.seconds) // 默认锁过期工夫为500ms，避免死锁 resp, err := rl.store.Eval(lockCommand, []string{rl.key}, []string{ rl.id, strconv.Itoa(int(seconds)*millisPerSecond + tolerance), }) if err == red.Nil { return false, nil } else if err != nil { logx.Errorf("Error on acquiring lock for %s, %s", rl.key, err.Error()) return false, err } else if resp == nil { return false, nil } reply, ok := resp.(string) if ok && reply == "OK" { return true, nil } logx.Errorf("Unknown reply when acquiring lock for %s: %v", rl.key, resp) return false, nil}// Release releases the lock.// 开释锁func (rl *RedisLock) Release() (bool, error) { resp, err := rl.store.Eval(delCommand, []string{rl.key}, []string{rl.id}) if err != nil { return false, err } reply, ok := resp.(int64) if !ok { return false, nil } return reply == 1, nil}// SetExpire sets the expire.// 须要留神的是须要在Acquire()之前调用// 不然默认为500ms主动开释func (rl *RedisLock) SetExpire(seconds int) { atomic.StoreUint32(&rl.seconds, uint32(seconds))}func randomStr(n int) string { b := make([]byte, n) for i := range b { b[i] = letters[rand.Intn(len(letters))] } return string(b)}对于分布式锁还有哪些实现计划
- etcd
- redis redlock
我的项目地址
https://github.com/zeromicro/go-zero
欢送应用 go-zero 并 star 反对咱们！
微信交换群
关注『微服务实际』公众号并点击 交换群 获取社区群二维码。