GO 中 ETCD 的编码案例分享

咱们来回顾一下上次咱们说到的服务注册和发现

分享了服务注册和发现是什么
CAP 定理是什么
ETCD 是什么，以及ETCD 和 Zookeeper的比照
ETCD 的分布式锁实现的简略原理

要是对服务注册与发现，ETCD 还有点趣味的话，欢送查看文章 服务注册与发现之ETCD

明天咱们来看看 GO 如何去操作 ETCD ，这个开源的、高可用的分布式key-value存储系统

感兴趣的小伙伴能够看看GO 的 ETCD 官网文档

https://pkg.go.dev/go.etcd.io...

依据官网文档，咱们本次分享几个点

ETCD 如何装置
ETCD 外面对于 KEY 的PUT 和GET操作
WATCH操作
Lease 租约
KeepAlive 保活
ETCD 分布式锁的实现

ETCD 如何装置

ETCD 的装置和部署

这里咱们就做一个简略的单机部署

到 github 上下载最新的 etcd 包，https://github.com/etcd-io/et...
解压后，将 etcd 和 etcdctl 拷贝到咱们的 $GOBIN目录下，或者退出咱们零碎的环境变量即可（目标是能够间接键入etcd 零碎可能运行该可执行文件）
能够应用 etcd --version 查看版本

对于 ETCD 的命令就不在此做过的分享了，明天次要是分享 GO 如何应用 ETCD

包的装置

本次咱们应用的是 ETCD 的 clientv3 包，咱们执行如下命令即可正确装置 ETCD

 go get go.etcd.io/etcd/clientv3

无论你是间接执行下面的命令，还是通过 go mod 的形式，去下载 ETCD 的 clientv3包，可能会呈现如下问题：

/root/go/pkg/mod/github.com/coreos/etcd@v3.3.25+incompatible/clientv3/balancer/picker/roundrobin_balanced.go:55:54: undefined: balancer.PickOptions# github.com/coreos/etcd/clientv3/balancer/resolver/endpoint/root/go/pkg/mod/github.com/coreos/etcd@v3.3.25+incompatible/clientv3/balancer/resolver/endpoint/endpoint.go:114:78: undefined: resolver.BuildOption/root/go/pkg/mod/github.com/coreos/etcd@v3.3.25+incompatible/clientv3/balancer/resolver/endpoint/endpoint.go:182:31: undefined: resolver.ResolveNowOption

如上问题，是因为包抵触了，咱们只须要将如下替换包的命令放到咱们 go.mod 上面即可

replace google.golang.org/grpc => google.golang.org/grpc v1.26.0

例如我的 go.mod 是这样的

