Go 客户端基础

etcd 提供了官方的 Go 客户端库，本章将学习如何使用 Go 连接和操作 etcd。

客户端概述

客户端类型

etcd 提供两种客户端包：

client/v3：gRPC 客户端（推荐）
client/v2：HTTP 客户端（已弃用）

安装

bash

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

连接客户端

基本连接

package main

import (
    "context"
    "fmt"
    "time"

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

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

    fmt.Println("Connected to etcd")
}

配置选项

cli, err := clientv3.New(clientv3.Config{
    Endpoints:            []string{"localhost:2379", "localhost:22379", "localhost:32379"},
    DialTimeout:          5 * time.Second,
    DialKeepAliveTime:    10 * time.Second,
    DialKeepAliveTimeout: 3 * time.Second,
    MaxCallSendMsgSize:   10 * 1024 * 1024, // 10MB
    MaxCallRecvMsgSize:   10 * 1024 * 1024, // 10MB
    Username:             "root",
    Password:             "password",
})

TLS 连接

cli, err := clientv3.New(clientv3.Config{
    Endpoints:   []string{"https://localhost:2379"},
    DialTimeout: 5 * time.Second,
    TLS: &tls.Config{
        Certificates: []tls.Certificate{cert},
        RootCAs:      caCertPool,
    },
})

客户端结构

主要组件

type Client struct {
    KV          // KV 操作
    Lease       // 租约操作
    Watch       // 监听操作
    Cluster     // 集群管理
    Auth        // 认证管理
    Maintenance // 维护操作
}

使用示例

// KV 操作
resp, err := cli.KV.Get(ctx, "/foo")

// 租约操作
leaseResp, err := cli.Lease.Grant(ctx, 60)

// 监听操作
watchChan := cli.Watch(ctx, "/foo")

// 集群管理
memberList, err := cli.Cluster.MemberList(ctx)

// 认证管理
authResp, err := cli.Auth.Enable(ctx)

// 维护操作
statusResp, err := cli.Maintenance.Status(ctx)

Context 使用

etcd 客户端的所有操作都接受 context.Context，用于控制超时和取消。

超时控制

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
resp, err := cli.Get(ctx, "/foo")
cancel()

if err == context.DeadlineExceeded {
    fmt.Println("操作超时")
}

取消操作

ctx, cancel := context.WithCancel(context.Background())

// 开始监听
watchCh := cli.Watch(ctx, "/foo")

// 取消监听
cancel()

连接管理

连接池

etcd 客户端内部维护 gRPC 连接池，自动管理连接复用。

┌──────────────────────────────────────────────────────┐
│                   Client                              │
│                                                      │
│  ┌──────────────────────────────────────────────┐  │
│  │              gRPC Connection Pool             │  │
│  │  ┌────────┐ ┌────────┐ ┌────────┐           │  │
│  │  │Conn 1  │ │Conn 2  │ │Conn 3  │           │  │
│  │  └────────┘ └────────┘ └────────┘           │  │
│  └──────────────────────────────────────────────┘  │
│                                                      │
│  自动负载均衡到多个 Endpoints                       │
└──────────────────────────────────────────────────────┘

多节点连接

// 连接多个节点，客户端自动负载均衡
cli, err := clientv3.New(clientv3.Config{
    Endpoints: []string{
        "node1:2379",
        "node2:2379",
        "node3:2379",
    },
    DialTimeout: 5 * time.Second,
})

// 客户端会：
// 1. 随机选择一个节点连接
// 2. 节点失败时自动切换到其他节点
// 3. 定期检查节点健康状态

获取实际连接节点

// 查询实际连接的节点
statusResp, err := cli.Maintenance.Status(ctx, "node1:2379")
fmt.Printf("Leader: %s\n", statusResp.Leader)

重试策略

自动重试

etcd 客户端内置重试机制：

// 默认重试策略
cli, err := clientv3.New(clientv3.Config{
    Endpoints:   []string{"localhost:2379"},
    DialTimeout: 5 * time.Second,
    // 默认重试次数：无限制（针对网络错误）
    // 默认重试间隔：基于 gRPC backoff 算法
})

自定义重试

import "github.com/grpc-ecosystem/go-grpc-middleware/retry"

cli, err := clientv3.New(clientv3.Config{
    Endpoints:   []string{"localhost:2379"},
    DialTimeout: 5 * time.Second,
    DialOptions: []grpc.DialOption{
        grpc.WithUnaryInterceptor(
            grpc_retry.UnaryClientInterceptor(
                grpc_retry.WithMax(3),
                grpc_retry.WithPerRetryTimeout(2*time.Second),
            ),
        ),
    },
})

错误处理

错误类型

import (
    "errors"
    clientv3 "go.etcd.io/etcd/client/v3"
    "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
)

func handleError(err error) {
    switch {
    case err == context.Canceled:
        fmt.Println("请求被取消")
    case err == context.DeadlineExceeded:
        fmt.Println("请求超时")
    case err == rpctypes.ErrKeyNotFound:
        fmt.Println("键不存在")
    case err == rpctypes.ErrLeaseNotFound:
        fmt.Println("租约不存在")
    case err == rpctypes.ErrCompacted:
        fmt.Println("请求的 revision 已被压缩")
    case errors.Is(err, rpctypes.ErrGRPCEmptyKey):
        fmt.Println("空键")
    case errors.Is(err, rpctypes.ErrGRPCTooManyOps):
        fmt.Println("事务操作过多")
    default:
        fmt.Printf("其他错误: %v\n", err)
    }
}

