Golang可重入鎖的實現(xiàn)
項目中遇到了可重入鎖的需求和實現(xiàn)，具體記錄下。
為什么需要可重入鎖
我們平時說的分布式鎖，一般指的是在不同服務器上的多個線程中，只有一個線程能搶到一個鎖，從而執(zhí)行一個任務。而我們使用鎖就是保證一個任務只能由一個線程來完成。所以我們一般是使用這樣的三段式邏輯：
Lock();
DoJob();
Unlock();
但是由于我們的系統(tǒng)都是分布式的，這個鎖一般不會只放在某個進程中，我們會借用第三方存儲，比如 Redis 來做這種分布式鎖。但是一旦借助了第三方存儲，我們就必須面對這個問題：Unlock是否能保證一定運行呢？
這個問題，我們面對的除了程序的bug之外，還有網(wǎng)絡的不穩(wěn)定，進程被殺死，服務器被down機等。我們是無法保證Unlock一定被運行的。
那么我們就一般在Lock的時候為這個鎖加一個超時時間作為兜底。
LockByExpire(duration);
DoJob();
Unlock();
這個超時時間是為了一旦出現(xiàn)異常情況導致Unlock沒有被運行，這個鎖在duration時間內也會被自動釋放。這個在redis中我們一般就是使用set ex 來進行鎖超時的設定。
但是有這個超時時間我們又遇上了問題，超時時間設置多久合適呢？當然要設置的比 DoJob 消耗的時間更長，否則的話，在任務還沒結束的時候，鎖就被釋放了，還是有可能導致并發(fā)任務的存在。
但是實際上，同樣由于網(wǎng)絡超時問題，系統(tǒng)運行狀況問題等，我們是無法準確知道DoJob這個函數(shù)要執(zhí)行多久的。那么這時候怎么辦呢？
有兩個辦法：
第一個方法，我們可以對DoJob做一個超時設置。讓DoJob最多只能執(zhí)行n秒，那么我的分布式鎖的超時時長設置比n秒長就可以了。為一個任務設置超時時間在很多語言是可以做到的。比如golang 中的 TimeoutContext。
而第二種方法，就是我們先為鎖設置一個比較小的超時時長，然后不斷續(xù)期這個鎖。對一個鎖的不斷需求，也可以理解為重新開始加鎖，這種可以不斷續(xù)期的鎖，就叫做可重入鎖。
除了主線程之外，可重入鎖必然有一個另外的線程（或者攜程）可以對這個鎖進行續(xù)期，我們叫這個額外的程序叫做watchDog（看門狗）。
具體實現(xiàn)
在Golang中，語言級別天生支持協(xié)程，所以這種可重入鎖就非常容易實現(xiàn)：
// DistributeLockRedis 基于redis的分布式可重入鎖，自動續(xù)租
type DistributeLockRedis struct {
key string // 鎖的key
expire int64 // 鎖超時時間
status bool // 上鎖成功標識
cancelFun context.CancelFunc // 用于取消自動續(xù)租攜程
redis redis.Client // redis句柄
}
// 創(chuàng)建可
func NewDistributeLockRedis(key string, expire int64) *DistributeLockRedis {
return &DistributeLockRedis{
key : key,
expire : expire,
}
}
// TryLock 上鎖
func (dl *DistributeLockRedis) TryLock() (err error) {
if err = dl.lock(); err != nil {
return err
}
ctx, cancelFun := context.WithCancel(context.Background())
dl.cancelFun = cancelFun
dl.startWatchDog(ctx) // 創(chuàng)建守護協(xié)程，自動對鎖進行續(xù)期
dl.status = true
return nil
}
// competition 競爭鎖
func (dl *DistributeLockRedis) lock() error {
if res, err := redis.String(dl.redis.Do(context.Background(), "SET", dl.key, 1, "NX", "EX", dl.expire)); err != nil {
return err
}
return nil
}
// guard 創(chuàng)建守護協(xié)程，自動續(xù)期
func (dl *DistributeLockRedis) startWatchDog(ctx context.Context) {
safeGo(func() error {
for {
select {
// Unlock通知結束
case <-ctx.Done():
return nil
default:
// 否則只要開始了，就自動重入（續(xù)租鎖）
if dl.status {
if res, err := redis.Int(dl.redis.Do(context.Background(), "EXPIRE", dl.key, dl.expire)); err != nil {
return nil
}
// 續(xù)租時間為 expire/2 秒
time.Sleep(time.Duration(dl.expire/2) * time.Second)
}
}
}
})
}
// Unlock 釋放鎖
func (dl *DistributeLockRedis) Unlock() (err error) {
// 這個重入鎖必須取消，放在第一個地方執(zhí)行
if dl.cancelFun != nil {
dl.cancelFun() // 釋放成功，取消重入鎖
}
var res int
if dl.status {
if res, err = redis.Int(dl.redis.Do(context.Background(), "Del", dl.key)); err != nil {
return fmt.Errorf("釋放鎖失敗")
}
if res == 1 {
dl.status = false
return nil
}
}
return fmt.Errorf("釋放鎖失敗")
}
這段代碼的邏輯基本上都以注釋的形式來寫了。其中主要就在startWatchDog，對鎖進行重新續(xù)期
ctx, cancelFun := context.WithCancel(context.Background())
dl.cancelFun = cancelFun
dl.startWatchDog(ctx) // 創(chuàng)建守護協(xié)程，自動對鎖進行續(xù)期
dl.status = true
首先創(chuàng)建一個cancelContext，它的context函數(shù)cancelFunc是給Unlock進行調用的。然后啟動一個goroutine進程來循環(huán)續(xù)期。
這個新啟動的goroutine在主goroutine處理結束，調用Unlock的時候，才會結束，否則會在 過期時間/2 的時候，調用一次redis的expire命令來進行續(xù)期。
至于外部，在使用的時候如下
func Foo() error {
key := foo
// 創(chuàng)建可重入的分布式鎖
dl := NewDistributeLockRedis(key, 10)
// 爭搶鎖
err := dl.TryLock()
if err != nil {
// 沒有搶到鎖
return err
}
// 搶到鎖的記得釋放鎖
defer func() {
dl.Unlock()
}
// 做真正的任務
DoJob()
}
參考文章
如果還想了解更多，以下的參考文章值得閱讀。
redisson
https://github.com/redisson/redisson
滴滴 曾奇：談談我所認識的分布式鎖
http://blog.itpub.net/69908606/viewspace-2644366/
Redis 分布式鎖｜從青銅到鉆石的五種演進方案
https://my.oschina.net/u/4499317/blog/5039486
分布式鎖中的王者方案 - Redisson
https://xie.infoq.cn/article/d8e897f768eb1a358a0fd6300#:~:text=Redisson%20%E6%98%AF%E4%B8%80%E4%B8%AA%E5%9C%A8Redis,In%2DMemory%20Data%20Grid%EF%BC%89%E3%80%82
redisson中的看門狗機制總結
https://www.cnblogs.com/jelly12345/p/14699492.html
Redis分布式鎖如何自動續(xù)期
https://blog.csdn.net/yangbindxj/article/details/123189395
到底什么是重入鎖，拜托，一次搞清楚！
https://zhuanlan.zhihu.com/p/71018541
Hi，我是軒脈刃，一個名不見經(jīng)傳碼農，體制內的小憤青，躁動的騷年，2022年想堅持寫一些學習/工作/思考筆記，謂之倒逼學習。歡迎關注個人公眾號：軒脈刃的刀光劍影。
推薦閱讀