关于 etcd 的安装和介绍看 这里 。
官方的实例可以看 这里

一、连接

首先是关于 golang 如何连接 etcd ，先是简单的连接。

  1. package main
  3. import (
  4.     "github.com/coreos/etcd/clientv3"
  5.     "log"
  6.     "time"
  7. )
  9. func connect()  {
  10.     cli, err := clientv3.New(clientv3.Config{
  11.         // etcd 集群的地址集合
  12.         Endpoints:            []string{"192.168.10.10:2379"},
  13.         // 请求超时时间
  14.         DialTimeout:          time.Second * 3,
  15.     })
  16.     if err != nil {
  17.         log.Fatal("connect etcd cluster: " + err.Error())
  18.     }
  19.     cli.Close()
  20. }

还有带 https 和 开启用户验证的连接

  1. func connectTlsAuth() {
  3.     tlsInfo := transport.TLSInfo{
  4.         CertFile:      "/tmp/cert.pem",
  5.         KeyFile:       "/tmp/key.pem",
  6.         TrustedCAFile: "/tmp/ca.pem",
  7.     }
  8.     tlsConfig, err := tlsInfo.ClientConfig()
  9.     if err != nil {
  10.         log.Fatal("parse tls config file: " + err.Error())
  11.     }
  13.     cli, err := clientv3.New(clientv3.Config{
  14.         Endpoints:   []string{"192.168.10.10:2379"},
  15.         DialTimeout: time.Second * 3,
  16.         TLS:         tlsConfig,
  17.         Username:    "root",
  18.         Password:    "root",
  19.     })
  21.     if err != nil {
  22.         log.Fatal("connect etcd cluster: " + err.Error())
  23.     }
  24.     cli.Close()
  25. }

二、KV 操作

2.1 简单的 curd

在连接基础上，接下来就可以对key做操作了。对key做 curd

  1. func kv() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     // etcdctl put foo 1
  14.     _, err := cli.Put(ctx, "foo", "1")
  15.     if err != nil {
  16.         log.Fatal("put key:" + err.Error())
  17.     }
  19.     // etcdctl get foo --prefix
  20.     // 带参数的请求
  21.     resp, err := cli.Get(ctx, "foo", clientv3.WithPrefix())
  22.     if err != nil {
  23.         log.Fatal("get key: " + err.Error())
  24.     }
  25.     for _, v := range resp.Kvs {
  26.         log.Printf("get %s => %s\n", v.Key, string(v.Value))
  27.     }
  29.     kvcli := clientv3.NewKV(cli)
  30.     // etcdctl del foo
  31.     _, err = kvcli.Delete(ctx, "foo")
  32.     if err != nil {
  33.         log.Fatal("delete key: " + err.Error())
  34.     }
  35. }

2.2 事务

使用事务如下：

  1. func txn() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     kvc := clientv3.NewKV(cli)
  15.     _, err := kvc.Put(ctx, "foo", "xyz")
  16.     if err != nil {
  17.         log.Fatal("put key: " + err.Error())
  18.     }
  20.     _, err = kvc.Txn(ctx).
  21.         // txn value comparisons are lexical
  22.         If(clientv3.Compare(clientv3.Value("foo"), ">", "abc")).
  23.         // the "Then" runs, since "xyz" > "abc"
  24.         Then(clientv3.OpPut("foo", "XYZ")).
  25.         // the "Else" does not run
  26.         Else(clientv3.OpPut("foo", "ABC")).
  27.         Commit()
  28.     if err != nil {
  29.         log.Fatal("run txn: " + err.Error())
  30.     }
  31. }

2.3 批量操作

批量指定操作

  1. func do() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     ops := []clientv3.Op{
  14.         clientv3.OpPut("key1", "123"),
  15.         clientv3.OpGet("key1"),
  16.         clientv3.OpPut("key2", "456"),
  17.     }
  19.     for _, op := range ops {
  20.         if _, err := cli.Do(ctx, op); err != nil {
  21.             log.Fatal(err.Error())
  22.         }
  23.     }
  24. }

2.3 watch