错误示例

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

resp, err := cli.Get(ctx, "/nonexistent-key")
if err != nil {
    if err == rpctypes.ErrKeyNotFound {
        fmt.Println("键不存在（正常情况）")
    } else {
        handleError(err)
    }
}

线性一致性读

etcd 默认提供线性一致性读，但可以通过配置优化性能。

线性一致性读（默认）

// 默认：线性一致性读（强一致性）
resp, err := cli.Get(ctx, "/foo")
// 等价于
resp, err := cli.Get(ctx, "/foo", clientv3.WithSerializable())

串行化读（性能优化）

// 串行化读：可能读到旧数据，但性能更高
resp, err := cli.Get(ctx, "/foo", clientv3.WithSerializable())

适用场景：

对一致性要求不高
读取频繁，需要优化性能
可以接受短暂的数据延迟

实用工具

打印响应

import (
    "encoding/json"
    "fmt"
    clientv3 "go.etcd.io/etcd/client/v3"
)

func printResponse(resp *clientv3.GetResponse) {
    data, _ := json.MarshalIndent(resp, "", "  ")
    fmt.Println(string(data))
}

// 示例输出
{
  "header": {
    "cluster_id": 123456789,
    "member_id": 987654321,
    "revision": 100,
    "raft_term": 2
  },
  "kvs": [
    {
      "key": "/foo",
      "create_revision": 1,
      "mod_revision": 100,
      "version": 10,
      "value": "bar"
    }
  ],
  "count": 1
}

检查连接

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

    _, err := cli.Get(ctx, "health-check-key")
    if err != nil && err != rpctypes.ErrKeyNotFound {
        return err
    }
    return nil
}

获取集群状态

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

    for _, endpoint := range cli.Endpoints() {
        resp, err := cli.Maintenance.Status(ctx, endpoint)
        if err != nil {
            fmt.Printf("%s: %v\n", endpoint, err)
            continue
        }
        fmt.Printf("%s: Leader=%s, DBSize=%d, Version=%s\n",
            endpoint,
            resp.Leader,
            resp.DbSize,
            resp.Version,
        )
    }
}

完整示例

基础客户端封装

package etcdutil

import (
    "context"
    "fmt"
    "time"

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

type Client struct {
    cli *clientv3.Client
}

func NewClient(endpoints []string) (*Client, error) {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, fmt.Errorf("failed to connect etcd: %w", err)
    }

    return &Client{cli: cli}, nil
}

func (c *Client) Close() error {
    return c.cli.Close()
}

func (c *Client) Put(ctx context.Context, key, value string) error {
    _, err := c.cli.Put(ctx, key, value)
    return err
}

func (c *Client) Get(ctx context.Context, key string) (string, error) {
    resp, err := c.cli.Get(ctx, key)
    if err != nil {
        return "", err
    }
    if len(resp.Kvs) == 0 {
        return "", fmt.Errorf("key not found")
    }
    return string(resp.Kvs[0].Value), nil
}

func (c *Client) Delete(ctx context.Context, key string) error {
    _, err := c.cli.Delete(ctx, key)
    return err
}

使用封装

func main() {
    client, err := etcdutil.NewClient([]string{"localhost:2379"})
    if err != nil {
        panic(err)
    }
    defer client.Close()

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

    // Put
    if err := client.Put(ctx, "/config/app", "production"); err != nil {
        panic(err)
    }

    // Get
    value, err := client.Get(ctx, "/config/app")
    if err != nil {
        panic(err)
    }
    fmt.Printf("Value: %s\n", value)

    // Delete
    if err := client.Delete(ctx, "/config/app"); err != nil {
        panic(err)
    }
}

最佳实践

连接管理
- 使用 defer 关闭客户端
- 复用客户端连接，避免频繁创建
- 连接多个节点实现高可用
超时控制
- 所有操作都设置合理的超时
- 避免使用无限超时的 context
错误处理
- 区分预期错误（如键不存在）和异常错误
- 网络错误时重试或切换节点
性能优化
- 读多场景使用串行化读
- 批量操作使用事务
- 避免频繁创建租约
资源管理
- 及时关闭 Watch channel
- 及时撤销不需要的租约
- 避免 context 泄漏

总结

本章学习了：

客户端连接和配置
Context 超时控制
连接池和负载均衡
重试和错误处理
线性一致性读
实用工具和封装

掌握客户端基础是进行 etcd 开发的前提。在下一章中，我们将深入学习 KV 操作的各种用法。

Go 客户端基础 ​

客户端概述 ​

客户端类型 ​

安装 ​

连接客户端 ​

基本连接 ​

配置选项 ​

TLS 连接 ​

客户端结构 ​

主要组件 ​

使用示例 ​

Context 使用 ​

超时控制 ​

取消操作 ​

连接管理 ​

连接池 ​

多节点连接 ​

获取实际连接节点 ​

重试策略 ​

自动重试 ​

自定义重试 ​

错误处理 ​

错误类型 ​

错误示例 ​

线性一致性读 ​

线性一致性读（默认） ​

串行化读（性能优化） ​

实用工具 ​

打印响应 ​

检查连接 ​

获取集群状态 ​

完整示例 ​

基础客户端封装 ​

使用封装 ​

最佳实践 ​

总结 ​