快捷導(dǎo)航

Golang使用etcd構(gòu)建分布式鎖的示例分享

更新時(shí)間：2025年01月09日 10:52:00 作者：夢想畫家

在本教程中,我們將學(xué)習(xí)如何使用Go和etcd構(gòu)建分布式鎖系統(tǒng),分布式鎖系統(tǒng)對(duì)于管理對(duì)分布式系統(tǒng)中共享資源的并發(fā)訪問至關(guān)重要,它有助于維護(hù)一致性,防止競爭條件,并確保在任何給定時(shí)間只有一個(gè)進(jìn)程獨(dú)占訪問資源,需要的朋友可以參考下

引言

我們將使用Go作為編程語言，并使用etcd作為分布式鍵值存儲(chǔ)。Go的并發(fā)特性和對(duì)分布式系統(tǒng)的出色支持使其成為本教程的理想選擇。Etcd是一種高度可靠的分布式鍵值存儲(chǔ)，被許多大型系統(tǒng)（如Kubernetes）用于配置管理和服務(wù)發(fā)現(xiàn)。

在本教程結(jié)束時(shí)，你將能夠構(gòu)建一個(gè)分布式鎖系統(tǒng)，Go開發(fā)人員可以使用該系統(tǒng)來管理對(duì)其應(yīng)用程序中共享資源的訪問。

環(huán)境準(zhǔn)備

要學(xué)習(xí)本教程，你應(yīng)該具備：

具備Go編程語言基礎(chǔ)知識(shí)
在你的系統(tǒng)上安裝Etcd或訪問Etcd服務(wù)器
Go安裝在你的系統(tǒng)上（版本1.13或更高版本）

新建Go項(xiàng)目

首先，讓我們用必要的依賴項(xiàng)創(chuàng)建一個(gè)新的Go項(xiàng)目：

$ mkdir distributed-lock && cd distributed-lock
$ go mod init example.com/distributed-lock
$ go get go.etcd.io/etcd/clientv3

這將設(shè)置一個(gè)新的Go項(xiàng)目并下載etcd客戶端庫。

實(shí)現(xiàn)加鎖和解鎖功能

現(xiàn)在是時(shí)候?qū)崿F(xiàn)Lock和Unlock函數(shù)了。我們將創(chuàng)建名為lock的新文件。首先導(dǎo)入必要的包并定義結(jié)構(gòu)來保存鎖信息：

package main

import (
	"context"
	"log"
	"time"

	"go.etcd.io/etcd/clientv3"
)

type DistributedLock struct {
	Key        string
	Value      string
	LeaseID    clientv3.LeaseID
	etcdClient *clientv3.Client
}

接下來，我們將實(shí)現(xiàn)Lock函數(shù)。該函數(shù)將執(zhí)行以下步驟：

創(chuàng)建具有指定TTL（生存時(shí)間）的新租約
將鎖鍵值對(duì)存入附帶租約的etcd中
如果鎖已被占用，則處理錯(cuò)誤并重試

func (dl *DistributedLock) Lock(ctx context.Context, ttl int64) error {
	lease, err := dl.etcdClient.Grant(ctx, ttl)
	if err != nil {
		return err
	}

	_, err = dl.etcdClient.Put(ctx, dl.Key, dl.Value, clientv3.WithLease(lease.ID))
	if err != nil {
		return err
	}

	dl.LeaseID = lease.ID
	log.Printf("Lock acquired: %s", dl.Key)
	return nil
}

ctx context.Context：這是 Go 語言中用于傳遞上下文信息的參數(shù)，常用于控制操作的超時(shí)、取消等情況，比如在分布式環(huán)境中協(xié)調(diào)多個(gè)操作的生命周期，確保它們能按照預(yù)期執(zhí)行或者在合適的時(shí)候被取消。
ttl int64：代表 “Time To Live”（存活時(shí)間），通常是以秒為單位的時(shí)間長度，用于指定分布式鎖的有效時(shí)長，過了這個(gè)時(shí)長，鎖可能會(huì)自動(dòng)釋放，以防止鎖被長期占用導(dǎo)致死鎖等問題。
這里使用dl.etcdClient再次調(diào)用Put方法，它的作用是向etcd中寫入鍵值對(duì)（Key-Value）。具體來說，寫入的鍵是dl.Key（從代碼推測應(yīng)該是用于標(biāo)識(shí)這個(gè)分布式鎖的唯一鍵，是DistributedLock結(jié)構(gòu)體中的一個(gè)字段），值是dl.Value（同樣是結(jié)構(gòu)體中的字段，可能是和鎖相關(guān)的一些其他附屬信息等）。
關(guān)鍵的一點(diǎn)是通過clientv3.WithLease(lease.ID)這個(gè)選項(xiàng)將前面申請(qǐng)到的租約的ID關(guān)聯(lián)到這個(gè)要寫入的鍵值對(duì)上，意思是這個(gè)鍵值對(duì)的存在時(shí)長會(huì)受租約的控制，當(dāng)租約到期（達(dá)到ttl設(shè)定的時(shí)間）時(shí)，etcd會(huì)自動(dòng)刪除這個(gè)鍵值對(duì)，從而實(shí)現(xiàn)了分布式鎖的自動(dòng)釋放機(jī)制。和前面獲取租約類似，Put操作也可能出錯(cuò)，所以同樣進(jìn)行錯(cuò)誤判斷，如果err不為空，就返回錯(cuò)誤給調(diào)用者。

