etcd 有如下的使用场景：

  • 服务发现（Service Discovery）
  • 消息发布与订阅
  • 负载均衡
  • 分布式通知与协调
  • 分布式锁
  • 分布式队列
  • 集群监控于Leader竞选。

一、服务发现

image.png
etcd 的常见使用场景之一就是服务发现。实现思路如下：
先准备 etcd 服务端，服务端的程序在第一次启动之后会连接到 etcd 服务器并设置一个格式为 ip:port 的键值对，并绑定一个 lease。之后的服务端内部维护一个定时器，每隔一段时间就更新服务端注册中心的 lease 的 TTL。
另外一个组件就是服务发现组件，discovery 会 watch 服务端的 key。每次该 key 变化时，discovery 就可以检测到时间并做出对应的操作。
代码的实现如下：

  1. // server.go
  2. package main
  4. import (
  5.     "context"
  6.     "crypto/md5"
  7.     "encoding/json"
  8.     "errors"
  9.     "flag"
  10.     "fmt"
  11.     "github.com/coreos/etcd/clientv3"
  12.     "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
  13.     "log"
  14.     "net"
  15.     "os"
  16.     "os/signal"
  17.     "strings"
  18.     "syscall"
  19.     "time"
  20. )
  22. var (
  23.     prefix     = "register"
  24.     client     *clientv3.Client
  25.     stopSignal = make(chan struct{}, 1)
  26.     srvKey     string
  27. )
  29. var (
  30.     serv     = flag.String("name", "hello", "service name")
  31.     port     = flag.Int("port", 30000, "service port")
  32.     endpoint = flag.String("endpoints", "http://127.0.0.1:2379", "etcd endpoints")
  33. )
  35. type SvConfig struct {
  36.     Name string `json:"name"`
  37.     Host string `json:"host"`
  38.     Port int    `json:"port"`
  39. }
  41. func Register(endpoints string, config *SvConfig, interval time.Duration, ttl int) error {
  42.     // 解析服务端的值
  43.     srvValue, _ := json.Marshal(config)
  44.     srvKey = fmt.Sprintf("%s/%x", prefix, md5.Sum(srvValue))
  46.     var err error
  47.     client, err = clientv3.New(clientv3.Config{
  48.         Endpoints:   strings.Split(endpoints, ","),
  49.         DialTimeout: time.Second * 2,
  50.     })
  51.     if err != nil {
  52.         return fmt.Errorf("register service failed: %v", err)
  53.     }
  55.     go func() {
  56.         timer := time.NewTicker(interval)
  57.         for {
  59.             resp, _ := client.Grant(context.TODO(), int64(ttl))
  61.             _, err = client.Get(context.TODO(), srvKey)
  62.             if err != nil {
  63.                 // 捕获 key 不存在的场合
  64.                 if errors.Is(err, rpctypes.ErrKeyNotFound) {
  65.                     _, err = client.Put(context.TODO(), srvKey, string(srvValue), clientv3.WithLease(resp.ID))
  66.                     if err != nil {
  67.                         log.Printf("register service %s at %s:%d\n", config.Name, config.Host, config.Port)
  68.                     }
  69.                 }
  70.             } else {
  71.                 // 如果key存在就更新ttl
  72.                 _, err = client.Put(context.TODO(), srvKey, string(srvValue), clientv3.WithLease(resp.ID))
  73.             }
  74.             select {
  75.             case <-stopSignal:
  76.                 return
  77.             case <-timer.C:
  78.             }
  79.         }
  80.     }()
  82.     return err
  83. }
  85. func Unregister() error {
  86.     stopSignal <- struct{}{}
  87.     stopSignal = make(chan struct{}, 1)
  88.     _, err := client.Delete(context.TODO(), srvKey)
  89.     return err
  90. }
  92. func main() {
  93.     flag.Parse()
  95.     // 绑定服务地址和端口
  96.     lis, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", *port))
  97.     if err != nil {
  98.         panic(err)
  99.     }
  101.     config := &SvConfig{
  102.         Name: *serv,
  103.         Host: "127.0.0.1",
  104.         Port: *port,
  105.     }
  106.     Register(*endpoint, config, time.Second*10, 15)
  108.     ch := make(chan os.Signal, 1)
  109.     signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT)
  110.     go func() {
  111.         <-ch
  112.         Unregister()
  113.         os.Exit(1)
  114.     }()
  116.     log.Printf("service %s start at %d", *serv, *port)
  117.     // server todo
  118.     for {
  119.         lis.Accept()
  120.     }
  121. }
  1. // discovery.go
  2. package main
  4. import (
  5.     "context"
  6.     "encoding/json"
  7.     "flag"
  8.     "fmt"
  9.     "github.com/coreos/etcd/clientv3"
  10.     "log"
  11.     "net"
  12.     "os"
  13.     "os/signal"
  14.     "strings"
  15.     "syscall"
  16.     "time"
  17. )
  19. var (
  20.     prefix = "register"
  21.     client *clientv3.Client
  22. )
  24. var (
  25.     port     = flag.Int("port", 30001, "service port")
  26.     endpoint = flag.String("endpoints", "http://127.0.0.1:2379", "etcd endpoints")
  27. )
  29. type SvConfig struct {
  30.     Name string `json:"name"`
  31.     Host string `json:"host"`
  32.     Port int    `json:"port"`
  33. }
  35. func watcher() error {
  36.     var err error
  37.     client, err = clientv3.New(clientv3.Config{
  38.         Endpoints:   strings.Split(*endpoint, ","),
  39.         DialTimeout: time.Second * 3,
  40.     })
  41.     if err != nil {
  42.         return fmt.Errorf("connect etcd cluster failed: %v", err.Error())
  43.     }
  45.     go func() {
  46.         resp := client.Watch(context.TODO(), prefix, clientv3.WithPrefix())
  47.         for ch := range resp {
  48.             for _, event := range ch.Events {
  49.                 switch event.Type {
  50.                 case clientv3.EventTypePut:
  51.                     if event.IsCreate() {
  52.                         srv := parseSrv(event.Kv.Value)
  53.                         log.Printf("discovery service %s at %s:%d", srv.Name, srv.Host, srv.Port)
  54.                     }
  55.                 case clientv3.EventTypeDelete:
  56.                     log.Printf("delete service %s", event.Kv.Key)
  57.                 }
  58.             }
  59.         }
  60.     }()
  62.     return err
  63. }
  65. func parseSrv(text []byte) *SvConfig {
  66.     svc := &SvConfig{}
  67.     json.Unmarshal(text, &svc)
  68.     return svc
  69. }
  71. func main() {
  72.     flag.Parse()
  74.     // 绑定服务地址和端口
  75.     lis, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", *port))
  76.     if err != nil {
  77.         panic(err)
  78.     }
  80.     ch := make(chan os.Signal, 1)
  81.     signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT)
  82.     go func() {
  83.         <-ch
  84.         os.Exit(1)
  85.     }()
  87.     watcher()
  89.     log.Printf("discovery start at %d", *port)
  90.     // server todo
  91.     for {
  92.         lis.Accept()
  93.     }
  94. }

