Go 語言 分布式鎖

原文鏈接：https://chai2010.cn/advanced-go-programming-book/ch6-cloud/ch6-02-lock.html

6.2 分布式鎖

在單機程序并發(fā)或并行修改全局變量時，需要對修改行為加鎖以創(chuàng)造臨界區(qū)。為什么需要加鎖呢？我們看看在不加鎖的情況下并發(fā)計數(shù)會發(fā)生什么情況：

package main

import (
    "sync"
)

// 全局變量
var counter int

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
        defer wg.Done()
            counter++
        }()
    }

    wg.Wait()
    println(counter)
}

多次運行會得到不同的結(jié)果：

??? go run local_lock.go
945
??? go run local_lock.go
937
??? go run local_lock.go
959

6.2.1 進程內(nèi)加鎖

想要得到正確的結(jié)果的話，要把對計數(shù)器（counter）的操作代碼部分加上鎖：

// ... 省略之前部分
var wg sync.WaitGroup
var l sync.Mutex
for i := 0; i < 1000; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        l.Lock()
        counter++
        l.Unlock()
    }()
}

wg.Wait()
println(counter)
// ... 省略之后部分

這樣就可以穩(wěn)定地得到計算結(jié)果了：

??? go run local_lock.go
1000

6.2.2 trylock

在某些場景，我們只是希望一個任務(wù)有單一的執(zhí)行者。而不像計數(shù)器場景一樣，所有 goroutine 都執(zhí)行成功。后來的 goroutine 在搶鎖失敗后，需要放棄其流程。這時候就需要 trylock 了。

trylock 顧名思義，嘗試加鎖，加鎖成功執(zhí)行后續(xù)流程，如果加鎖失敗的話也不會阻塞，而會直接返回加鎖的結(jié)果。在 Go 語言中我們可以用大小為 1 的 Channel 來模擬 trylock：

package main

import (
    "sync"
)

// Lock try lock
type Lock struct {
    c chan struct{}
}

// NewLock generate a try lock
func NewLock() Lock {
    var l Lock
    l.c = make(chan struct{}, 1)
    l.c <- struct{}{}
    return l
}

// Lock try lock, return lock result
func (l Lock) Lock() bool {
    lockResult := false
    select {
    case <-l.c:
        lockResult = true
    default:
    }
    return lockResult
}

// Unlock , Unlock the try lock
func (l Lock) Unlock() {
    l.c <- struct{}{}
}

var counter int

func main() {
    var l = NewLock()
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            if !l.Lock() {
                // log error
                println("lock failed")
                return
            }
            counter++
            println("current counter", counter)
            l.Unlock()
        }()
    }
    wg.Wait()
}

因為我們的邏輯限定每個 goroutine 只有成功執(zhí)行了 Lock 才會繼續(xù)執(zhí)行后續(xù)邏輯，因此在 Unlock 時可以保證 Lock 結(jié)構(gòu)體中的 channel 一定是空，從而不會阻塞，也不會失敗。上面的代碼使用了大小為 1 的 channel 來模擬 trylock，理論上還可以使用標準庫中的 CAS 來實現(xiàn)相同的功能且成本更低，讀者可以自行嘗試。

在單機系統(tǒng)中，trylock 并不是一個好選擇。因為大量的 goroutine 搶鎖可能會導致 CPU 無意義的資源浪費。有一個專有名詞用來描述這種搶鎖的場景：活鎖。

活鎖指的是程序看起來在正常執(zhí)行，但 CPU 周期被浪費在搶鎖，而非執(zhí)行任務(wù)上，從而程序整體的執(zhí)行效率低下?；铈i的問題定位起來要麻煩很多。所以在單機場景下，不建議使用這種鎖。

6.2.3 基于 Redis 的 setnx

在分布式場景下，我們也需要這種 “搶占” 的邏輯，這時候怎么辦呢？我們可以使用 Redis 提供的 setnx 命令：

package main

import (
    "fmt"
    "sync"
    "time"

    "github.com/go-redis/redis"
)