监视key

  1. func watch() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     go func() {
  14.         timer := time.NewTicker(time.Second)
  15.         for {
  16.             select {
  17.             case <-timer.C:
  18.                 // change foo value every second
  19.                 _, _ = cli.Put(context.TODO(), "foo", time.Now().String())
  20.                 _, _ = cli.Put(context.TODO(), "foo1", time.Now().String())
  21.                 _, _ = cli.Put(context.TODO(), "foo2", time.Now().String())
  22.                 _, _ = cli.Put(context.TODO(), "foo3", time.Now().String())
  23.                 _, _ = cli.Put(context.TODO(), "foo4", time.Now().String())
  24.             }
  25.         }
  26.     }()
  28.     //rch := cli.Watch(ctx, "foo")
  29.     rch := cli.Watch(ctx, "foo", clientv3.WithPrefix())
  30.     //rch := cli.Watch(ctx, "foo", clientv3.WithRange("foo4"))
  31.     for wresp := range rch {
  32.         for _, ev := range wresp.Events {
  33.             fmt.Printf("%s %q: %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
  34.         }
  35.     }
  36. }
  1. func watchWithProcessNotify() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     rch := cli.Watch(ctx, "foo", clientv3.WithProgressNotify())
  14.     wresp := <- rch
  15.     fmt.Printf("wresp.Header.Revision: %d\n", wresp.Header.Revision)
  16.     fmt.Println("wresp.IsProgressNotify:", wresp.IsProgressNotify())
  17. }

三、lease

2.1 创建 lease

  1. func grant() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     // etcdctl lease grant 5
  14.     // grant lease 5s
  15.     resp, err := cli.Grant(ctx, 5)
  16.     if err != nil {
  17.         log.Fatal("grant lease: " + err.Error())
  18.     }
  20.     // after 5 seconds, the key 'foo' will be removed
  21.     _, err = cli.Put(ctx, "foo", "bar", clientv3.WithLease(resp.ID))
  22.     if err != nil {
  23.         log.Fatal("put key with lease: " + err.Error())
  24.     }
  25. }

2.2 删除 lease

  1. func revoke() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     resp, err := cli.Grant(ctx, 5)
  14.     if err != nil {
  15.         log.Fatal("grant lease: " + err.Error())
  16.     }
  18.     _, err = cli.Put(ctx, "foo", "bar", clientv3.WithLease(resp.ID))
  19.     if err != nil {
  20.         log.Fatal(err)
  21.     }
  23.     // revoking lease expires the key attached to its lease ID
  24.     _, err = cli.Revoke(ctx, resp.ID)
  25.     if err != nil {
  26.         log.Fatal(err)
  27.     }
  28. }

2.3 续租

  1. func keepAlive() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     resp, err := cli.Grant(ctx, 5)
  14.     if err != nil {
  15.         log.Fatal("grant lease: " + err.Error())
  16.     }
  18.     _, err = cli.Put(ctx, "foo", "bar", clientv3.WithLease(resp.ID))
  19.     if err != nil {
  20.         log.Fatal(err)
  21.     }
  23.     ch, err := cli.KeepAlive(ctx, resp.ID)
  24.     if err != nil {
  25.         log.Fatal(err.Error())
  26.     }
  28.     ka := <- ch
  29.     fmt.Println("ttl:", ka.TTL)
  31.     // 官方提示：多数情况下使用 KeepAlive 来代替 KeepAliveOnce
  32.     kaa, err := cli.KeepAliveOnce(ctx, resp.ID)
  33.     if err != nil {
  34.         log.Fatal(err)
  35.     }
  37.     fmt.Println("ttl:", kaa.TTL)
  38. }

2.4 查询 lease

  1. func leases() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     _, err := cli.Grant(ctx, 5)
  14.     if err != nil {
  15.         log.Fatal("grant lease: " + err.Error())
  16.     }
  18.     _, err = cli.Grant(ctx, 10)
  19.     if err != nil {
  20.         log.Fatal("grant lease: " + err.Error())
  21.     }
  23.     _, err = cli.Grant(ctx, 15)
  24.     if err != nil {
  25.         log.Fatal("grant lease: " + err.Error())
  26.     }
  28.     resp, err := cli.Lease.Leases(ctx)
  29.     if err != nil {
  30.         log.Fatal(err)
  31.     }
  33.     for _, lease := range resp.Leases {
  34.         ttl, err := cli.Lease.TimeToLive(ctx, lease.ID, clientv3.WithAttachedKeys())
  35.         if err == nil {
  36.             fmt.Printf("lease: %d, ttl: %d, grantedTTL: %d\n", ttl.ID, ttl.TTL, ttl.GrantedTTL)
  37.         }
  38.     }
  39. }