二、消息发布与订阅

image.png
消息发布和订阅使用的场景也很多的。利用 etcd 的实现思路也很简单：只要消息的发布者向 etcd 发布一系列相同前缀的key，订阅者 watch 指定的前缀即可。
代码如下：

  1. package main
  3. import (
  4.     "context"
  5.     "flag"
  6.     "fmt"
  7.     "github.com/coreos/etcd/clientv3"
  8.     "log"
  9.     "strings"
  10.     "time"
  11. )
  13. var (
  14.     prefix = "/etcd"
  15.     client   *clientv3.Client
  16.     endponts = flag.String("endpoints", "http://127.0.0.1:2379", "etcd endpoints")
  17. )
  19. func publisher(client *clientv3.Client) {
  20.     go func() {
  21.         timer := time.NewTicker(time.Second)
  22.         for range timer.C {
  23.             now := time.Now()
  24.             key := fmt.Sprintf("%s/%d", prefix, now.Second())
  25.             value := now.String()
  26.             // 可以在这里添加 key 的 lease
  27.             // resp, _ := client.Grant(context.TODO(), 10)
  28.             // client.Put(context.TODO(), key, value, clientv3.WithLease(resp.ID))
  29.             client.Put(context.TODO(), key, value)
  30.         }
  31.     }()
  32. }
  34. func subscriber(client *clientv3.Client) {
  35.     watcher := client.Watch(context.TODO(), prefix, clientv3.WithPrefix())
  36.     for ch := range watcher {
  37.         for _, e := range ch.Events {
  38.             if e.IsCreate() {
  39.                 log.Printf("received %s => %s\n", e.Kv.Key, e.Kv.Value)
  40.             }
  41.         }
  42.     }
  43. }
  45. func main() {
  46.     flag.Parse()
  48.     client, err := clientv3.New(clientv3.Config{
  49.         Endpoints:            strings.Split(*endponts, ","),
  50.         DialTimeout:          time.Second * 2,
  51.     })
  52.     if err != nil {
  53.         log.Fatalln("connect etcd cluster failed: " + err.Error())
  54.     }
  56.     publisher(client)
  57.     subscriber(client)
  59.     select {
  60.     //
  61.     }
  62. }

