Go語言使用Etcd實現分布式鎖

更新時間：2023年05月21日 11:49:45 作者：small_to_large

etcd是近幾年比較火熱的一個開源的、分布式的鍵值對數據存儲系統(tǒng)，本文將介紹如何利用Etcd實現分布式鎖，感興趣的小伙伴可以跟隨小編一起了解一下

1 分布式鎖概述

談到分布式鎖，必然是因為單機鎖無法滿足要求，在現階段微服務多實例部署的情況下，單機語言級別的鎖，無法滿足并發(fā)互斥資源的安全訪問。常見的單機鎖如Java的jvm鎖Lock、synchronized，golang的Mutex等對于分布式鎖有很多種實現方式，常見的有以下幾種：

基于數據庫：通過數據庫事務鎖例如for update操作
基于緩存中間件：redis分布式鎖、etcd分布式鎖等
基于ZK臨時節(jié)點：zookeeper 臨時節(jié)點實現分布式鎖

每種方式實現的分布式鎖各有優(yōu)缺點簡單介紹一下：

數據庫實現不用額外引入新的中間件，減少系統(tǒng)的依賴性和不穩(wěn)定性，但性能不會太高，且并發(fā)量大時，對數據庫壓力比較大。
ZK實現分布式，因為zk滿足了CP，能夠保證其數據一致性，不會出現加鎖成功后又丟失的問題，但相反性能會降低，并且可用性降低 CAP A不是滿足的，詳細可以自行了解zk細節(jié)
redis 實現：最大的優(yōu)點性能高，能保證AP，保證其高可用。但無法保證一致性，因為redis滿足的是AP，可能存在某一個時間節(jié)點集群數據S-M同步不一致。

2 分布式鎖要點

實現分布式鎖需要滿足一下幾點：

鎖載體：redis 受用 K-V 鍵值作為鎖載體，ZK使用臨時節(jié)點作為載體
鎖租期：進程持有分布式鎖后不能一直占用，如果因為宕機情況造成鎖釋放失敗，就會一直占用，reds 可以設置過期時間，zk臨時節(jié)點也會自動刪除。
其他要求：比如減少驚群效應、可重入機制、公平鎖機制，不同的實現方式有的不能完全滿足。

分布式鎖選擇：

qps不大的情況下，那種方式都可以
結合目前技術體系，在不引入新的技術中間件情況下解決問題
qps并發(fā)極高，但容忍極少的數據丟失或者不一致，建議使用redis實現分布式鎖
如果業(yè)務要求任何情況下都不允許數據丟失，可以使用zk或者etcd實現

3 Etcd 實現機制

鎖載體： 使用 k-v 結構實現
鎖租期： Etcd 通過lease可以對 kv 設置租約，當租約到期，kv 將失效刪除；避免長時間占用鎖不釋放放。
自動續(xù)期： Etcd 可以對租約進行自動續(xù)期，通過KeepAlive實現
公平鎖： 多個程序同時搶鎖時，會根據 Revision 值大小依次獲得鎖，可以有效避免 “驚群效應”，公平獲取。
Watch 機制： 監(jiān)聽機制，Watch 機制支持 Watch 某個固定的 key或者目錄， key 或目錄發(fā)生變化，客戶端可以收到通知。

4 代碼實現

操作步驟：

初始化客戶端
創(chuàng)建一個session并設置默認租期30s
獲取指定前綴的鎖對象
加鎖
執(zhí)行業(yè)務
釋放鎖

代碼：

package main
import (
	"context"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
	"log"
	"time"
)
func main() {
	// 初始化客戶端
	log.Println("客戶端初始化")
	client, err := clientv3.New(clientv3.Config{Endpoints: []string{"localhost:2379"}, DialTimeout: time.Second * 3})
	if err != nil {
		log.Fatalf("客戶端初始化失敗:%v\n", err)
	}
	// 創(chuàng)建一個session并設置默認租期30s,即鎖默認超過30s會自動釋放(內部會自動續(xù)期Etcd KeepAlive)
	log.Println("Session初始化")
	session, err := concurrency.NewSession(client, concurrency.WithTTL(30))
	if err != nil {
		log.Fatalf("Session初始化失敗:%v\n", err)
		return
	}
	defer func(session *concurrency.Session) {
		err := session.Close()
		if err != nil {
			log.Fatalf("Session關閉失敗:%v\n", err)
		}
	}(session)
	// 獲取指定前綴的鎖對象
	mutex := concurrency.NewMutex(session, "my-lock")
	// 加鎖默認等待3s
	log.Println("TryLock加鎖失敗不會等待")
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	err = mutex.TryLock(ctx)
	if err != nil {
		log.Fatalf("加鎖失敗立即返回:%v\n", err)
		return
	}
	//log.Println("加鎖最多等待3s")
	//ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	//defer cancel()
	//err = mutex.Lock(ctx)
	//if err != nil {
	//	log.Fatalf("加鎖失敗:%v\n", err)
	//	return
	//}
	// Exe biz
	log.Println("加鎖成功開始執(zhí)行業(yè)務")
	for i := 1; i <= 10; i++ {
		time.Sleep(time.Second)
		log.Printf("執(zhí)行 %%%d ...", i*10)
	}
	// 釋放鎖
	err = mutex.Unlock(context.TODO())
	if err != nil {
		log.Fatalf("釋放鎖失敗:%v\n", err)
		return
	}
	log.Println("釋放鎖完成")
}

測試結果