1. 操作ETCD

这里使用官方的etcd/clientv3包来连接etcd并进行相关操作。

1.1.1. 安装

  1.     go get go.etcd.io/etcd/clientv3

1.1.2. put和get操作

put命令用来设置键值对数据，get命令用来根据key获取值。

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "time"
  8.     "go.etcd.io/etcd/clientv3"
  9. )
  11. // etcd client put/get demo
  12. // use etcd/clientv3
  14. func main() {
  15.     cli, err := clientv3.New(clientv3.Config{
  16.         Endpoints:   []string{"127.0.0.1:2379"},
  17.         DialTimeout: 5 * time.Second,
  18.     })
  19.     if err != nil {
  20.         // handle error!
  21.         fmt.Printf("connect to etcd failed, err:%v\n", err)
  22.         return
  23.     }
  24.     fmt.Println("connect to etcd success")
  25.     defer cli.Close()
  26.     // put
  27.     ctx, cancel := context.WithTimeout(context.Background(), time.Second)
  28.     _, err = cli.Put(ctx, "lmh", "lmh")
  29.     cancel()
  30.     if err != nil {
  31.         fmt.Printf("put to etcd failed, err:%v\n", err)
  32.         return
  33.     }
  34.     // get
  35.     ctx, cancel = context.WithTimeout(context.Background(), time.Second)
  36.     resp, err := cli.Get(ctx, "lmh")
  37.     cancel()
  38.     if err != nil {
  39.         fmt.Printf("get from etcd failed, err:%v\n", err)
  40.         return
  41.     }
  42.     for _, ev := range resp.Kvs {
  43.         fmt.Printf("%s:%s\n", ev.Key, ev.Value)
  44.     }
  45. }

1.1.3. watch操作

watch用来获取未来更改的通知。

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "time"
  8.     "go.etcd.io/etcd/clientv3"
  9. )
  11. // watch demo
  13. func main() {
  14.     cli, err := clientv3.New(clientv3.Config{
  15.         Endpoints:   []string{"127.0.0.1:2379"},
  16.         DialTimeout: 5 * time.Second,
  17.     })
  18.     if err != nil {
  19.         fmt.Printf("connect to etcd failed, err:%v\n", err)
  20.         return
  21.     }
  22.     fmt.Println("connect to etcd success")
  23.     defer cli.Close()
  24.     // watch key:lmh change
  25.     rch := cli.Watch(context.Background(), "lmh") // <-chan WatchResponse
  26.     for wresp := range rch {
  27.         for _, ev := range wresp.Events {
  28.             fmt.Printf("Type: %s Key:%s Value:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
  29.         }
  30.     }
  31. }

将上面的代码保存编译执行，此时程序就会等待etcd中lmh这个key的变化。

例如：我们打开终端执行以下命令修改、删除、设置lmh这个key。

  1.     etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 put lmh "lmh1"
  2.     OK
  4.     etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 del lmh
  5.     1
  7.     etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 put lmh "lmh2"
  8.     OK

上面的程序都能收到如下通知。

  1.     watch>watch.exe
  2.     connect to etcd success
  3.     Type: PUT Key:lmh Value:lmh1
  4.     Type: DELETE Key:lmh Value:
  5.     Type: PUT Key:lmh Value:lmh2

1.1.4. lease租约

  1. package main
  3. import (
  4.     "fmt"
  5.     "time"
  6. )
  8. // etcd lease
  10. import (
  11.     "context"
  12.     "log"
  14.     "go.etcd.io/etcd/clientv3"
  15. )
  17. func main() {
  18.     cli, err := clientv3.New(clientv3.Config{
  19.         Endpoints:   []string{"127.0.0.1:2379"},
  20.         DialTimeout: time.Second * 5,
  21.     })
  22.     if err != nil {
  23.         log.Fatal(err)
  24.     }
  25.     fmt.Println("connect to etcd success.")
  26.     defer cli.Close()
  28.     // 创建一个5秒的租约
  29.     resp, err := cli.Grant(context.TODO(), 5)
  30.     if err != nil {
  31.         log.Fatal(err)
  32.     }
  34.     // 5秒钟之后, /lmh/ 这个key就会被移除
  35.     _, err = cli.Put(context.TODO(), "/lmh/", "lmh", clientv3.WithLease(resp.ID))
  36.     if err != nil {
  37.         log.Fatal(err)
  38.     }
  39. }

1.1.5. keepAlive

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "log"
  7.     "time"
  9.     "go.etcd.io/etcd/clientv3"
  10. )
  12. // etcd keepAlive
  14. func main() {
  15.     cli, err := clientv3.New(clientv3.Config{
  16.         Endpoints:   []string{"127.0.0.1:2379"},
  17.         DialTimeout: time.Second * 5,
  18.     })
  19.     if err != nil {
  20.         log.Fatal(err)
  21.     }
  22.     fmt.Println("connect to etcd success.")
  23.     defer cli.Close()
  25.     resp, err := cli.Grant(context.TODO(), 5)
  26.     if err != nil {
  27.         log.Fatal(err)
  28.     }
  30.     _, err = cli.Put(context.TODO(), "/lmh/", "lmh", clientv3.WithLease(resp.ID))
  31.     if err != nil {
  32.         log.Fatal(err)
  33.     }
  35.     // the key 'foo' will be kept forever
  36.     ch, kaerr := cli.KeepAlive(context.TODO(), resp.ID)
  37.     if kaerr != nil {
  38.         log.Fatal(kaerr)
  39.     }
  40.     for {
  41.         ka := <-ch
  42.         fmt.Println("ttl:", ka.TTL)
  43.     }
  44. }

1.1.6. 基于etcd实现分布式锁

go.etcd.io/etcd/clientv3/concurrency在etcd之上实现并发操作，如分布式锁、屏障和选举。

导入该包：

  1.     import "go.etcd.io/etcd/clientv3/concurrency"

基于etcd实现的分布式锁示例：

  1. cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
  2. if err != nil {
  3.     log.Fatal(err)
  4. }
  5. defer cli.Close()
  7. // 创建两个单独的会话用来演示锁竞争
  8. s1, err := concurrency.NewSession(cli)
  9. if err != nil {
  10.     log.Fatal(err)
  11. }
  12. defer s1.Close()
  13. m1 := concurrency.NewMutex(s1, "/my-lock/")
  15. s2, err := concurrency.NewSession(cli)
  16. if err != nil {
  17.     log.Fatal(err)
  18. }
  19. defer s2.Close()
  20. m2 := concurrency.NewMutex(s2, "/my-lock/")
  22. // 会话s1获取锁
  23. if err := m1.Lock(context.TODO()); err != nil {
  24.     log.Fatal(err)
  25. }
  26. fmt.Println("acquired lock for s1")
  28. m2Locked := make(chan struct{})
  29. go func() {
  30.     defer close(m2Locked)
  31.     // 等待直到会话s1释放了/my-lock/的锁
  32.     if err := m2.Lock(context.TODO()); err != nil {
  33.         log.Fatal(err)
  34.     }
  35. }()
  37. if err := m1.Unlock(context.TODO()); err != nil {
  38.     log.Fatal(err)
  39. }
  40. fmt.Println("released lock for s1")
  42. <-m2Locked
  43. fmt.Println("acquired lock for s2")

输出：

  1.     acquired lock for s1
  2.     released lock for s1
  3.     acquired lock for s2

查看文档了解更多