三、负载均衡

image.png
etcd 可以配合 grpc 实现负载均衡的功能。可以在服务发现的基础上，利用 grpc 自带的 client 负载均衡实现。首先实现服务发现：

  1. // register.go
  2. package balance
  4. import (
  5.     "fmt"
  6.     "log"
  7.     "strings"
  8.     "time"
  10.     "context"
  11.     etcd3 "github.com/coreos/etcd/clientv3"
  12.     "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
  13. )
  15. // 服务的前缀
  16. // 用这个来区分不同项目的服务
  17. var Prefix = "etcd3_naming"
  18. var client etcd3.Client
  19. var serviceKey string
  21. var stopSignal = make(chan bool, 1)
  23. // 服务注册
  24. func Register(name string, host string, port int, target string, interval time.Duration, ttl int) error {
  25.     serviceValue := fmt.Sprintf("%s:%d", host, port)
  26.     serviceKey = fmt.Sprintf("/%s/%s/%s", Prefix, name, serviceValue)
  28.     // 解析 etcd 的 endpoints
  29.     // 开启 etcd 客户端用于注册服务
  30.     var err error
  31.     client, err := etcd3.New(etcd3.Config{
  32.         Endpoints: strings.Split(target, ","),
  33.     })
  34.     if err != nil {
  35.         return fmt.Errorf("grpclb: create etcd3 client failed: %v", err)
  36.     }
  38.     go func() {
  39.         // 启动一个定时器自动注册服务
  40.         ticker := time.NewTicker(interval)
  41.         for {
  42.             // 在 etcd 中创建一个 lease 绑定服务的地址
  43.             resp, _ := client.Grant(context.TODO(), int64(ttl))
  44.             // 检测服务地址是否存在，若不存在就创建
  45.             _, err := client.Get(context.Background(), serviceKey)
  46.             if err != nil {
  47.                 if err == rpctypes.ErrKeyNotFound {
  48.                     // 服务地址不存在
  49.                     if _, err := client.Put(context.TODO(), serviceKey, serviceValue, etcd3.WithLease(resp.ID)); err != nil {
  50.                         log.Printf("grpclb: set service '%s' with ttl to etcd3 failed: %s", name, err.Error())
  51.                     }
  52.                 } else {
  53.                     log.Printf("grpclb: service '%s' connect to etcd3 failed: %s", name, err.Error())
  54.                 }
  55.             } else {
  56.                 // 刷新服务地址 lease
  57.                 if _, err := client.Put(context.Background(), serviceKey, serviceValue, etcd3.WithLease(resp.ID)); err != nil {
  58.                     log.Printf("grpclb: refresh service '%s' with ttl to etcd3 failed: %s", name, err.Error())
  59.                 }
  60.                 //log.Panicln(serviceKey)
  61.             }
  62.             select {
  63.             case <-stopSignal:
  64.                 return
  65.             case <-ticker.C:
  66.             }
  67.         }
  68.     }()
  70.     return nil
  71. }
  73. // 删除服务注册信息
  74. func UnRegister() error {
  75.     stopSignal <- true
  76.     // 获取 chan 之后马上留空，防止多个 UnRegister 造成死锁
  77.     stopSignal = make(chan bool, 1)
  78.     var err error
  79.     if _, err := client.Delete(context.Background(), serviceKey); err != nil {
  80.         log.Printf("grpclb: deregister '%s' failed: %s", serviceKey, err.Error())
  81.     } else {
  82.         log.Printf("grpclb: deregister '%s' ok.", serviceKey)
  83.     }
  84.     return err
  85. }
  1. package balance
  3. import (
  4.     "fmt"
  6.     "context"
  7.     etcd3 "github.com/coreos/etcd/clientv3"
  8.     "github.com/coreos/etcd/mvcc/mvccpb"
  9.     "google.golang.org/grpc/naming"
  10. )
  12. // watcher is the implementaion of grpc.naming.Watcher
  13. type watcher struct {
  14.     re            *resolver // re: Etcd Resolver
  15.     client        etcd3.Client
  16.     isInitialized bool
  17. }
  19. // Close do nothing
  20. func (w *watcher) Close() {
  21. }
  23. // Next to return the updates
  24. func (w *watcher) Next() ([]*naming.Update, error) {
  25.     // prefix is the etcd prefix/value to watch
  26.     prefix := fmt.Sprintf("/%s/%s/", Prefix, w.re.serviceName)
  27.     fmt.Println("prefix", prefix)
  28.     // check if is initialized
  29.     if !w.isInitialized {
  30.         // query addresses from etcd
  31.         w.isInitialized = true
  32.         resp, err := w.client.Get(context.Background(), prefix, etcd3.WithPrefix())
  33.         if err == nil {
  34.             addrs := extractAddrs(resp)
  35.             //if not empty, return the updates or watcher new dir
  36.             if l := len(addrs); l != 0 {
  37.                 updates := make([]*naming.Update, l)
  38.                 for i := range addrs {
  39.                     updates[i] = &naming.Update{Op: naming.Add, Addr: addrs[i]}
  40.                 }
  41.                 return updates, nil
  42.             }
  43.         }
  44.     }
  46.     // generate etcd Watcher
  47.     rch := w.client.Watch(context.Background(), prefix, etcd3.WithPrefix())
  48.     for wresp := range rch {
  49.         for _, ev := range wresp.Events {
  50.             switch ev.Type {
  51.             case mvccpb.PUT:
  52.                 return []*naming.Update{{Op: naming.Add, Addr: string(ev.Kv.Value)}}, nil
  53.             case mvccpb.DELETE:
  54.                 return []*naming.Update{{Op: naming.Delete, Addr: string(ev.Kv.Value)}}, nil
  55.             }
  56.         }
  57.     }
  58.     return nil, nil
  59. }
  61. func extractAddrs(resp *etcd3.GetResponse) []string {
  62.     addrs := []string{}
  64.     if resp == nil || resp.Kvs == nil {
  65.         return addrs
  66.     }
  68.     for i := range resp.Kvs {
  69.         if v := resp.Kvs[i].Value; v != nil {
  70.             addrs = append(addrs, string(v))
  71.         }
  72.     }
  74.     return addrs
  75. }
  1. // resolver.go
  2. package balance
  4. import (
  5.     "errors"
  6.     "fmt"
  7.     "strings"
  9.     etcd3 "github.com/coreos/etcd/clientv3"
  10.     "google.golang.org/grpc/naming"
  11. )
  13. // resolver is the implementaion of grpc.naming.Resolver
  14. type resolver struct {
  15.     serviceName string // service name to resolve
  16. }
  18. // NewResolver return resolver with service name
  19. func NewResolver(serviceName string) *resolver {
  20.     return &resolver{serviceName: serviceName}
  21. }
  23. // Resolve to resolve the service from etcd, target is the dial address of etcd
  24. // target example: "http://127.0.0.1:2379,http://127.0.0.1:12379,http://127.0.0.1:22379"
  25. func (re *resolver) Resolve(target string) (naming.Watcher, error) {
  26.     if re.serviceName == "" {
  27.         return nil, errors.New("grpclb: no service name provided")
  28.     }
  30.     // generate etcd client
  31.     client, err := etcd3.New(etcd3.Config{
  32.         Endpoints: strings.Split(target, ","),
  33.     })
  34.     if err != nil {
  35.         return nil, fmt.Errorf("grpclb: creat etcd3 client failed: %s", err.Error())
  36.     }
  38.     // Return watcher
  39.     return &watcher{re: re, client: *client}, nil
  40. }

