etcd框架实践【Go版】

FunTester

发布于 2024-07-05 14:16:47

1240

发布于 2024-07-05 14:16:47

文章被收录于专栏：FunTesterFunTester

etcd是一个分布式的键值存储系统,由CoreOS公司开发,主要用于为分布式系统提供可靠和高可用的配置管理和服务发现功能。etcd基于Raft一致性算法设计,可以有效地处理网络分区等容错问题,确保数据在集群中的一致性和可靠性。

etcd被广泛应用于Kubernetes、Cloud Foundry、Mesos等分布式系统和云原生应用中,充当了可信赖的配置存储和服务注册发现等重要角色。除此之外,etcd也可作为分布式锁、队列服务、消息发布订阅系统等使用。总之,etcd作为一个可靠的分布式键值存储框架,为构建分布式系统提供了很好的基础支持。

特点与使用场景

etcd作为一个分布式的键值存储系统,具有以下一些显著的特点:

简单的数据模型。etcd采用键值对的数据模型,非常简单直观,易于使用和理解。同时支持监视机制和原子事务操作。
强一致性保证。etcd基于Raft一致性算法,能够有效处理网络分区等容错场景,确保数据在集群中的完全一致性。在任何时候,集群中只有一个主节点处理写入操作。
高可用性。etcd通过Raft算法自动处理节点故障切换,即使部分节点宕机,只要集群中存在多数派节点,整个集群依然可用。
良好的扩展性。etcd支持动态添加或删除集群节点,实现水平扩展或缩减集群规模。易于按需配置适合的集群大小。
监视和通知机制。etcd支持监视某个键前缀的变化,并实时通知。适合存储配置信息,实现配置中心。
完善的访问控制。etcd支持基于RBAC的访问控制,并支持认证、传输加密等安全特性。
方便的集成能力。etcd提供了易于使用的RESTful HTTP API,支持多种语言的客户端库,便于集成到应用程序中。

于此对应的，etcd主要应用于以下几个场景：

服务发现。etcd 常用于服务发现，在微服务架构中尤为重要。服务可以将自身的信息（如 IP 地址、端口等）注册到 etcd 中，其他服务可以从 etcd 中查找所需的服务地址，简化了服务间的通信和协调。
配置管理。etcd 是一个理想的配置管理存储系统，能够存储应用程序和系统的配置信息，并且支持实时更新。通过监听机制，应用程序可以实时响应配置的变更，避免了配置文件频繁修改带来的麻烦。
分布式锁。etcd 提供了原子操作和分布式锁功能，可以用于协调分布式系统中的任务调度。通过使用 etcd 的分布式锁机制，多个节点可以安全地进行同步操作，防止竞争条件和数据不一致问题。
领导选举。在分布式系统中，领导选举是一个常见需求。etcd 通过其强一致性的特性和分布式锁机制，能够实现高效的领导选举，确保系统中只有一个领导节点在工作。
集群管理。etcd 经常用作集群管理工具，例如 Kubernetes 的核心组件之一就是 etcd。它用于存储整个集群的状态数据，包括节点信息、Pod 状态、配置数据等，确保集群的一致性和可靠性。
元数据存储。etcd 可以作为分布式系统的元数据存储。例如，在大数据处理系统中，etcd 可以存储任务调度信息、节点状态等元数据，帮助系统高效运行。
分布式协调和一致性。etcd 的强一致性和高可用性特性，使其适合作为分布式系统的协调和一致性存储。在需要多个组件协同工作的场景中，etcd 可以提供可靠的数据存储和一致性保证。

Go语言实践

依赖

首先我们添加依赖，这次我依然选择了命令行添加。

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

执行完之后，mod文件增加了一下内容：

go.etcd.io/etcd/api/v3 v3.5.14 // indirect  
go.etcd.io/etcd/client/pkg/v3 v3.5.14 // indirect  
go.etcd.io/etcd/client/v3 v3.5.14 // indirect

准备

首先我们使用前两天学到的zap框架添加一个全局的日志对象logger：

var Logger *zap.SugaredLogger // 日志  
  