四、访问控制

  1. func auth() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     auth := clientv3.NewAuth(cli)
  15.     // create role
  16.     if _, err := auth.RoleAdd(ctx, "root"); err != nil {
  17.         log.Fatal(err)
  18.     }
  20.     // create role
  21.     if _, err := auth.UserAdd(ctx, "root", "123"); err != nil {
  22.         log.Fatal(err)
  23.     }
  25.     // grant role root to user root
  26.     if _, err := auth.UserGrantRole(ctx, "root", "root"); err != nil {
  27.         log.Fatal(err)
  28.     }
  29.     if _, err := auth.UserChangePassword(ctx, "root", "123"); err != nil {
  30.         log.Fatal(err)
  31.     }
  33.     if _, err := auth.RoleAdd(ctx, "guest"); err != nil {
  34.         log.Fatal(err)
  35.     }
  36.     if _, err := auth.UserAdd(ctx, "xingyys", ""); err != nil {
  37.         log.Fatal(err)
  38.     }
  39.     if _, err := auth.UserGrantRole(ctx, "xingyys", "guest"); err != nil {
  40.         log.Fatal(err)
  41.     }
  42.     // 不知道为什么，需要在grant后更新密码
  43.     // 否则密码无效
  44.     if _, err := auth.UserChangePassword(ctx, "xingyys", "123"); err != nil {
  45.         log.Fatal(err)
  46.     }
  48.     // 添加指定key的访问权限
  49.     // read, write, readwrite
  50.     if _, err := auth.RoleGrantPermission(ctx,
  51.         "guest",
  52.         "foo",
  53.         "zoo",
  54.         clientv3.PermissionType(clientv3.PermReadWrite)); err != nil {
  55.         log.Fatal(err)
  56.     }
  58.     if _, err := auth.AuthEnable(ctx); err != nil {
  59.         log.Fatal(err)
  60.     }
  63.     authCli, _ := clientv3.New(clientv3.Config{
  64.         // etcd 集群的地址集合
  65.         Endpoints: []string{"192.168.10.10:2379"},
  66.         // 请求超时时间
  67.         DialTimeout: time.Second * 3,
  68.         Username: "xingyys",
  69.         Password: "123",
  70.     })
  71.     defer authCli.Close()
  73.     _, _ = authCli.Put(ctx, "foo", "1")
  74.     resp, _ := authCli.Get(ctx, "foo")
  75.     for _, v := range resp.Kvs {
  76.         log.Printf("%s => %q\n", v.Key, v.Value)
  77.     }
  79.     _, err := authCli.Txn(ctx).
  80.         If(clientv3.Compare(clientv3.Value("zoo1"), ">", "abc")).
  81.         Then(clientv3.OpPut("zoo1", "XYZ")).
  82.         Else(clientv3.OpPut("zoo1", "ABC")).
  83.         Commit()
  84.     log.Println(err)
  85. }

五、集群

  1. func member() {
  2.     cli, _ := clientv3.New(clientv3.Config{
  3.         // etcd 集群的地址集合
  4.         Endpoints: []string{"192.168.10.10:2379"},
  5.         // 请求超时时间
  6.         DialTimeout: time.Second * 3,
  7.     })
  8.     defer cli.Close()
  10.     ctx, cancel := context.WithCancel(context.Background())
  11.     defer cancel()
  13.     cluster := clientv3.NewCluster(cli)
  14.     resp, err := cluster.MemberList(ctx)
  15.     if err != nil {
  16.         log.Fatal(err)
  17.     }
  19.     for _, member := range resp.Members {
  20.         fmt.Printf("ID: %d | Name: %s | ClientURL: %q | PeerURL: %q\n",
  21.             member.ID,
  22.             member.Name,
  23.             member.ClientURLs,
  24.             member.PeerURLs)
  25.     }
  27.     //_, _ = cluster.MemberAdd(ctx, []string{"192.168.10.10:2370", "192.168.10.11:2379"})
  28.     //_, _ = cluster.MemberRemove(ctx, // id)
  29.     //_, _ = cluster.MemberUpdate(ctx, // id, // peer)
  30. }