实现服务发现和服务的解析之后，使用protobuf来定义服务的内容：

  1. syntax = "proto3";
  3. option java_multiple_files = true;
  4. option java_package = "com.midea.jr.test.grpc";
  5. option java_outer_classname = "HelloWorldProto";
  6. option objc_class_prefix = "HLW";
  8. package pb;
  10. // The greeting service definition.
  11. service Greeter {
  12.     //   Sends a greeting
  13.     rpc SayHello (HelloRequest) returns (HelloReply) {
  14.     }
  15. }
  17. // The request message containing the user's name.
  18. message HelloRequest {
  19.     string name = 1;
  20. }
  22. // The response message containing the greetings
  23. message HelloReply {
  24.     string message = 1;
  25. }

将proto文件编译成go代码：

  1. # 需要先安装 protoc-gen-go  和 proto
  2. # go get -u github.com/golang/protobuf/protoc-gen-go 
  3. # go get -u github.com/golang/protobuf/proto
  4. $ protoc -I ./pb --go_out=plugins=grpc:pb ./pb/helloworld.proto

服务端的代码如下：

  1. // server.go
  2. package main
  4. import (
  5.     "flag"
  6.     "fmt"
  7.     "log"
  8.     "net"
  9.     "os"
  10.     "os/signal"
  11.     "syscall"
  12.     "time"
  14.     "context"
  15.     "google.golang.org/grpc"
  16.     grpclb "xingyys.com/mysite/balance"
  17.     "xingyys.com/mysite/pb"
  18. )
  20. var (
  21.     serv = flag.String("service", "hello_service", "service name")
  22.     port = flag.Int("port", 50001, "listening port")
  23.     reg  = flag.String("reg", "http://127.0.0.1:2379", "register etcd address")
  24. )
  26. func main() {
  27.     flag.Parse()
  29.     // 绑定服务端的监听地址
  30.     lis, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%d", *port))
  31.     if err != nil {
  32.         panic(err)
  33.     }
  35.     // 向 etcd 注册服务，指定检测周期为10s，生存周期为15s
  36.     err = grpclb.Register(*serv, "127.0.0.1", *port, *reg, time.Second*10, 15)
  37.     if err != nil {
  38.         panic(err)
  39.     }
  41.     ch := make(chan os.Signal, 1)
  42.     signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT)
  43.     go func() {
  44.         s := <-ch
  45.         log.Printf("receive signal '%v'", s)
  46.         grpclb.UnRegister()
  47.         os.Exit(1)
  48.     }()
  50.     // 启动服务端
  51.     log.Printf("starting hello service at %d", *port)
  52.     s := grpc.NewServer()
  53.     pb.RegisterGreeterServer(s, &server{})
  54.     s.Serve(lis)
  55. }
  57. // server is used to implement helloworld.GreeterServer.
  58. type server struct{}
  60. // SayHello implements helloworld.GreeterServer
  61. func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
  62.     fmt.Printf("%v: Receive is %s\n", time.Now(), in.Name)
  63.     return &pb.HelloReply{Message: "Hello " + in.Name}, nil
  64. }

