用 Go + Redis 實現(xiàn)分布式鎖

為什么需要分布式鎖

1. 用戶下單

鎖住 uid，防止重復(fù)下單。

2. 庫存扣減

鎖住庫存，防止超賣。

3. 余額扣減

鎖住賬戶，防止并發(fā)操作。分布式系統(tǒng)中共享同一個資源時往往需要分布式鎖來保證變更資源一致性。

分布式鎖需要具備特性

1. 排他性

鎖的基本特性，并且只能被第一個持有者持有。

2. 防死鎖

高并發(fā)場景下臨界資源一旦發(fā)生死鎖非常難以排查，通常可以通過設(shè)置超時時間到期自動釋放鎖來規(guī)避。

3. 可重入

鎖持有者支持可重入，防止鎖持有者再次重入時鎖被超時釋放。

4. 高性能高可用

鎖是代碼運行的關(guān)鍵前置節(jié)點，一旦不可用則業(yè)務(wù)直接就報故障了。高并發(fā)場景下，高性能高可用是基本要求。

實現(xiàn) Redis 鎖應(yīng)先掌握哪些知識點

1. set 命令

SET key value [EX seconds] [PX milliseconds] [NX|XX]

• EXsecond ：設(shè)置鍵的過期時間為 second 秒。SET key value EX second 效果等同于 SETEX key second value 。
• PXmillisecond ：設(shè)置鍵的過期時間為 millisecond 毫秒。SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。
• NX：只在鍵不存在時，才對鍵進行設(shè)置操作。SET key value NX 效果等同于 SETNX key value 。
• XX：只在鍵已經(jīng)存在時，才對鍵進行設(shè)置操作。

2. Redis.lua 腳本

使用 redis lua 腳本能將一系列命令操作封裝成 pipline 實現(xiàn)整體操作的原子性。

go-zero 分布式鎖 RedisLock 源碼分析

core/stores/redis/redislock.go

1. 加鎖流程

--?KEYS[1]:?鎖key
--?ARGV[1]:?鎖value,隨機字符串
--?ARGV[2]:?過期時間
--?判斷鎖key持有的value是否等于傳入的value
--?如果相等說明是再次獲取鎖并更新獲取時間，防止重入時過期
--?這里說明是“可重入鎖”
if?redis.call("GET",?KEYS[1])?==?ARGV[1]?then
????--?設(shè)置
????redis.call("SET",?KEYS[1],?ARGV[1],?"PX",?ARGV[2])
????return?"OK"

else
????--?鎖key.value不等于傳入的value則說明是第一次獲取鎖
????--?SET?key?value?NX?PX?timeout?:?當(dāng)key不存在時才設(shè)置key的值
????--?設(shè)置成功會自動返回“OK”，設(shè)置失敗返回“NULL?Bulk?Reply”
????--?為什么這里要加“NX”呢，因為需要防止把別人的鎖給覆蓋了
????return?redis.call("SET",?KEYS[1],?ARGV[1],?"NX",?"PX",?ARGV[2])
end

2. 解鎖流程

--?釋放鎖
--?不可以釋放別人的鎖
if?redis.call("GET",?KEYS[1])?==?ARGV[1]?then
????--?執(zhí)行成功返回“1”
????return?redis.call("DEL",?KEYS[1])
else
????return?0
end

3. 源碼解析

package?redis

import?(
????"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?0
end`
????randomLen?=?16
????//?默認(rèn)超時時間，防止死鎖
????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)
????//?默認(rèn)鎖過期時間為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()之前調(diào)用
//?不然默認(rèn)為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)
}

關(guān)于分布式鎖還有哪些實現(xiàn)方案

1. etcd
2. redis redlock

項目地址

https://github.com/zeromicro/go-zero

歡迎使用 go-zero 并 star 支持我們！

推薦閱讀

• Go 執(zhí)行腳本命令源碼分析