Go 客户端 - 事务和锁

本章将深入学习 etcd 的事务机制和并发控制，包括分布式锁、选举等高级用法。

事务机制

事务概述

etcd 事务提供原子性的多操作执行，支持条件判断。

Txn 结构：
┌─────────────────────────────────────┐
│  Compare (条件判断)                  │
│  ├── Key1.Value == "expected"       │
│  ├── Key2.CreateRevision > 0        │
│  └── Key3.Version < 10              │
├─────────────────────────────────────┤
│  Success (条件满足时执行)            │
│  ├── Put Key1 "new-value"           │
│  ├── Put Key2 "another-value"       │
│  └── Get Key3                       │
├─────────────────────────────────────┤
│  Failure (条件不满足时执行)          │
│  ├── Get Key1                       │
│  └── Get Key2                       │
└─────────────────────────────────────┘

基本事务

package main

import (
    "context"
    "fmt"
    "time"

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

func main() {
    cli, _ := clientv3.New(clientv3.Config{
        Endpoints:   []string{"localhost:2379"},
        DialTimeout: 5 * time.Second,
    })
    defer cli.Close()

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

    // 基本事务：检查键是否存在
    txnResp, err := cli.Txn(ctx).
        If(clientv3.Compare(clientv3.Value("/foo"), "=", "bar")).
        Then(clientv3.OpPut("/foo", "new-bar")).
        Else(clientv3.OpGet("/foo")).
        Commit()

    if err != nil {
        panic(err)
    }

    if txnResp.Succeeded {
        fmt.Println("Transaction succeeded, value updated")
    } else {
        fmt.Println("Transaction failed, value not equal to 'bar'")
        getResp := (*clientv3.GetResponse)(txnResp.Responses[0].GetResponseRange())
        if len(getResp.Kvs) > 0 {
            fmt.Printf("Current value: %s\n", getResp.Kvs[0].Value)
        }
    }
}

Compare 操作

// 多种比较类型
txnResp, _ := cli.Txn(ctx).
    If(
        // 比较值
        clientv3.Compare(clientv3.Value("/key"), "=", "expected"),

        // 比较创建 revision
        clientv3.Compare(clientv3.CreateRevision("/key"), "=", 0),

        // 比较修改 revision
        clientv3.Compare(clientv3.ModRevision("/key"), ">", 100),

        // 比较版本
        clientv3.Compare(clientv3.Version("/key"), "<", 10),

        // 比较租约
        clientv3.Compare(clientv3.Lease("/key"), "=", leaseID),
    ).
    Then(clientv3.OpPut("/key", "new-value")).
    Commit()

比较运算符

运算符	说明
"="	等于
"<"	小于
"<="	小于等于
">"	大于
">="	大于等于
"!="	不等于

多条件事务

// 多个条件（AND 逻辑）
txnResp, _ := cli.Txn(ctx).
    If(
        clientv3.Compare(clientv3.Value("/key1"), "=", "value1"),
        clientv3.Compare(clientv3.Value("/key2"), "=", "value2"),
    ).
    Then(
        clientv3.OpPut("/key1", "new-value1"),
        clientv3.OpPut("/key2", "new-value2"),
    ).
    Else(
        clientv3.OpGet("/key1"),
        clientv3.OpGet("/key2"),
    ).
    Commit()

批量操作（无条件）

// 无条件执行批量操作
txnResp, _ := cli.Txn(ctx).
    Then(
        clientv3.OpPut("/key1", "value1"),
        clientv3.OpPut("/key2", "value2"),
        clientv3.OpPut("/key3", "value3"),
    ).
    Commit()

读取批量操作结果

// 执行多个 Get 操作
txnResp, _ := cli.Txn(ctx).
    Then(
        clientv3.OpGet("/key1"),
        clientv3.OpGet("/key2"),
        clientv3.OpGet("/key3"),
    ).
    Commit()

// 解析结果
for i, resp := range txnResp.Responses {
    getResp := (*clientv3.GetResponse)(resp.GetResponseRange())
    if len(getResp.Kvs) > 0 {
        fmt.Printf("Key %d: %s\n", i+1, getResp.Kvs[0].Value)
    }
}

分布式锁

原理

etcd 分布式锁基于事务和租约实现：

1. 检查锁键是否存在（CreateRevision == 0）
2. 如果不存在，创建锁键并绑定租约
3. 如果存在，等待或返回失败

基本锁实现

func acquireLock(cli *clientv3.Client, key string, ttl int64) (clientv3.LeaseID, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // 创建租约
    leaseResp, err := cli.Lease.Grant(ctx, ttl)
    if err != nil {
        return 0, err
    }

    // 尝试获取锁
    txnResp, err := cli.Txn(ctx).
        If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
        Then(clientv3.OpPut(key, "locked", clientv3.WithLease(leaseResp.ID))).
        Commit()

    if err != nil {
        cli.Lease.Revoke(ctx, leaseResp.ID)
        return 0, err
    }

    if !txnResp.Succeeded {
        cli.Lease.Revoke(ctx, leaseResp.ID)
        return 0, fmt.Errorf("lock already held")
    }

    // 启动 KeepAlive
    go func() {
        keepAliveCh, _ := cli.Lease.KeepAlive(context.Background(), leaseResp.ID)
        for resp := range keepAliveCh {
            if resp == nil {
                return
            }
        }
    }()

    return leaseResp.ID, nil
}

func releaseLock(cli *clientv3.Client, leaseID clientv3.LeaseID) error {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    _, err := cli.Lease.Revoke(ctx, leaseID)
    return err
}

等待锁释放

func waitForLock(cli *clientv3.Client, key string) error {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // 监听锁键的删除事件
    watchCh := cli.Watch(ctx, key)

    for watchResp := range watchCh {
        for _, event := range watchResp.Events {
            if event.Type == clientv3.EventTypeDelete {
                fmt.Println("Lock released")
                return nil
            }
        }
    }

    return fmt.Errorf("timeout waiting for lock")
}

使用 Mutex SDK

etcd 官方提供了并发控制 SDK：go.etcd.io/etcd/client/v3/concurrency

go get go.etcd.io/etcd/client/v3/concurrency@v3.5.17

import (
    "context"
    "time"

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

func mutexExample() {
    cli, _ := clientv3.New(clientv3.Config{
        Endpoints:   []string{"localhost:2379"},
        DialTimeout: 5 * time.Second,
    })
    defer cli.Close()

    // 创建 Session（管理租约）
    session, _ := concurrency.NewSession(cli, concurrency.WithTTL(10))

    // 创建 Mutex
    mutex := concurrency.NewMutex(session, "/locks/my-resource")

    // 获取锁（阻塞）
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    if err := mutex.Lock(ctx); err != nil {
        fmt.Println("Failed to acquire lock:", err)
        return
    }

    fmt.Println("Lock acquired")

    // 执行关键操作
    doCriticalWork()

    // 释放锁
    mutex.Unlock(context.Background())
    fmt.Println("Lock released")

    session.Close()
}

尝试获取锁（非阻塞）

// TryLock 不阻塞，立即返回结果
if err := mutex.TryLock(ctx); err != nil {
    fmt.Println("Lock not available:", err)
    return
}

// 成功获取锁
doCriticalWork()
mutex.Unlock(ctx)

带超时的锁

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

if err := mutex.Lock(ctx); err != nil {
    if err == context.DeadlineExceeded {
        fmt.Println("Timeout acquiring lock")
    }
    return
}

// 成功获取锁
mutex.Unlock(context.Background())

Leader 选举

原理

Leader 选举基于分布式锁：获得锁的节点成为 Leader，其他节点成为 Follower。

候选者流程：
1. 尝试获取选举锁
2. 成功 → 成为 Leader，保持锁
3. 失败 → 成为 Follower，监听锁释放
4. 锁释放 → 再次尝试获取锁

使用 Election SDK

func electionExample() {
    cli, _ := clientv3.New(clientv3.Config{
        Endpoints:   []string{"localhost:2379"},
        DialTimeout: 5 * time.Second,
    })
    defer cli.Close()

    // 创建 Session
    session, _ := concurrency.NewSession(cli, concurrency.WithTTL(10))

    // 创建 Election
    election := concurrency.NewElection(session, "/elections/my-leader")

    // 参加选举
    ctx := context.Background()

    // Campaign 成为 Leader（阻塞）
    if err := election.Campaign(ctx, "node-1"); err != nil {
        fmt.Println("Campaign failed:", err)
        return
    }

    fmt.Println("I am the Leader!")

    // 执行 Leader 工作
    go doLeaderWork()

    // 监听 Leader 变更
    go observeLeader(cli, election)

    // 保持 Leader 状态
    select {}

    // 主动放弃 Leader
    election.Resign(ctx)
}

func observeLeader(cli *clientv3.Client, election *concurrency.Election) {
    ch := election.Observe(context.Background())

    for resp := range ch {
        if len(resp.Kvs) > 0 {
            leader := string(resp.Kvs[0].Value)
            fmt.Printf("Current leader: %s\n", leader)
        }
    }
}

查询当前 Leader

// 查询当前 Leader
leader, _ := election.Leader(context.Background())
fmt.Printf("Current leader: %s\n", leader)

// Proclaim 更新 Leader 值
election.Proclaim(ctx, "new-leader-value")

选举示例：主备切换

type LeaderService struct {
    cli      *clientv3.Client
    session  *concurrency.Session
    election *concurrency.Election
    isLeader bool
}

func NewLeaderService(cli *clientv3.Client, name string) *LeaderService {
    session, _ := concurrency.NewSession(cli, concurrency.WithTTL(10))
    election := concurrency.NewElection(session, "/elections/service")

    ls := &LeaderService{
        cli:      cli,
        session:  session,
        election: election,
    }

    // 启动选举循环
    go ls.runElection(name)

    return ls
}

func (ls *LeaderService) runElection(name string) {
    for {
        ctx := context.Background()

        // 参加选举
        err := ls.election.Campaign(ctx, name)
        if err != nil {
            fmt.Println("Campaign error:", err)
            time.Sleep(1 * time.Second)
            continue
        }

        // 成为 Leader
        ls.isLeader = true
        fmt.Println("I am the Leader now!")

        // 执行 Leader 工作
        ls.doLeaderWork()

        // Leader 工作结束（可能是主动 Resign 或 Session 失效）
        ls.isLeader = false
        fmt.Println("I am no longer the Leader")
    }
}

func (ls *LeaderService) doLeaderWork() {
    // Leader 负责的工作
    // 例如：定期执行任务、处理关键请求等
}

func (ls *LeaderService) Resign() error {
    return ls.election.Resign(context.Background())
}

func (ls *LeaderService) Close() {
    ls.session.Close()
}

STM（软件事务内存）

etcd 提供 STM（Software Transactional Memory）支持更复杂的事务逻辑。

基本 STM

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

func stmExample(cli *clientv3.Client) {
    // STM 事务：原子性地读取和修改多个键
    resp, err := concurrency.NewSTM(cli, func(stm concurrency.STM) error {
        // 读取值
        value1 := stm.Get("/key1")
        value2 := stm.Get("/key2")

        // 修改值
        stm.Put("/key1", value1+"-modified")
        stm.Put("/key2", value2+"-modified")

        return nil
    })

    if err != nil {
        fmt.Println("STM failed:", err)
    }
}

银行转账示例

func transfer(cli *clientv3.Client, from, to string, amount int) error {
    _, err := concurrency.NewSTM(cli, func(stm concurrency.STM) error {
        // 获取账户余额
        fromBalance := stm.Get("/accounts/" + from)
        toBalance := stm.Get("/accounts/" + to)

        // 解析余额
        fromInt, _ := strconv.Atoi(fromBalance)
        toInt, _ := strconv.Atoi(toBalance)

        // 检查余额
        if fromInt < amount {
            return fmt.Errorf("insufficient balance")
        }

        // 更新余额
        stm.Put("/accounts/"+from, strconv.Itoa(fromInt-amount))
        stm.Put("/accounts/"+to, strconv.Itoa(toInt+amount))

        return nil
    })

    return err
}

STM 嵌套操作

// STM 支持复杂的条件逻辑
resp, err := concurrency.NewSTM(cli, func(stm concurrency.STM) error {
    value := stm.Get("/config/feature")

    if value == "enabled" {
        stm.Put("/config/count", stm.Get("/config/count")+"1")
    } else {
        stm.Put("/config/feature", "enabled")
    }

    return nil
})

实用示例

限流器

func rateLimiter(cli *clientv3.Client, key string, limit int, window time.Duration) bool {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    now := time.Now().Unix()
    windowStart := now - int64(window.Seconds())

    // 获取窗口内的计数
    resp, _ := cli.Get(ctx, key)

    var count int64 = 0
    if len(resp.Kvs) > 0 {
        count, _ = strconv.ParseInt(string(resp.Kvs[0].Value), 10, 64)
    }

    // 检查是否超过限制
    if count >= int64(limit) {
        return false
    }

    // 增加计数（事务）
    txnResp, _ := cli.Txn(ctx).
        If(clientv3.Compare(clientv3.Value(key), "=", strconv.FormatInt(count, 10))).
        Then(clientv3.OpPut(key, strconv.FormatInt(count+1, 10))).
        Commit()

    return txnResp.Succeeded
}

唯一 ID 生成

func generateUniqueID(cli *clientv3.Client, prefix string) (int64, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    key := "/ids/" + prefix

    // 原子性地获取并递增 ID
    txnResp, err := cli.Txn(ctx).
        Then(
            clientv3.OpGet(key),
            clientv3.OpPut(key, "next-value"),
        ).
        Commit()

    if err != nil {
        return 0, err
    }

    // 解析当前 ID
    getResp := (*clientv3.GetResponse)(txnResp.Responses[0].GetResponseRange())
    var id int64 = 1
    if len(getResp.Kvs) > 0 {
        id, _ = strconv.ParseInt(string(getResp.Kvs[0].Value), 10, 64)
        id++
    }

    // 更新 ID（使用事务保证原子性）
    _, err = cli.Txn(ctx).
        If(clientv3.Compare(clientv3.Value(key), "=", string(getResp.Kvs[0].Value))).
        Then(clientv3.OpPut(key, strconv.FormatInt(id, 10))).
        Commit()

    return id, err
}

配置更新（原子性）

func updateConfig(cli *clientv3.Client, key string, updater func(string) string) error {
    return concurrency.NewSTM(cli, func(stm concurrency.STM) error {
        current := stm.Get(key)
        updated := updater(current)
        stm.Put(key, updated)
        return nil
    })
}

性能考虑

事务限制

• 单个事务最多支持 128 个操作
• 事务中的操作应尽量少，避免大事务

事务冲突

• 多个事务并发修改同一键时，只有一个能成功
• 设计时应减少冲突概率

STM 重试

STM 会自动重试失败的事务，但重试次数有限：

// 设置 STM 重试选项
resp, err := concurrency.NewSTM(cli, func(stm concurrency.STM) error {
    stm.Put("/key", "value")
    return nil
}, concurrency.WithAbortContext(context.Background()))

最佳实践

1. 事务设计

   • 操作尽量少，避免大事务
   • 减少热点键的冲突
   • 使用合适的比较条件

2. 锁使用

   • 设置合理的 TTL
   • 实现 KeepAlive 保持锁
   • 及时释放锁

3. 选举设计

   • 监听 Leader 变更
   • 实现 Follower 备份逻辑
   • 处理网络抖动

4. 错误处理

   • 区分超时、冲突等错误
   • 实现重试逻辑
   • 处理 Session 失效

总结

本章学习了：

• 事务机制：Compare、Then、Else
• 分布式锁：基于事务和租约
• Leader 选举：基于选举锁
• STM：复杂事务逻辑
• 实用示例：限流器、唯一 ID、配置更新

这些机制是构建分布式系统的核心工具。在下一章中，我们将学习 etcd 的生产部署和运维实践。