负载均衡的代码下client中实现：

  1. // client.go
  2. package main
  4. import (
  5.     "flag"
  6.     "fmt"
  7.     "time"
  9.     "strconv"
  11.     "context"
  12.     "google.golang.org/grpc"
  13.     grpclb "xingyys.com/mysite/balance"
  14.     "xingyys.com/mysite/pb"
  15. )
  17. var (
  18.     serv = flag.String("service", "hello_service", "service name")
  19.     reg  = flag.String("reg", "http://127.0.0.1:2379", "register etcd address")
  20. )
  22. func main() {
  23.     flag.Parse()
  24.     fmt.Println("serv", *serv)
  25.     r := grpclb.NewResolver(*serv)
  26.     b := grpc.RoundRobin(r)
  28.     ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
  29.     conn, err := grpc.DialContext(ctx, *reg, grpc.WithInsecure(), grpc.WithBalancer(b))
  30.     if err != nil {
  31.         panic(err)
  32.     }
  33.     fmt.Println("conn...")
  35.     ticker := time.NewTicker(1 * time.Second)
  36.     for t := range ticker.C {
  37.         client := pb.NewGreeterClient(conn)
  38.         resp, err := client.SayHello(context.Background(), &pb.HelloRequest{Name: "world " + strconv.Itoa(t.Second())})
  39.         if err == nil {
  40.             fmt.Printf("%v: Reply is %s\n", t, resp.Message)
  41.         } else {
  42.             fmt.Println(err)
  43.         }
  44.     }
  45. }

