etcd分布式锁实践 - 元素码农

发布时间: 2025-04-07 14:12

↑

# etcd分布式锁实践

本文将详细介绍如何使用etcd实现分布式锁功能，包括基本概念、实现原理和最佳实践。

## 分布式锁概述

### 什么是分布式锁

分布式锁是分布式系统中的一种同步机制，用于在分布式环境下对共享资源进行互斥访问。它具有以下特点：

1. 互斥性：任何时刻只能有一个客户端持有锁
2. 可重入性：同一个客户端可以多次获取同一把锁
3. 高可用性：锁服务要保证高可用
4. 防死锁：即使客户端崩溃，锁也能被释放

### 为什么选择etcd

etcd实现分布式锁的优势：

1. 基于租约的自动过期机制
2. 事务操作保证原子性
3. 高可用的分布式架构
4. Watch机制支持锁状态监控

## 实现原理

### 基本流程

```plaintext
1. 创建租约
2. 基于租约创建key
3. 使用事务确保key创建的原子性
4. 获取锁成功后执行业务逻辑
5. 完成后释放锁或等待租约过期
```

### 核心实现

1. 分布式锁接口
```go
type Lock interface {
    // 获取锁
    Lock(ctx context.Context) error
    // 释放锁
    Unlock(ctx context.Context) error
}

type EtcdLock struct {
    client     *clientv3.Client
    key        string
    lease      clientv3.Lease
    leaseID    clientv3.LeaseID
    cancelFunc context.CancelFunc
}
```

2. 创建锁实例
```go
func NewEtcdLock(client *clientv3.Client, key string) *EtcdLock {
    return &EtcdLock{
        client: client,
        key:    key,
        lease:  clientv3.NewLease(client),
    }
}
```

3. 获取锁
```go
func (l *EtcdLock) Lock(ctx context.Context) error {
    // 创建租约
    resp, err := l.lease.Grant(ctx, 10)
    if err != nil {
        return fmt.Errorf("创建租约失败: %v", err)
    }
    l.leaseID = resp.ID

// 创建可取消的Context用于自动续约
    ctx, cancel := context.WithCancel(context.Background())
    l.cancelFunc = cancel

// 自动续约
    keepAliveCh, err := l.client.KeepAlive(ctx, l.leaseID)
    if err != nil {
        return fmt.Errorf("设置自动续约失败: %v", err)
    }

// 处理续约响应
    go func() {
        for {
            select {
            case <-ctx.Done():
                return
            case resp := <-keepAliveCh:
                if resp == nil {
                    return
                }
            }
        }
    }()

// 尝试获取锁
    txn := l.client.Txn(ctx)
    txn.If(
        clientv3.Compare(clientv3.CreateRevision(l.key), "=", 0),
    ).Then(
        clientv3.OpPut(l.key, "", clientv3.WithLease(l.leaseID)),
    ).Else(
        clientv3.OpGet(l.key),
    )

txnResp, err := txn.Commit()
    if err != nil {
        l.cancelFunc()
        return fmt.Errorf("获取锁失败: %v", err)
    }

if !txnResp.Succeeded {
        l.cancelFunc()
        return fmt.Errorf("锁已被占用")
    }

return nil
}
```

4. 释放锁
```go
func (l *EtcdLock) Unlock(ctx context.Context) error {
    defer func() {
        // 取消自动续约
        if l.cancelFunc != nil {
            l.cancelFunc()
        }
        // 撤销租约
        if l.leaseID != 0 {
            l.lease.Revoke(context.Background(), l.leaseID)
        }
    }()

// 删除key
    _, err := l.client.Delete(ctx, l.key)
    if err != nil {
        return fmt.Errorf("释放锁失败: %v", err)
    }
    return nil
}
```

## 最佳实践

### 锁超时处理

1. 设置获取锁超时
```go
func (l *EtcdLock) LockWithTimeout(timeout time.Duration) error {
    ctx, cancel := context.WithTimeout(context.Background(), timeout)
    defer cancel()
    return l.Lock(ctx)
}
```

2. 阻塞等待锁
```go
func (l *EtcdLock) LockWithRetry(ctx context.Context, retryInterval time.Duration) error {
    for {
        err := l.Lock(ctx)
        if err == nil {
            return nil
        }

select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(retryInterval):
            continue
        }
    }
}
```

### 可重入锁实现

1. 可重入锁结构
```go
type ReentrantLock struct {
    *EtcdLock
    ownerKey string
    count    int
}

func NewReentrantLock(client *clientv3.Client, key, owner string) *ReentrantLock {
    return &ReentrantLock{
        EtcdLock: NewEtcdLock(client, key),
        ownerKey: fmt.Sprintf("%s/owner", key),
    }
}
```