func incr() {
    client := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "", // no password set
        DB:       0,  // use default DB
    })

    var lockKey = "counter_lock"
    var counterKey = "counter"

    // lock
    resp := client.SetNX(lockKey, 1, time.Second*5)
    lockSuccess, err := resp.Result()

    if err != nil || !lockSuccess {
        fmt.Println(err, "lock result:", lockSuccess)
        return
    }

    // counter ++
    getResp := client.Get(counterKey)
    cntValue, err := getResp.Int64()
    if err == nil || err == redis.Nil {
        cntValue++
        resp := client.Set(counterKey, cntValue, 0)
        _, err := resp.Result()
        if err != nil {
            // log err
            println("set value error!")
        }
    }
    println("current counter is", cntValue)

    delResp := client.Del(lockKey)
    unlockSuccess, err := delResp.Result()
    if err == nil && unlockSuccess > 0 {
        println("unlock success!")
    } else {
        println("unlock failed", err)
    }
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            incr()
        }()
    }
    wg.Wait()
}

看看運行結(jié)果：

??? go run redis_setnx.go
<nil> lock result:  false
<nil> lock result:  false
<nil> lock result:  false
<nil> lock result:  false
<nil> lock result:  false
<nil> lock result:  false
<nil> lock result:  false
<nil> lock result:  false
<nil> lock result:  false
current counter is  2028
unlock success!

通過代碼和執(zhí)行結(jié)果可以看到，我們遠程調(diào)用 setnx 運行流程上和單機的 trylock 非常相似，如果獲取鎖失敗，那么相關(guān)的任務(wù)邏輯就不應該繼續(xù)向前執(zhí)行。

setnx 很適合在高并發(fā)場景下，用來爭搶一些 “唯一” 的資源。比如交易撮合系統(tǒng)中賣家發(fā)起訂單，而多個買家會對其進行并發(fā)爭搶。這種場景我們沒有辦法依賴具體的時間來判斷先后，因為不管是用戶設(shè)備的時間，還是分布式場景下的各臺機器的時間，都是沒有辦法在合并后保證正確的時序的。哪怕是我們同一個機房的集群，不同的機器的系統(tǒng)時間可能也會有細微的差別。

所以，我們需要依賴于這些請求到達 Redis 節(jié)點的順序來做正確的搶鎖操作。如果用戶的網(wǎng)絡(luò)環(huán)境比較差，那也只能自求多福了。

6.2.4 基于 ZooKeeper

package main

import (
    "time"

    "github.com/samuel/go-zookeeper/zk"
)

func main() {
    c, _, err := zk.Connect([]string{"127.0.0.1"}, time.Second) //*10)
    if err != nil {
        panic(err)
    }
    l := zk.NewLock(c, "/lock", zk.WorldACL(zk.PermAll))
    err = l.Lock()
    if err != nil {
        panic(err)
    }
    println("lock succ, do your business logic")

    time.Sleep(time.Second * 10)

    // do some thing
    l.Unlock()
    println("unlock succ, finish business logic")
}

基于 ZooKeeper 的鎖與基于 Redis 的鎖的不同之處在于 Lock 成功之前會一直阻塞，這與我們單機場景中的 mutex.Lock 很相似。

其原理也是基于臨時 Sequence 節(jié)點和 watch API，例如我們這里使用的是 /lock 節(jié)點。Lock 會在該節(jié)點下的節(jié)點列表中插入自己的值，只要節(jié)點下的子節(jié)點發(fā)生變化，就會通知所有 watch 該節(jié)點的程序。這時候程序會檢查當前節(jié)點下最小的子節(jié)點的 id 是否與自己的一致。如果一致，說明加鎖成功了。

這種分布式的阻塞鎖比較適合分布式任務(wù)調(diào)度場景，但不適合高頻次持鎖時間短的搶鎖場景。按照 Google 的 Chubby 論文里的闡述，基于強一致協(xié)議的鎖適用于 粗粒度 的加鎖操作。這里的粗粒度指鎖占用時間較長。我們在使用時也應思考在自己的業(yè)務(wù)場景中使用是否合適。