四、分布式通知与协调

image.png
和消息发布与订阅相似，都是用到 etcd 的 watch 机制，通过注册与异步通知机制，实现分布式环境下不同系统之间的通知与协调，从而对数据变更做到实时处理。实现思路如下：不同的系统在 etcd 注册目录，并监控目录下 key 的变化，到检测到变化时，watcher 做出放映。

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "github.com/coreos/etcd/clientv3"
  7.     "log"
  8.     "time"
  9. )
  11. func main() {
  12.     client, err := clientv3.New(clientv3.Config{
  13.         Endpoints:   []string{"192.168.10.10:2379"},
  14.         DialTimeout: time.Second * 3,
  15.     })
  16.     if err != nil {
  17.         log.Fatalln(err)
  18.     }
  20.     prefix := "/job"
  21.     id := "test_job"
  22.     key := fmt.Sprintf("%s/%s", prefix, id)
  23.     go func() {
  24.         timer := time.NewTicker(time.Millisecond * 10)
  25.         i := 0
  26.         for range timer.C {
  27.             if i > 100 {
  28.                 return
  29.             }
  31.             if _, err := client.Put(context.TODO(), key, string(i)); err == nil {
  32.                 log.Printf("job process: %d%%", i)
  33.             }
  35.             i++
  36.         }
  37.     }()
  39.     watcher := client.Watch(context.TODO(), key)
  40.     for ch := range watcher {
  41.         for _, e := range ch.Events {
  42.             if e.Kv.Value[0] == 100 {
  43.                 log.Println("job Done!")
  44.                 return
  45.             }
  46.         }
  47.     }
  48. }

五、分布式锁

etcd的使用实例 - 图5
因为etcd使用Raft算法保持了数据的强一致性，某次操作存储到集群中的值必然是全局一致的，所以很容易实现分布式锁。实现的思路：多个 session 同时使用开启事物抢占同一 key，最先抢到的 session 获得锁，其他 session 等待锁的释放。如果是 trylock，session 在抢不到 session 时不再等待直接报错。
在 etcd clientv3的版本中，官方自带锁的实现，支持locks 和 trylock（需要 etcd v3.4.3）示例看 这里

六、分布式队列

image.png