六、并发

6.1 锁

  1. func lock() {
  2.     cli, err := clientv3.New(clientv3.Config{
  3.         Endpoints: []string{"192.168.10.10:2379"},
  4.     })
  5.     if err != nil {
  6.         log.Fatal(err)
  7.     }
  8.     defer cli.Close()
  10.     // 注册session
  11.     s1, err := concurrency.NewSession(cli)
  12.     if err != nil {
  13.         log.Fatal(err)
  14.     }
  15.     defer s1.Close()
  16.     m1 := concurrency.NewMutex(s1, "/lock")
  18.     s2, err := concurrency.NewSession(cli)
  19.     if err != nil {
  20.         log.Fatal(err)
  21.     }
  22.     defer s2.Close()
  23.     m2 := concurrency.NewMutex(s2, "/lock")
  25.     // acquired lock for s1
  26.     if err := m1.Lock(context.TODO()); err != nil {
  27.         log.Fatal(err)
  28.     }
  29.     fmt.Println("acquired lock for s1")
  31.     m2Locked := make(chan struct{})
  32.     go func() {
  33.         defer close(m2Locked)
  34.         // wait util s1 is locks /lock
  35.         if err := m2.Lock(context.TODO()); err != nil {
  36.             log.Fatal(err)
  37.         }
  38.     }()
  40.     if err := m1.Unlock(context.TODO()); err != nil {
  41.         log.Fatal(err)
  42.     }
  43.     fmt.Println("release lock for s1")
  45.     <-m2Locked
  46.     fmt.Println("acquired lock for s2")
  47. }
  49. func tryLock() {
  50.     cli, err := clientv3.New(clientv3.Config{
  51.         Endpoints: []string{"192.168.10.10:2379"},
  52.     })
  53.     if err != nil {
  54.         log.Fatal(err)
  55.     }
  56.     defer cli.Close()
  58.     // 注册session
  59.     s1, err := concurrency.NewSession(cli)
  60.     if err != nil {
  61.         log.Fatal(err)
  62.     }
  63.     defer s1.Close()
  64.     m1 := concurrency.NewMutex(s1, "/lock")
  66.     s2, err := concurrency.NewSession(cli)
  67.     if err != nil {
  68.         log.Fatal(err)
  69.     }
  70.     defer s2.Close()
  71.     m2 := concurrency.NewMutex(s2, "/lock")
  73.     // acquire lock for s1
  74.     if err = m1.Lock(context.TODO()); err != nil {
  75.         log.Fatal(err)
  76.     }
  77.     fmt.Println("acquired lock for s1")
  79.     if err = m2.TryLock(context.TODO()); err == nil {
  80.         log.Fatal("should not acquire lock")
  81.     }
  82.     if err == concurrency.ErrLocked {
  83.         fmt.Println("cannot acquire lock for s2, as already locked in another session")
  84.     }
  86.     if err = m1.Unlock(context.TODO()); err != nil {
  87.         log.Fatal(err)
  88.     }
  89.     fmt.Println("released lock for s1")
  90.     if err = m2.TryLock(context.TODO()); err != nil {
  91.         log.Fatal(err)
  92.     }
  93.     fmt.Println("acquired lock for s2")
  94. }

