go語言分布式id生成器及分布式鎖源碼分析

本文小編為大家詳細介紹“go語言分布式id生成器及分布式鎖源碼分析”，內容詳細，步驟清晰，細節(jié)處理妥當，希望這篇“go語言分布式id生成器及分布式鎖源碼分析”文章能幫助大家解決疑惑，下面跟著小編的思路慢慢深入，一起來學習新知識吧。

分布式 id 生成器

在分布式場景中，唯一 id 的生成算比較重要。

而通常在高并發(fā)場景中，需要類似 MySQL 自增 id 一樣不斷增長且又不會重復的 id，即 MySql 的主鍵 id。

比如，在電商 618 或者雙 11 搞活動的時候，一般在 0 點 開始，會有千萬到億級的訂單量寫入，每秒大概需要處理 10 萬加的訂單。

而在將訂單插入數(shù)據(jù)庫之前，我們在業(yè)務上需要給訂單一個唯一的 id，即利用 idMaker 生存唯一的訂單號，再插入數(shù)據(jù)庫內。如果生成的 id 是隨機且沒有含義的純數(shù)字的話，在大訂單量的情況下，對數(shù)據(jù)庫進行增刪改查時就不能起到提高效率的作用。所以 此 id 應該應該包含一些時間信息，機器信息等，這樣即使后端的系統(tǒng)對消息進行了分庫分表，也能夠以時間順序對這些消息進行排序了。

比較典型的就是推特的【雪花算法】了，在以上場景下可以算是最優(yōu)解，原理如圖：

首先確定的是，id 數(shù)值長度是 64 位，int64 類型，除去開頭的符號位 unused ，其它可以分為四個部分：

• 41 位來表示收到請求時的時間戳，單位為毫秒

• 5 位表示數(shù)據(jù)中心的 id

• 5 位表求機器的實例 id

• 12 位為循環(huán)自增 id，到達 1111,1111,1111 后歸就會 0

以上機制原理生成的 id，可以支持一臺機器在一毫秒內能夠產(chǎn)生 4096 條消息。也就是一秒共 409.6w 條消息。單單從值域上來講是完全夠用。

數(shù)據(jù)中心 id 加上實例 id 共有 10 位，每個數(shù)據(jù)中心可以部署 32 臺實例，搭建 32 個數(shù)據(jù)中心，所以可以一共部署 1024 臺實例。

而 41 位的時間戳（毫秒為單位）能夠使用 69 年。

worker_id 如何分配

timestamp（時間戳），datacenter_id（數(shù)據(jù)中心），worker_id（機器 ID） 和 sequence_id（序號） 這四個字段中，timestamp 和 sequence_id 是由程序在運行期生成的。但 datacenter_id 和 worker_id 需要在部署階段就要能夠獲取得到，并且一旦程序啟動之后，就是不可更改的了，因為如果可以隨意更改，可能會造成最終生成的 id 有沖突。

不過一般不同數(shù)據(jù)中心的機器，會提供對應的獲取數(shù)據(jù)中心 id 的 API，因此 datacenter_id 我們可以在部署階段輕松地獲取到。而 worker_id 是我們邏輯上給機器分配的一個 id，比較簡單的做法就是由能夠提供這種自增 id 功能的工具來支持，比如 MySql:

mysql> insert into a (ip) values("10.115.4.66");
Query OK, 1 row affected (0.00 sec)
mysql> select last_insert_id();
+------------------+
| last_insert_id() |
+------------------+
|                2 |
+------------------+
1 row in set (0.00 sec)

從 MySql 中獲取到 worker_id 之后，就把這個 worker_id 直接持久化到本地，以避免每次上線時都需要獲取新的 worker_id。讓單實例的 worker_id 可以始終保持不變。

但是，使用 MySQL 的話，相當于給 id 生成服務增加了一個外部依賴。當然依賴越多，服務的運維成本就會增加。