etcd 分布式队列有两种实现方式，一种等待所有条件都满足后才开始执行任务。另一种是先入先出列队。第一种的思路就是在 watch 一个目录，当目录下存在必要的 key 时就进行对应操作。

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "github.com/coreos/etcd/clientv3"
  7.     "log"
  8.     "math/rand"
  9.     "time"
  10. )
  12. func random(max int) int {
  13.     rand.Seed(time.Now().UnixNano())
  14.     return rand.Intn(max)
  15. }
  17. func main() {
  18.     client, _ := clientv3.New(clientv3.Config{
  19.         Endpoints:   []string{"192.168.10.10:2379"},
  20.         DialTimeout: time.Second * 2,
  21.     })
  23.     prefix := "/queue"
  25.     client.Delete(context.TODO(), prefix, clientv3.WithPrefix())
  27.     // 每隔1s，condition 变为 0 1 2 中的随机一个
  28.     go func() {
  29.         timer := time.NewTicker(time.Second * 1)
  30.         key := prefix + "/1"
  31.         for range timer.C {
  32.             rd := random(3)
  33.             client.Put(context.TODO(), key, fmt.Sprintf("%d", rd))
  34.         }
  35.     }()
  37.     // 每隔2s，condition 变为 0 1 2 中的随机一个
  38.     go func() {
  39.         timer := time.NewTicker(time.Second * 1)
  40.         key := prefix + "/2"
  41.         for range timer.C {
  42.             rd := random(3)
  43.             client.Put(context.TODO(), key, fmt.Sprintf("%d", rd))
  44.         }
  45.     }()
  47.     // 每隔3s，condition 变为 0 1 2 中的随机一个
  48.     go func() {
  49.         timer := time.NewTicker(time.Second * 1)
  50.         key := prefix + "/3"
  51.         for range timer.C {
  52.             rd := random(3)
  53.             client.Put(context.TODO(), key, fmt.Sprintf("%d", rd))
  54.         }
  55.     }()
  57.     watcher := client.Watch(context.TODO(), prefix, clientv3.WithPrefix())
  58.     for range watcher {
  59.         // 满足以下条件是退出
  60.         // /queue/1 = 0
  61.         // /queue/2 = 2
  62.         // /queue/3 = 1
  63.         resp, _ := client.Get(context.TODO(), prefix, clientv3.WithRange(prefix+"/4"))
  64.         fmt.Println(resp.Kvs)
  65.         if string(resp.Kvs[0].Value[0]) == "0" &&
  66.             string(resp.Kvs[1].Value[0]) == "2" &&
  67.             string(resp.Kvs[2].Value[0]) == "1" {
  68.             log.Println("Done!")
  69.             return
  70.         }
  71.     }
  72. }

第二种实现思路：

  1. package main
  3. import (
  4.     "context"
  5.     "crypto/md5"
  6.     "fmt"
  7.     "github.com/coreos/etcd/clientv3"
  8.     "time"
  9. )
  11. func main() {
  13.     ctx, cancel := context.WithCancel(context.Background())
  14.     defer cancel()
  16.     client, _ := clientv3.New(clientv3.Config{
  17.         Endpoints:   []string{"192.168.10.10:2379"},
  18.         DialTimeout: time.Second * 2})
  20.     prefix := "/queue"
  21.     client.Delete(ctx, prefix, clientv3.WithPrefix())
  22.     for i := 0; i < 10; i++ {
  23.         key := fmt.Sprintf("%s/%x", prefix, md5.Sum([]byte(time.Now().String())))
  24.         client.Put(ctx, key, string(i))
  25.         fmt.Printf("key %s push queue\n", key)
  26.     }
  28.     fmt.Println("\n\n")
  30.     ops := make([]clientv3.OpOption, 0)
  31.     // 换成 clientv3.WithLastRev() 就是先进先出队列了
  32.     ops = append(ops, clientv3.WithFirstRev()...)
  33.     ops = append(ops, clientv3.WithPrefix())
  34.     ops = append(ops, clientv3.WithLimit(1))
  35.     for i := 0; i < 10; i++ {
  36.         resp, _ := client.Get(context.TODO(), prefix, ops...)
  37.         if resp.Count > 0 {
  38.             key := string(resp.Kvs[0].Key)
  39.             fmt.Printf("count %d => key %s pop queue\n", resp.Count, key)
  40.             client.Delete(context.TODO(), key)
  41.         }
  42.         //fmt.Println(resp.Kvs)
  43.     }
  44. }

七、集群监控与Leader竞选。

通过etcd来进行监控实现起来非常简单并且实时性强。

  1. Watcher机制，当某个节点消失或有变动时，Watcher会第一时间发现并告知用户。
  2. 节点可以设置TTL key，比如每隔30s发送一次心跳使代表该机器存活的节点继续存在，否则节点消失。

　　这样就可以第一时间检测到各节点的健康状态，以完成集群的监控要求。

使用分布式锁，可以完成Leader竞选。
　　这种场景通常是一些长时间CPU计算或者使用IO操作的机器，只需要竞选出的Leader计算或处理一次，就可以把结果复制给其他的Follower。从而避免重复劳动，节省计算资源。
这个的经典场景是搜索系统中建立全量索引。如果每个机器都进行一遍索引的建立，不但耗时而且建立索引的一致性不能保证。通过在etcd的CAS机制同时创建一个节点，创建成功的机器作为Leader，进行索引计算，然后把计算结果分发到其它节点。

etcd的使用实例 - 图7

同样官方自带示例：详细看 这里