func init() {  
    encoderConfig := zapcore.EncoderConfig{ // 创建编码配置  
       TimeKey:        "T",                           // 时间键  
       LevelKey:       "L",                           // 日志级别键  
       NameKey:        "log",                         // 日志名称键  
       CallerKey:      "C",                           // 日志调用键  
       MessageKey:     "msg",                         // 日志消息键  
       StacktraceKey:  "stacktrace",                  // 堆栈跟踪键  
       LineEnding:     zapcore.DefaultLineEnding,     // 行结束符,默认为 \n       EncodeLevel:    zapcore.CapitalLevelEncoder,   // 日志级别编码器,将日志级别转换为大写  
       EncodeTime:     zapcore.ISO8601TimeEncoder,    // 时间编码器,将时间格式化为 ISO8601 格式  
       EncodeDuration: zapcore.StringDurationEncoder, // 持续时间编码器,将持续时间编码为字符串  
       EncodeCaller:   zapcore.ShortCallerEncoder,    // 调用编码器,显示文件名和行号  
    }  
    encoder := zapcore.NewConsoleEncoder(encoderConfig)                    // 创建控制台编码器,使用编码配置  
    atomicLevel := zap.NewAtomicLevel()                                    // 创建原子级别,用于动态设置日志级别  
    atomicLevel.SetLevel(zap.InfoLevel)                                    // 设置日志级别,只有 Info 级别及以上的日志才会输出  
    core := zapcore.NewCore(encoder, zapcore.Lock(os.Stdout), atomicLevel) // 将日志输出到标准输出  
    Logger = zap.New(core, zap.AddCaller(), zap.Development()).Sugar()     // 创建 Logger,添加调用者和开发模式  
}

服务端

单节点的etcd服务比较简单，集群的稍微麻烦一些，由于我只是用来作为演示服务，所以选择了最简单的方法：

brew install etcd

然后执行 etcd 命令即可启动一个etcd服务，默认端口号是 2379 。

客户端

我定义了一个全局的客户端，代码如下：

const timeOut = 5 * time.Second // 超时时间

var cli *clientv3.Client // etcd 客户端,全局变量  
  
// init  
//  
//  @Description: 初始化连接  
func init() {  
    client, err := clientv3.New(clientv3.Config{  
       Endpoints:   []string{"localhost:2379"}, // etcd 服务器地址  
       DialTimeout: timeOut,                    // 连接超时时间  
    })  
    if err != nil {  
       panic("连接服务器失败!!!") // 初始化失败  
    } else {  
       cli = client // 初始化成功  
    }  
}

然后我定义了一个关闭客户端的方法：

// close  
//  
//  @Description: 关闭连接  
func close() {  
    func(cli *clientv3.Client) {  
       err := cli.Close()  
       if err != nil {  
          Logger.Error("关闭连接失败!!!")  
       }  
    }(cli)  
}

读写测试

下面是根据官方文档写了一个读写测试：

func TestEtcd(t *testing.T) {  
    defer close()                                                     // 关闭连接  
    ctx, cancel := context.WithTimeout(context.Background(), timeOut) //  
    _, _ = cli.Put(ctx, "key", "value")                               // 写入键值对  
    cancel()                                                          // 取消上下文  
    ctx, cancel = context.WithTimeout(context.Background(), timeOut)  // 重新创建上下文  
    resp, _ := cli.Get(ctx, "", clientv3.WithPrefix())                // 获取所有键值对  
    cancel()                                                          // 取消上下文  
    for _, kv := range resp.Kvs {                                     // 遍历键值对  
       Logger.Infof("%s: %s\n", kv.Key, kv.Value)       // 打印键值对  
       cli.Delete(context.Background(), string(kv.Key)) // 删除键值对  
    }  
}

控制台输出内容：

=== RUN   TestEtcd
2024-06-04T20:10:16.212+0800 INFO test/etcd_test.go:79 key: value

2024-06-04T20:10:16.222+0800 INFO test/etcd_test.go:85 
2024-06-04T20:10:16.222+0800 INFO test/etcd_test.go:86 删除结果: 0
--- PASS: TestEtcd (0.04s)
PASS

锁

etcd框架一个主要特征就是分布式，可以用来进行分布式锁的实现，以及基于分布式锁其他功能的实现，下面分享etcd锁的使用。

func TestLock(t *testing.T) {  
    defer close()                                         // 关闭连接  
    session, _ := concurrency.NewSession(cli)             // 创建会话  
    defer session.Close()                                 // 关闭会话  
    mutex := concurrency.NewMutex(session, "/funtester/") // 创建互斥锁,锁定 funtester 键  
    ctx := context.Background()                           // 创建上下文  
    if err := mutex.Lock(ctx); err != nil {               // 加锁,如果失败,打印错误信息  
       Logger.Error("加锁失败:", err)  
       return  
    }  
    Logger.Info("加锁成功")                       // 加锁成功  
    if err := mutex.Unlock(ctx); err != nil { // 解锁,如果失败,打印错误信息  
       Logger.Error("解锁失败:", err)  
       return  
    }  
    Logger.Info("解锁成功") // 解锁成功  
}

控制台打印日志如下：

=== RUN   TestLock
2024-06-04T22:14:17.820+0800 INFO test/etcd_test.go:92 加锁成功
2024-06-04T22:14:17.826+0800 INFO test/etcd_test.go:97 解锁成功
--- PASS: TestLock (0.06s)
PASS

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2024-06-30，如有侵权请联系 cloudcommunity@tencent.com 删除