6.2.5 基于 etcd

etcd 是分布式系統(tǒng)中，功能上與 ZooKeeper 類似的組件，這兩年越來越火了。上面基于 ZooKeeper 我們實現(xiàn)了分布式阻塞鎖，基于 etcd，也可以實現(xiàn)類似的功能：

package main

import (
    "log"

    "github.com/zieckey/etcdsync"
)

func main() {
    m, err := etcdsync.New("/lock", 10, []string{"http://127.0.0.1:2379"})
    if m == nil || err != nil {
        log.Printf("etcdsync.New failed")
        return
    }
    err = m.Lock()
    if err != nil {
        log.Printf("etcdsync.Lock failed")
        return
    }

    log.Printf("etcdsync.Lock OK")
    log.Printf("Get the lock. Do something here.")

    err = m.Unlock()
    if err != nil {
        log.Printf("etcdsync.Unlock failed")
    } else {
        log.Printf("etcdsync.Unlock OK")
    }
}

etcd 中沒有像 ZooKeeper 那樣的 Sequence 節(jié)點。所以其鎖實現(xiàn)和基于 ZooKeeper 實現(xiàn)的有所不同。在上述示例代碼中使用的 etcdsync 的 Lock 流程是：

1. 先檢查 ?/lock? 路徑下是否有值，如果有值，說明鎖已經(jīng)被別人搶了
2. 如果沒有值，那么寫入自己的值。寫入成功返回，說明加鎖成功。寫入時如果節(jié)點被其它節(jié)點寫入過了，那么會導致加鎖失敗，這時候到 3
3. watch ?/lock? 下的事件，此時陷入阻塞
4. 當 ?/lock? 路徑下發(fā)生事件時，當前進程被喚醒。檢查發(fā)生的事件是否是刪除事件（說明鎖被持有者主動 unlock），或者過期事件（說明鎖過期失效）。如果是的話，那么回到 1，走搶鎖流程。

值得一提的是，在 etcdv3 的 API 中官方已經(jīng)提供了可以直接使用的鎖 API，讀者可以查閱 etcd 的文檔做進一步的學習。

6.2.7 如何選擇合適的鎖

業(yè)務(wù)還在單機就可以搞定的量級時，那么按照需求使用任意的單機鎖方案就可以。

如果發(fā)展到了分布式服務(wù)階段，但業(yè)務(wù)規(guī)模不大，qps 很小的情況下，使用哪種鎖方案都差不多。如果公司內(nèi)已有可以使用的 ZooKeeper、etcd 或者 Redis 集群，那么就盡量在不引入新的技術(shù)棧的情況下滿足業(yè)務(wù)需求。

業(yè)務(wù)發(fā)展到一定量級的話，就需要從多方面來考慮了。首先是你的鎖是否在任何惡劣的條件下都不允許數(shù)據(jù)丟失，如果不允許，那么就不要使用 Redis 的 setnx 的簡單鎖。

對鎖數(shù)據(jù)的可靠性要求極高的話，那只能使用 etcd 或者 ZooKeeper 這種通過一致性協(xié)議保證數(shù)據(jù)可靠性的鎖方案。但可靠的背面往往都是較低的吞吐量和較高的延遲。需要根據(jù)業(yè)務(wù)的量級對其進行壓力測試，以確保分布式鎖所使用的 etcd 或 ZooKeeper 集群可以承受得住實際的業(yè)務(wù)請求壓力。需要注意的是，etcd 和 Zookeeper 集群是沒有辦法通過增加節(jié)點來提高其性能的。要對其進行橫向擴展，只能增加搭建多個集群來支持更多的請求。這會進一步提高對運維和監(jiān)控的要求。多個集群可能需要引入 proxy，沒有 proxy 那就需要業(yè)務(wù)去根據(jù)某個業(yè)務(wù) id 來做分片。如果業(yè)務(wù)已經(jīng)上線的情況下做擴展，還要考慮數(shù)據(jù)的動態(tài)遷移。這些都不是容易的事情。

在選擇具體的方案時，還是需要多加思考，對風險早做預估。