現(xiàn)在，讓我們實(shí)現(xiàn)Unlock函數(shù)。該函數(shù)將執(zhí)行以下步驟：

刪除etcd中的鎖鍵值對(duì)
解除租賃

func (dl *DistributedLock) Unlock(ctx context.Context) error {
	_, err := dl.etcdClient.Delete(ctx, dl.Key)
	if err != nil {
		return err
	}

	_, err = dl.etcdClient.Revoke(ctx, dl.LeaseID)
	if err != nil {
		return err
	}

	log.Printf("Lock released: %s", dl.Key)
	return nil
}

測試分布式鎖

最后，讓我們創(chuàng)建一個(gè)簡單的測試應(yīng)用程序來查看分布式鎖的實(shí)際情況。基本上。Go文件，添加以下代碼：

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"go.etcd.io/etcd/clientv3"
)

func main() {
	endpoints := []string{"localhost:2379"}

	cfg := clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: 5 * time.Second,
	}

	client, err := clientv3.New(cfg)
	if err != nil {
		fmt.Printf("Error connecting to etcd: %v", err)
		os.Exit(1)
	}
	defer client.Close()

	ctx := context.Background()
	lockKey := "my-lock"
	lockValue := "my-value"

	dl := DistributedLock{
		Key:        lockKey,
		Value:      lockValue,
		etcdClient: client,
	}

	err = dl.Lock(ctx, 10)
	if err != nil {
		fmt.Printf("Error acquiring lock: %v", err)
		os.Exit(1)
	}

	// Simulate a critical section
	time.Sleep(5 * time.Second)

	err = dl.Unlock(ctx)
	if err != nil {
		fmt.Printf("Error releasing lock: %v", err)
		os.Exit(1)
	}
}

使用以下命令運(yùn)行測試應(yīng)用程序：

$ go run main.go

如果一切正常，您應(yīng)該看到鎖在5秒睡眠后被獲取和釋放。

重構(gòu)實(shí)現(xiàn)失敗重試

以下是在原代碼基礎(chǔ)上添加申請(qǐng)鎖失敗重試機(jī)制的示例代碼，在 Go 語言中可以使用循環(huán)結(jié)合退避策略等方式來實(shí)現(xiàn)，以下是一種常見的實(shí)現(xiàn)思路及代碼示例，假設(shè)使用了time包來進(jìn)行時(shí)間控制以及添加了適當(dāng)?shù)腻e(cuò)誤處理和重試次數(shù)限制等邏輯：

package main

import (
    "context"
    "fmt"
    "go.etcd.io/etcd/clientv3"
    "log"
    "time"
)

type DistributedLock struct {
    etcdClient *clientv3.Client
    Key        string
    Value      string
    LeaseID    clientv3.LeaseID
}

func (dl *DistributedLock) Lock(ctx context.Context, ttl int64) error {
    maxRetries := 3       // 最大重試次數(shù)，可根據(jù)實(shí)際情況調(diào)整
    retryDelay := 1 * time.Second // 初始重試間隔時(shí)間，可根據(jù)實(shí)際情況調(diào)整
    for retry := 0; retry < maxRetries; retry++ {
        lease, err := dl.etcdClient.Grant(ctx, ttl)
        if err!= nil {
            if retry == maxRetries-1 {
                return fmt.Errorf("failed to grant lease after %d retries: %v", maxRetries, err)
            }
            // 等待一段時(shí)間后重試，這里可以采用退避策略，比如指數(shù)退避等，此處簡單使用固定間隔
            time.Sleep(retryDelay)
            continue
        }

        _, err = dl.etcdClient.Put(ctx, dl.Key, dl.Value, clientv3.WithLease(lease.ID))
        if err!= nil {
            if retry == maxRetries-1 {
                return fmt.Errorf("failed to put key-value with lease after %d retries: %v", maxRetries, err)
            }
            // 釋放本次申請(qǐng)到的租約，避免資源浪費(fèi)，雖然租約到期也會(huì)自動(dòng)釋放，但及時(shí)釋放更好
            _, _ = dl.etcdClient.Revoke(ctx, lease.ID)
            time.Sleep(retryDelay)
            continue
        }

        dl.LeaseID = lease.ID
        log.Printf("Lock acquired: %s", dl.Key)
        return nil
    }
    return fmt.Errorf("exceeded max retries for lock acquisition")
}

出來重試部分，其他邏輯不變，這里解釋第二部分重試邏輯：

當(dāng)租約申請(qǐng)成功后，嘗試將鍵值對(duì)寫入etcd并關(guān)聯(lián)租約，若這個(gè)操作出現(xiàn)錯(cuò)誤（err不為nil），同樣有如下處理：
先判斷是否達(dá)到最大重試次數(shù)，如果是，返回帶有詳細(xì)失敗信息的error告知調(diào)用者設(shè)置鍵值關(guān)聯(lián)租約操作經(jīng)過多次重試后失敗以及具體錯(cuò)誤原因。
如果沒達(dá)到最大重試次數(shù)，為了避免已經(jīng)申請(qǐng)到的租約一直占用資源（雖然租約到期會(huì)自動(dòng)釋放，但及時(shí)主動(dòng)釋放更合理），調(diào)用dl.etcdClient.Revoke方法來撤銷（釋放）剛剛申請(qǐng)到的租約，然后讓當(dāng)前協(xié)程暫停執(zhí)行retryDelay設(shè)定的時(shí)間間隔后，通過continue進(jìn)入下一次循環(huán)，再次嘗試整個(gè)申請(qǐng)鎖的流程。