基于Etcdserver包将自己的Go程序打造成高可用系统

我们每一个系统开发人员都希望自己的程序永远不宕机，高可用是很多系统的目标。那我们如何把自己的系统改造成高可用的系统呢？带着这个问题，本文就给大家演示下，如何自己动手，从零开始基于raft协议来改造我们的已有系统。很多同学都知道Raft协议是一种分布式一致性算法。从用户的角度出发，它提供给程序设计人员的功能主要有以下2个方面

1. 它提供了一个全局一致的存储状态，这样我们的程序就可以通过它在多个节点上存储信息
2. 它提供了容错的功能，当leader不可用后，系统自动开始选举新的leader.而且每个节点知道自己的身份是follower还是leader.这样我们就可以利用这个功能实现读写分离。

当然很多同学讲到，我们可以直接部署高可用的分布式键值存储系统etcd,它本身具有高可用、高并发、一致性等特点，已经被广泛应用于云计算、微服务、容器等领域了，是很多云原生系统的底层基石之一。但是大家很快发现这样做又增加了我们系统的依赖，所以本文给出的解决方案是直接采用etcdserver包（go.etcd.io/etcd/server/v3/etcdserver是etcd的Go语言实现，提供了etcd服务器的主要功能，包括集群管理、数据存储、数据同步等）本文接下来的内容主要分为2部分，首先介绍下etcdserver的使用，然后以一个例子阐述下如何引用etcdserver包来实现高可用的系统的构建，这种构建方法不依赖于外部第三方的组件，所以它的分发与部署是比较轻便与简单的。由于内容比较多，所以就暂定分两期来介绍。

我们通过go.etcd.io/etcd/server/v3/embed这个包来快速启动集成的etcdserver。

package main

import (
 _ "context"
 "go.etcd.io/etcd/server/v3/embed"
 "log"
)

func main() {
 cfg := embed.NewConfig()
 cfg.Dir = "/Users/dongluyang1/Documents/workspace/toutiao/etcdserversample" //etcd 数据存储的目录，用于持久化存储 etcd 数据。
 cfg.WalDir = ""
 cfg.Name = "test" //节点名称
 cfg.InitialCluster = "test=http://localhost:2380" //集群名称
 cfg.ClusterState = embed.ClusterStateFlagNew //etcd 集群的初始状态，可以是 new 或 existing。当设置为 new 时，将启动一个新的 etcd 集群；当设置为 existing 时，将加入一个已经存在的 etcd 集群。
 cfg.AutoCompactionMode = "periodic"
 cfg.AutoCompactionRetention = "1"
 cfg.QuotaBackendBytes = 8 * 1024 * 1024 * 1024

 e, err := embed.StartEtcd(cfg)
 if err != nil {
 log.Fatalf("Failed to start etcd: %v", err)
 }
 defer e.Close()
 
 select {} //阻止主程序退出，导致etcdserver退出
}

上面的cfg参数通过StartEtcd方法传入embed，如下所示，实际上它的值最终传给config.ServerConfig来实现对etcdserver的配置。

embed.NewConfig的值传给了config.ServerConfig来控制etcdserver的配置

我们上面的代码简单的给出了常用的配置，下面具体给出配置的含义

go.etcd.io/etcd/server/v3/config 包中的 serverConfig 结构体包含了 etcd 服务器的配置信息，以下是该结构体中各个参数的含义：