考慮到集群中即使有單個 id 生成服務的實例掛了，也就是損失一段時間的一部分 id，所以我們也可以更簡單暴力一些，把 worker_id 直接寫在 worker 的配置中，上線時，由部署腳本完成 worker_id 字段替換即可。

開源示例：標準雪花算法

github.com/bwmarrin/snowflake 是一個相對輕量級的 snowflake 的 Go 實現(xiàn)。其文檔對各位使用的定義如下圖所示：

此庫和標準的 snowflake 實現(xiàn)方式全完一致，使用也比較簡單，直接上示例代碼：

package main
import (
  "fmt"
 "github.com/bwmarrin/snowflake"
)
func main() {
 node, err := snowflake.NewNode(1)
 if err != nil {
  println(err.Error())
  os.Exit(1)
 }
 for i := 0; i < 20; i++ {
  id := node.Generate()
  fmt.Printf("Int64  ID: %d\n", id)
  fmt.Printf("String ID: %s\n", id)
  fmt.Printf("ID Time  : %d\n", id.Time())
  fmt.Printf("ID Node  : %d\n", id.Node())
  fmt.Printf("ID Step  : %d\n", id.Step())
  fmt.Println("--------- end ----------")
 }
}

分布式鎖

單機程序并發(fā)或并行修改全局共享變量時，需要對修改行為加鎖。因為如果不加鎖，多個協(xié)程序就會對該變量競爭，然后得到的結果就會不準確，或者說得到的結果不是我們所預期的，比如下面的例子：

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

多次運行結果不同：

?  go run main.go
884
?  go run main.go
957
?  go run main.go
923

預期的結果是：999

進程內加鎖

而如果想要得到正確（預期）的結果，要把計數(shù)器的操作代碼部分加上鎖：

package main
import (
 "fmt"
 "sync"
)
func main() {
 var wg sync.WaitGroup
 var lock sync.Mutex
 var count = 0
 for i := 1; i < 1000; i++ {
  wg.Add(1)
  go func() {
   defer wg.Done()
   lock.Lock()  // 加鎖
   count++
   lock.Unlock()  // 釋放鎖
  }()
 }
 wg.Wait()
 fmt.Println(count)
}

這樣能夠得到正確結果：

?  go run main.go
999

嘗試加鎖 tryLock

在某些場景，我們往往只希望一個任務有單一的執(zhí)行者，而不像計數(shù)器一樣，所有的 Goroutine 都成功執(zhí)行。后續(xù)的 Goroutine 在搶鎖失敗后，需要放棄執(zhí)行，這時候就需要用到嘗試加鎖，即實現(xiàn) trylock。

嘗試加鎖，在加鎖成功后執(zhí)行后續(xù)流程，失敗時不可以阻塞，而是直接返回加鎖的結果。

在 Go 語言中可以用大小為 1 的 Channel 來模擬 trylock：

package main
import (
  "fmt"
  "sync"
)
type MyLock struct {
 lockCh chan struct{}
}
func NewLock() MyLock {
 var myLock MyLock
 myLock = MyLock{
  lockCh:make(chan struct{}, 1),
 }
 myLock.lockCh <- struct{}{}
 return myLock
}
func (l *MyLock) Lock() bool {
 result := false
 select {
 case <-l.lockCh:
  result = true
 default:  // 這里去掉就會阻塞，直到獲取到鎖
 }
 return result
}
func (l *MyLock) Unlock() {
 l.lockCh <- struct{}{}
}
func main() {
 var wg sync.WaitGroup
 var count int
 l := NewLock()
 for i := 0; i < 10; i++ {
  wg.Add(1)
  go func() {
   defer wg.Done()
   if !l.Lock() {
    fmt.Println("get lock failed")
    return
   }
   count++
   fmt.Println("count=", count)
   l.Unlock()
  }()
 }
 wg.Wait()
}