2. 可重入锁实现
```go
func (l *ReentrantLock) Lock(ctx context.Context) error {
    // 检查当前持有者
    resp, err := l.client.Get(ctx, l.ownerKey)
    if err != nil {
        return err
    }

// 如果是当前持有者，增加计数
    if len(resp.Kvs) > 0 && string(resp.Kvs[0].Value) == l.owner {
        l.count++
        return nil
    }

// 获取新锁
    if err := l.EtcdLock.Lock(ctx); err != nil {
        return err
    }

// 设置持有者
    _, err = l.client.Put(ctx, l.ownerKey, l.owner, clientv3.WithLease(l.leaseID))
    if err != nil {
        l.EtcdLock.Unlock(ctx)
        return err
    }

l.count = 1
    return nil
}

func (l *ReentrantLock) Unlock(ctx context.Context) error {
    if l.count > 1 {
        l.count--
        return nil
    }
    return l.EtcdLock.Unlock(ctx)
}
```

### 读写锁实现

1. 读写锁结构
```go
type RWLock struct {
    client  *clientv3.Client
    prefix  string
    wlock   *EtcdLock
    readers *sync.Map
}

func NewRWLock(client *clientv3.Client, prefix string) *RWLock {
    return &RWLock{
        client:  client,
        prefix:  prefix,
        wlock:   NewEtcdLock(client, prefix+"/write"),
        readers: &sync.Map{},
    }
}
```

2. 读写锁方法
```go
func (l *RWLock) RLock(ctx context.Context) error {
    // 确保没有写锁
    resp, err := l.client.Get(ctx, l.prefix+"/write")
    if err != nil {
        return err
    }
    if len(resp.Kvs) > 0 {
        return fmt.Errorf("写锁已被占用")
    }

// 创建读锁
    id := uuid.New().String()
    rlock := NewEtcdLock(l.client, l.prefix+"/read/"+id)
    if err := rlock.Lock(ctx); err != nil {
        return err
    }

l.readers.Store(id, rlock)
    return nil
}

func (l *RWLock) RUnlock(ctx context.Context) error {
    var lastErr error
    l.readers.Range(func(key, value interface{}) bool {
        rlock := value.(*EtcdLock)
        if err := rlock.Unlock(ctx); err != nil {
            lastErr = err
            return false
        }
        l.readers.Delete(key)
        return true
    })
    return lastErr
}

func (l *RWLock) Lock(ctx context.Context) error {
    // 等待所有读锁释放
    resp, err := l.client.Get(ctx, l.prefix+"/read/", clientv3.WithPrefix())
    if err != nil {
        return err
    }
    if len(resp.Kvs) > 0 {
        return fmt.Errorf("存在读锁")
    }

return l.wlock.Lock(ctx)
}

func (l *RWLock) Unlock(ctx context.Context) error {
    return l.wlock.Unlock(ctx)
}
```

## 应用示例

### 完整的分布式锁示例

```go
package main

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

func main() {
    // 创建etcd客户端
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"localhost:2379"},
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

// 创建分布式锁
    lock := NewEtcdLock(client, "/locks/my-lock")

// 获取锁
    ctx := context.Background()
    err = lock.Lock(ctx)
    if err != nil {
        log.Fatal(err)
    }

log.Println("获取锁成功")

// 模拟业务逻辑
    time.Sleep(5 * time.Second)

// 释放锁
    err = lock.Unlock(ctx)
    if err != nil {
        log.Fatal(err)
    }

log.Println("释放锁成功")
}
```

## 注意事项

1. 锁超时设置
   - 合理设置租约TTL
   - 考虑业务执行时间
   - 实现自动续约机制

2. 异常处理
   - 处理网络异常
   - 实现锁重试机制
   - 确保锁能被释放

3. 性能优化
   - 避免长时间持有锁
   - 合理设置重试间隔
   - 使用读写锁提高并发

4. 死锁预防
   - 设置锁超时时间
   - 实现锁重试机制
   - 保证锁的可靠释放

## 总结

etcd实现分布式锁的优点：

1. 基于租约的自动过期机制防止死锁
2. 事务操作保证加锁过程的原子性
3. Watch机制支持锁状态的实时监控
4. 分布式架构保证锁服务的高可用

在实际应用中，需要注意：

1. 正确处理锁的获取和释放
2. 实现可靠的异常处理机制
3. 优化锁的性能
4. 防止死锁发生