module my_etcdgo 1.15require (        github.com/coreos/etcd v3.3.25+incompatible // indirect        github.com/coreos/go-semver v0.3.0 // indirect        github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf // indirect        github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f // indirect        github.com/gogo/protobuf v1.3.2 // indirect        github.com/google/uuid v1.2.0 // indirect        go.etcd.io/etcd v3.3.25+incompatible        go.uber.org/zap v1.17.0 // indirect        google.golang.org/grpc v1.38.0 // indirect)replace google.golang.org/grpc => google.golang.org/grpc v1.26.0

这里顺便插一句， go mod进行包治理的形式从 GO 1.14之后就开始有了，go mod治理包十分不便，这里简略分享一下如何应用

在和 main.go 的同级目录下，初始化一个go mod，执行如下命令

go mod init xxx

写好咱们的代码在 main.go 文件中，即可在 main.go 的同级目录下执行 go build 进行编译 go程序
若编译呈现上述问题，那么就能够在生成的go.mod 文件中退出上述替换包的语句即可

包装置好了，咱们能够开始进行编码了

ETCD 的设置 KEY 和获取 KEY 操作

ETCD 的默认端口是这样的：

2379 端口

提供 HTTP API 服务

2380 端口

用来与 peer 通信

咱们开始写一个 GET 和 PUT KEY 的DEMO

package mainimport (   "context"   "log"   "time"   "go.etcd.io/etcd/clientv3")func main() {   // 设置 log 参数 ，打印以后工夫 和 以后行数   log.SetFlags(log.Ltime | log.Llongfile)    // ETCD 默认端口号是 2379   // 应用 ETCD 的 clientv3 包   client, err := clientv3.New(clientv3.Config{      Endpoints:   []string{"127.0.0.1:2379"},      //超时工夫 10 秒      DialTimeout: 10 * time.Second,   })   if err != nil {      log.Printf("connect to etcd error : %v\n", err)      return   }   log.Printf("connect to etcd successfully ...")   // defer 最初敞开 连贯   defer client.Close()   // PUT KEY 为 name , value 为 xiaomotong   ctx, cancel := context.WithTimeout(context.Background(), time.Second)   _, err = client.Put(ctx, "name", "xiaomotong")   cancel()   if err != nil {      log.Printf("PUT key to etcd error : %v\n", err)      return   }   // 获取ETCD 的KEY   ctx, cancel = context.WithTimeout(context.Background(), time.Second)   resp, err := client.Get(ctx, "name")   cancel()   if err != nil {      log.Printf("GET key-value from etcd error : %v\n", err)      return   }       // 遍历读出 KEY 和对应的 value   for _, ev := range resp.Kvs {      log.Printf("%s : %s\n", ev.Key, ev.Value)   }}

感兴趣的小伙伴能够将上述代码拷贝到你的环境中进行运行，即可看到你想要的答案

ETCD 的 WATCH操作

WATCH操作就是拍一个哨兵监控某一个key对应值的变动，包含新增，删除，批改

func main() {   // 设置 log 参数 ，打印以后工夫 和 以后行数   log.SetFlags(log.Ltime | log.Llongfile)    // ETCD 默认端口号是 2379   // 应用 ETCD 的 clientv3 包   client, err := clientv3.New(clientv3.Config{      Endpoints:   []string{"127.0.0.1:2379"},      DialTimeout: 10 * time.Second,   })   if err != nil {      log.Printf("connect to etcd error : %v\n", err)      return   }   log.Printf("connect to etcd successfully ...")   defer client.Close()   // 派一个哨兵 始终监督 name 的变动   // respCh 是一个通道   respCh := client.Watch(context.Background(), "name")   // 若 respCh 为空，会阻塞在这里   for watchResp := range respCh {      for _, v := range watchResp.Events {         log.Printf("type =  %s , Key = %s , Value = %s\n",             v.Type, v.Kv.Key, v.Kv.Value)      }   }}

上述代码因为 respCh是一个通道，若外面没有数据的话，上面的 for 循环，会阻塞的等，因而须要咱们本人在终端下面模仿新增，删除，批改 name 对应的值，那么，咱们的程序就会做出对应的相应

例如，我在终端命令中敲入：etcdctl --endpoints=http://127.0.0.1:2379 put name "xiaomotong"

那么，咱们上述代码运行的程序就会输入如下语句

./my_etcd22:18:39 /home/xiaomotong/my_etcd/main.go:23: connect to etcd successfully ...22:18:43 /home/xiaomotong/my_etcd/main.go:31:type =  PUT , Key = name , Value = xiaomotong

ETCD 的 LEASE 操作

LEASE ，租约，就是将本人的某一个 key 设置一个无效工夫 / 过期工夫，相似于 REDIS 外面的 SETNX

func main() {   // 设置 log 参数 ，打印以后工夫 和 以后行数   log.SetFlags(log.Ltime | log.Llongfile)   // ETCD 默认端口号是 2379   // 应用 ETCD 的 clientv3 包   client, err := clientv3.New(clientv3.Config{      Endpoints:   []string{"127.0.0.1:2379"},      DialTimeout: 10 * time.Second,   })   if err != nil {      log.Printf("connect to etcd error : %v\n", err)      return   }   log.Printf("connect to etcd successfully ...")   defer client.Close()   // 咱们创立一个 20秒钟的租约   resp, err := client.Grant(context.TODO(), 20)   if err != nil {      log.Printf("client.Grant error : %v\n", err)      return   }   // 20秒钟之后, /name 这个key就会被移除   _, err = client.Put(context.TODO(), "/name", "xiaomotong", clientv3.WithLease(resp.ID))   if err != nil {      log.Printf("client.Put error : %v\n", err)      return   }}

上述 name , 20 秒钟之后就会主动生效

ETCD 的保活操作

顺便说一下，keepalived 也是一个开源的组件，用作高可用，感兴趣的能够深刻理解一下

此处的 keepalived 是保活，这里是 ETCD 的保活，能够在上述代码中做一个调整，上述的 name ，不生效

func main() {   // 设置 log 参数 ，打印以后工夫 和 以后行数   log.SetFlags(log.Ltime | log.Llongfile)   // ETCD 默认端口号是 2379   // 应用 ETCD 的 clientv3 包   client, err := clientv3.New(clientv3.Config{      Endpoints:   []string{"127.0.0.1:2379"},      DialTimeout: 10 * time.Second,   })   if err != nil {      log.Printf("connect to etcd error : %v\n", err)      return   }   log.Printf("connect to etcd successfully ...")   defer client.Close()   // 咱们创立一个 20秒钟的租约   resp, err := client.Grant(context.TODO(), 20)   if err != nil {      log.Printf("client.Grant error : %v\n", err)      return   }   // 20秒钟之后, /name 这个key就会被移除   _, err = client.Put(context.TODO(), "/name", "xiaomotong", clientv3.WithLease(resp.ID))   if err != nil {      log.Printf("client.Put error : %v\n", err)      return   }   // 这个key  name ，将永恒被保留   ch, kaerr := client.KeepAlive(context.TODO(), resp.ID)   if kaerr != nil {      log.Fatal(kaerr)   }   for {      ka := <-ch      log.Println("ttl:", ka.TTL)   }}

咱能够看看 keepalived 的官网阐明 ,

KeepAlive 使给定的租约永远存活。如果发送到通道的 keepalive 响应没有立刻被应用，租期客户端将至多每秒钟持续向 etcd 服务器发送keepalive 申请，直到应用最新的响应。

// KeepAlive keeps the given lease alive forever. If the keepalive response// posted to the channel is not consumed immediately, the lease client will// continue sending keep alive requests to the etcd server at least every// second until latest response is consumed.//// The returned "LeaseKeepAliveResponse" channel closes if underlying keep// alive stream is interrupted in some way the client cannot handle itself;// given context "ctx" is canceled or timed out. "LeaseKeepAliveResponse"// from this closed channel is nil.//// If client keep alive loop halts with an unexpected error (e.g. "etcdserver:// no leader") or canceled by the caller (e.g. context.Canceled), the error// is returned. Otherwise, it retries.//// TODO(v4.0): post errors to last keep alive message before closing// (see https://github.com/coreos/etcd/pull/7866)KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error)

来看看 ETCD 的分布式锁实现

这里须要引入一个新的包，"github.com/coreos/etcd/clientv3/concurrency"

不过应用go mod 治理形式的小伙伴就不必操心了，写完代码，间接 go build，GO 工具会间接帮咱们下载相干包，并编译好

Go 这一点真的相当不戳

package mainimport (    "context"    "github.com/coreos/etcd/clientv3"    "github.com/coreos/etcd/clientv3/concurrency"    "log")func main (){   // 设置 log 参数 ，打印以后工夫 和 以后行数   log.SetFlags(log.Ltime | log.Llongfile)   // ETCD 默认端口号是 2379   // 应用 ETCD 的 clientv3 包   // Endpoints 需填入 url 列表   client, err := clientv3.New(clientv3.Config{Endpoints: []string{"/name"}})   if err != nil {      log.Printf("connect to etcd error : %v\n", err)      return   }   defer client.Close()   // 创立第一个 会话   session1, err := concurrency.NewSession(client)   if err != nil {      log.Printf("concurrency.NewSession 1 error : %v\n", err)      return   }   defer session1.Close()   // 设置锁   myMu1 := concurrency.NewMutex(session1, "/lock")   // 创立第二个 会话   session2, err := concurrency.NewSession(client)   if err != nil {      log.Printf("concurrency.NewSession 2 error : %v\n", err)      return   }   defer session2.Close()   // 设置锁   myMu2 := concurrency.NewMutex(session2, "/lock")   // 会话s1获取锁   if err := myMu1.Lock(context.TODO()); err != nil {      log.Printf("myMu1.Lock error : %v\n", err)      return   }   log.Println("Get session1 lock ")   m2Chan := make(chan struct{})   go func() {      defer close(m2Chan)      // 如果加锁不胜利会阻塞，晓得加锁胜利为止      // 这里是应用一个通道的形式来通信      // 当 myMu2 能加锁胜利，阐明myMu1 解锁胜利      // 当 myMu2 加锁胜利的时候，会敞开 通道      // 敞开通道，从通道中读出来的就是nil      if err := myMu2.Lock(context.TODO()); err != nil {         log.Printf("myMu2.Lock error : %v\n", err)         return      }   }()   // 解锁   if err := myMu1.Unlock(context.TODO()); err != nil {      log.Printf("myMu1.Unlock error : %v\n", err)      return   }   log.Println("Release session1 lock ")   // 读取到nil   <-m2Chan   log.Println("Get session2 lock")}

在上述代码中，咱们创立 2 个会话来模仿分布式锁

咱们先让第 1 个会话拿到锁，并且第 2 个会话会去尝试加锁

当第 2个会话，正确加锁胜利的时候，会敞开一个通道，来确认本人真的加到锁了

上述第 2 个会话加锁的逻辑如下：

如果加锁不胜利会阻塞，晓得加锁胜利为止
这里是应用一个通道的形式来通信
当 myMu2 能加锁胜利，阐明 myMu1 解锁胜利
当 myMu2 加锁胜利的时候，会敞开 m2Chan 通道
敞开通道，从 m2Chan通道中读出来的就是nil ，确认会话 2 加锁胜利

总结

分享了ETCD的简略单点部署，ETCD 应用到的包装置，以及会遇到的问题
ETCD 的设置和获取KEY
ETCD 的WATCH 监控 KEY的简化
ETCD 的租约和保活机制
ETCD 的分布式锁的简略实现

如上的编码案例，大家能够拿下来本人运行看看成果，一起学习，一起提高

若想更多的深刻理解和学习，能够看文章最开始说到的官网文档，官网文档中的案例更加详尽

具体的源码也是十分具体的，就怕你学不会

欢送点赞，关注，珍藏

敌人们，你的反对和激励，是我保持分享，提高质量的能源

好了，本次就到这里，下一次分享GO 中 string 的实现原理

技术是凋谢的，咱们的心态，更应是凋谢的。拥抱变动，背阴而生，致力向前行。

我是小魔童哪吒，欢送点赞关注珍藏，下次见~