每個 Goruntine 只有獲取到鎖（成功執(zhí)行了 Lock）才會繼續(xù)執(zhí)行后續(xù)代碼，然后在 Unlock()時可以保證 Lock 結構體里的 Channel 一定是空的，所以不會阻塞也不會失敗。

在單機系統(tǒng)中，tryLock 并不是一個好選擇，因為大量的 Goruntine 搶鎖會無意義地占用 cpu 資源，這就是活鎖，所有不建議使用這種鎖。

基于 Redis 的 setnx 分布式鎖

在分布式場景中，也需要“搶占”的邏輯，可以用 Redis 的 setnx 實現(xiàn)：

package main
import (
 "github.com/go-redis/redis"
 "sync"
 "time"
)
func setnx() {
 client := redis.NewClient(&redis.Options{})
 var lockKey = "counter_lock"
 var counterKey = "counter"
 // lock
 resp := client.SetNX(lockKey, 1, time.Second*6)
 lockStatus, err := resp.Result()
 if err != nil || !lockStatus {
  println("lock failed")
  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 {
   println(err)
  }
 }
 println("current counter is ", cntValue)
 // unlock
 delResp := client.Del(lockKey)
 unlockStatus, err := delResp.Result()
 if err == nil && unlockStatus > 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()
   setnx()
  }()
 }
 wg.Wait()
}

運行結果：

?  go run main.go
lock failed
lock failed
lock failed
lock failed
lock failed
current counter is  34
lock failed
unlock success

通過上面的代碼和執(zhí)行結果可以看到，遠程調用 setnx 運行流程上和單機的 troLock 非常相似，如果獲取鎖失敗，那么相關的任務邏輯就不會繼續(xù)向后執(zhí)行。

setnx 很適合高并發(fā)場景下用來爭搶一些“唯一”的資源。比如，商城秒殺的商品，在某個時間點，多個買家會對其進行下單并發(fā)爭搶。這種場景我們沒有辦法依賴具體的時間來判斷先后，因為不同設備的時間不能保證使用的是統(tǒng)一的時間，也就不能保證時序。

所以，我們需要依賴于這些請求到達 redis 節(jié)點的順序來做正確的搶鎖操作。

如果用戶的網(wǎng)絡環(huán)境比較差，是有可能搶不到的。

基于 ZooKeeper 分布式鎖

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

package main
import (
 "github.com/go-zookeeper/zk"
 "time"
)
func main() {
 c, _, err := zk.Connect([]string{"127.0.0.1"}, time.Second)
 if err != nil {
  panic(err)
 }
 l := zk.NewLock(c, "/lock", zk.WorldACL(zk.PermAll))
 err = l.Lock()
 if err != nil {
  panic(err)
 }
 println("lock success, do your business logic")
 time.Sleep(time.Second * 10) // 模擬業(yè)務處理
 l.Unlock()
 println("unlock success, finish business logic")
}

其原理也是基于臨時 Sequence 節(jié)點和 watch API，例如我們這里使用的是 /lock 節(jié)點。

Lock 會在該節(jié)點下的節(jié)點列表中插入自己的值，只要節(jié)點下的子節(jié)點發(fā)生變化，就會通知所有 watch 該節(jié)點的程序。這時候程序會檢查當前節(jié)點下最小的子節(jié)點的 id 是否與自己的一致。如果一致，說明加鎖成功了。

這種分布式的阻塞鎖比較適合分布式任務調度場景，但不適合高頻次持鎖時間短的搶鎖場景。

一般基于強一致協(xié)議的鎖適用于粗粒度的加鎖操作。這里的粗粒度指鎖占用時間較長。我們在使用時也應思考在自己的業(yè)務場景中使用是否合適。

讀到這里，這篇“go語言分布式id生成器及分布式鎖源碼分析”文章已經(jīng)介紹完畢，想要掌握這篇文章的知識點還需要大家自己動手實踐使用過才能領會，如果想了解更多相關內容的文章，歡迎關注億速云行業(yè)資訊頻道。