Go语言etcd分布式锁实现高并发场景互斥访问实战

import (
"context"
"log"
"time"

"go.etcd.io/etcd/client/v3"
)

type EtcdClient struct {
client *clientv3.Client
}

func NewEtcdClient(endpoints []string) (*EtcdClient, error) {
cli, err := clientv3.New(clientv3.Config{
Endpoints: endpoints,
DialTimeout: 5 * time.Second,
})
if err != nil {
return nil, err
}
// 验证连接
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_, err = cli.Get(ctx, "health-check")
if err != nil {
return nil, err
}
return &EtcdClient{client: cli}, nil
}

import (
"context"
"fmt"
"time"

"go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/client/v3/concurrency"
)

type DistributedLock struct {
client *clientv3.Client
lease clientv3.Lease
leaseID clientv3.LeaseID
key string
ttl int64
cancelKeepAlive context.CancelFunc
}

func (e *EtcdClient) NewLock(key string, ttl int64) *DistributedLock {
return &DistributedLock{
client: e.client,
key: key,
ttl: ttl,
}
}

func (dl *DistributedLock) Acquire(ctx context.Context) error {
// 创建租约
lease := clientv3.NewLease(dl.client)
leaseResp, err := lease.Grant(ctx, dl.ttl)
if err != nil {
return fmt.Errorf("创建租约失败: %w", err)
}
dl.lease = lease
dl.leaseID = leaseResp.ID

// 启动自动续约
keepAliveCtx, keepAliveCancel := context.WithCancel(ctx)
dl.cancelKeepAlive = keepAliveCancel

ch, err := dl.client.KeepAlive(keepAliveCtx, dl.leaseID)
if err != nil {
lease.Revoke(ctx, dl.leaseID)
return fmt.Errorf("启动续约失败: %w", err)
}
go func() {
for range ch {
// 消费续约响应，防止channel阻塞
}
}()

// 使用事务原子创建key
key := dl.key
txn := dl.client.Txn(ctx).
If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
Then(clientv3.OpPut(key, fmt.Sprintf("%d", dl.leaseID), clientv3.WithLease(dl.leaseID))).
Else(clientv3.OpGet(key))

txnResp, err := txn.Commit()
if err != nil {
keepAliveCancel()
lease.Revoke(ctx, dl.leaseID)
return fmt.Errorf("事务提交失败: %w", err)
}

if !txnResp.Succeeded {
// 锁已被持有
keepAliveCancel()
lease.Revoke(ctx, dl.leaseID)
return fmt.Errorf("锁已被其他实例持有")
}

return nil
}

import (
"context"
"log"
"time"

"go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/client/v3/concurrency"
)

func (e *EtcdClient) WithLock(ctx context.Context, key string, fn func() error) error {
// 创建session，TTL设为15秒
session, err := concurrency.NewSession(e.client,
concurrency.WithTTL(15))
if err != nil {
return fmt.Errorf("创建session失败: %w", err)
}
defer session.Close()

mutex := concurrency.NewMutex(session, key)

// 获取锁，支持取消和超时
err = mutex.Lock(ctx)
if err != nil {
return fmt.Errorf("获取锁失败: %w", err)
}
defer mutex.Unlock(ctx)

log.Printf("成功获取锁: %s", key)

// 执行业务逻辑
return fn()
}

// 使用示例
func ExampleUsage() {
client, _ := NewEtcdClient([]string{"http://127.0.0.1:2379"})
defer client.Close()

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

err := client.WithLock(ctx, "/lock/order-1234", func() error {
// 在锁保护下执行业务操作
log.Println("正在处理订单...")
time.Sleep(2 * time.Second)
log.Println("订单处理完成")
return nil
})

for i := 0; i < maxRetries; i++ {
lockCtx, cancel := context.WithTimeout(ctx, 3*time.Second)

session, err := concurrency.NewSession(e.client, concurrency.WithTTL(10))
if err != nil {
cancel()
lastErr = err
time.Sleep(time.Duration(i+1) * 100 * time.Millisecond) // 指数退避
continue
}

mutex := concurrency.NewMutex(session, key)
err = mutex.Lock(lockCtx)
cancel()

if err == nil {
return &DistributedLock{
client: e.client,
session: session,
mutex: mutex,
key: key,
}, nil
}

session.Close()
lastErr = err

if err == context.DeadlineExceeded {
// 超时，等待后重试
time.Sleep(time.Duration(i+1) * 200 * time.Millisecond)
} else {
break
}
}

func (e *EtcdClient) WithLockMetrics(ctx context.Context, key string, fn func() error) (*LockMetrics, error) {
metrics := &LockMetrics{Key: key}
start := time.Now()
metrics.AcquireTime = start

err := e.WithLock(ctx, key, func() error {
metrics.WaitDuration = time.Since(start).Milliseconds()
holdStart := time.Now()
err := fn()
metrics.HoldDuration = time.Since(holdStart).Milliseconds()
metrics.Success = (err == nil)
return err
})