• Name： etcd 集群中当前节点的名称，可以是任何字符串，建议为集群中唯一的名称。
• DataDir：etcd 数据存储的目录，用于持久化存储 etcd 数据。
• ListenClientUrls： etcd 服务器监听客户端请求的 URL 地址，支持多个 URL，以逗号分隔。例如：http://localhost:2379,http://localhost:4001。不填默认2379
• ListenPeerUrls： etcd 服务器监听节点之间通信的 URL 地址，支持多个 URL，以逗号分隔。例如：http://localhost:2380,http://localhost:7001。不填默认2380
• InitialCluster： etcd 集群中所有节点的信息，以 name=URL 的形式表示，各节点信息之间以逗号分隔。例如：node1=http://localhost:2380,node2=http://localhost:7001。
• InitialClusterState： etcd 集群的初始状态，可以是 new 或 existing。当设置为 new 时，将启动一个新的 etcd 集群；当设置为 existing 时，将加入一个已经存在的 etcd 集群。
• InitialClusterToken： etcd 集群的唯一标识符，用于区分不同的 etcd 集群。当启动一个新的 etcd 集群时，需要指定一个唯一的标识符。
• AutoCompactionRetention： etcd 自动压缩功能的保留时间，以天为单位。当 etcd 启用自动压缩功能时，将保留指定天数内的数据，过期数据将被删除。
• AutoCompactionMode： etcd 自动压缩功能的模式，可以是 periodic 或 revision。当设置为 periodic 时，将按时间间隔压缩数据；当设置为 revision 时，将按事务数压缩数据。

cli, err := clientv3.New(clientv3.Config{
 Endpoints: []string{"localhost:2379"},
 DialTimeout: 5 * time.Second,
})
go func() {
 if err == nil {
 for {
 _, err = cli.Put(context.Background(), "yandaojiumozhi", fmt.Sprintf("mibao-%d", rand.Intn(100)))
 if err != nil {
 // handle error
 } else {
 resp, err := cli.Get(context.Background(), "yandaojiumozhi")
 if err != nil {
 // handle error
 }
 for _, ev := range resp.Kvs {
 fmt.Printf("%s : %s\n", ev.Key, ev.Value)
 }
 }
 time.Sleep(5 * time.Second)
 }
 }
}()
defer cli.Close()

上面的代码简单测试了，通过localhost:2379随机写入yandaojiumozhi，然后读取这个key。

演示结果

上面的代码使得我们不需要额外部署与维护第三方etcd组件，便可以再启动我们后台程序的同时通过embed启动etcdserver来实现存储了。

go.etcd.io/etcd/server/v3/etcdserver 包是 etcd 服务器的核心包，它包含了 etcd 服务器的所有核心逻辑。其中 EtcdServer 结构体是 etcd 服务器的核心，它负责管理 etcd 服务器的所有组件、监听客户端请求、处理事务和维护 etcd 数据库等核心任务。这个包主要负责 etcd 服务器的启动、停止和管理。而go.etcd.io/etcd/server/v3/embed 包则负责将go.etcd.io/etcd/server/v3/etcdserver 封装到一个可嵌入的包中，使得在应用程序中嵌入 etcd 服务器变得更加容易。所以搞明白embed如何启动的etcdserver，我们便可以直接在我们的代码里面启动etcdserver,这样便可以有更大的灵活性来做一些功能。比如判断是否是leader等。

embed启动etcdserver的流程图如下所示，它的核心是在2379，2380启动监听器，然后配置config.ServerConfig,以及调用NewServer,最后Start它。

embed创建并启动etcdserver的流程

所以我们自己也可以写一个embed,来创建并启动etcdserver,然后通过下面的方法来判断是不是leader。通过isLeader的判断，来完成分布式环境下面的，不同角色自己该干的事情。

本期就先介绍这些，下一期给大家演示下分布式环境下的，节点加入，选举等相关操作。

func check(srv *etcdserver.EtcdServer, ctx context.Context) {
 log.Info("start check LeaderChanged")
 for {
 select {
 case <-ctx.Done():
 log.Info("has Done")
 return
 case <-srv.LeaderChangedNotify():
 {
 log.Info("Leader changed")
 /*这个isLeader可以判断当前节点是不是leader,如果是leader的话，可以做一些
 leader可以做的业务，比如它可以接受写请求，其他的收到写请求，都转发leader
 */
 isLeader := srv.Leader() == srv.ID() 
 ......
 }

 }
 }
}