背景

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

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

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


(资料图片)

通过embed启动etcdserver

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

package mainimport (   _ "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。不填默认2379ListenPeerUrls： etcd 服务器监听节点之间通信的 URL 地址，支持多个 URL，以逗号分隔。例如：http://localhost:2380,http://localhost:7001。不填默认2380InitialCluster： 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 时，将按事务数压缩数据。客户端访问etcdserver
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来实现存储了。

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()            ......         }      }   }}

推荐内容