6.2 领导选举

  1. func election() {
  2.     cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"192.168.10.10:2379"}})
  3.     if err != nil {
  4.         log.Fatal(err)
  5.     }
  6.     defer cli.Close()
  8.     // create two separate sessions for election competition
  9.     s1, err := concurrency.NewSession(cli)
  10.     if err != nil {
  11.         log.Fatal(err)
  12.     }
  13.     defer s1.Close()
  14.     e1 := concurrency.NewElection(s1, "/my-election/")
  16.     s2, err := concurrency.NewSession(cli)
  17.     if err != nil {
  18.         log.Fatal(err)
  19.     }
  20.     defer s2.Close()
  21.     e2 := concurrency.NewElection(s2, "/my-election/")
  23.     // create competing candidates, with e1 initially losing to e2
  24.     var wg sync.WaitGroup
  25.     wg.Add(2)
  26.     electc := make(chan *concurrency.Election, 2)
  27.     go func() {
  28.         defer wg.Done()
  29.         // delay candidacy so e2 wins first
  30.         time.Sleep(3 * time.Second)
  31.         if err := e1.Campaign(context.Background(), "e1"); err != nil {
  32.             log.Fatal(err)
  33.         }
  34.         electc <- e1
  35.     }()
  36.     go func() {
  37.         defer wg.Done()
  38.         if err := e2.Campaign(context.Background(), "e2"); err != nil {
  39.             log.Fatal(err)
  40.         }
  41.         electc <- e2
  42.     }()
  44.     cctx, cancel := context.WithCancel(context.TODO())
  45.     defer cancel()
  47.     e := <-electc
  48.     fmt.Println("completed first election with", string((<-e.Observe(cctx)).Kvs[0].Value))
  50.     // resign so next candidate can be elected
  51.     if err := e.Resign(context.TODO()); err != nil {
  52.         log.Fatal(err)
  53.     }
  55.     e = <-electc
  56.     fmt.Println("completed second election with", string((<-e.Observe(cctx)).Kvs[0].Value))
  58.     wg.Wait()
  59. }

6.3 软件事务内存

  1. func stm() {
  2.     cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"192.168.10.10:2379"}})
  3.     if err != nil {
  4.         log.Fatal(err)
  5.     }
  6.     defer cli.Close()
  8.     // set up "accounts"
  9.     totalAccounts := 5
  10.     for i := 0; i < totalAccounts; i++ {
  11.         k := fmt.Sprintf("accts/%d", i)
  12.         if _, err = cli.Put(context.TODO(), k, "100"); err != nil {
  13.             log.Fatal(err)
  14.         }
  15.     }
  17.     exchange := func(stm concurrency.STM) error {
  18.         from, to := rand.Intn(totalAccounts), rand.Intn(totalAccounts)
  19.         if from == to {
  20.             // nothing to do
  21.             return nil
  22.         }
  23.         // read values
  24.         fromK, toK := fmt.Sprintf("accts/%d", from), fmt.Sprintf("accts/%d", to)
  25.         fromV, toV := stm.Get(fromK), stm.Get(toK)
  26.         fromInt, toInt := 0, 0
  27.         fmt.Sscanf(fromV, "%d", &fromInt)
  28.         fmt.Sscanf(toV, "%d", &toInt)
  30.         // transfer amount
  31.         xfer := fromInt / 2
  32.         fromInt, toInt = fromInt-xfer, toInt+xfer
  34.         // write back
  35.         stm.Put(fromK, fmt.Sprintf("%d", fromInt))
  36.         stm.Put(toK, fmt.Sprintf("%d", toInt))
  37.         return nil
  38.     }
  40.     // concurrently exchange values between accounts
  41.     var wg sync.WaitGroup
  42.     wg.Add(10)
  43.     for i := 0; i < 10; i++ {
  44.         go func() {
  45.             defer wg.Done()
  46.             if _, serr := concurrency.NewSTM(cli, exchange); serr != nil {
  47.                 log.Fatal(serr)
  48.             }
  49.         }()
  50.     }
  51.     wg.Wait()
  53.     // confirm account sum matches sum from beginning.
  54.     sum := 0
  55.     accts, err := cli.Get(context.TODO(), "accts/", clientv3.WithPrefix())
  56.     if err != nil {
  57.         log.Fatal(err)
  58.     }
  59.     for _, kv := range accts.Kvs {
  60.         v := 0
  61.         fmt.Sscanf(string(kv.Value), "%d", &v)
  62.         sum += v
  63.     }
  65.     fmt.Println("account sum is", sum)
  66. }