一 、参考链接

  1. go操作etcd
  2. go.etcd.io/etcd/clientv3
  3. https://pkg.go.dev/go.etcd.io/etcd/client/v3#section-readme
  4. github.com/coreos/etcd
  5. https://etcd.io/docs/v3.3/demo/
  6. etcd：从应用场景到实现原理的全方位解读

    二 、 函数

    ```go func NewLocker(s *Session, pfx string) sync.Locker

func NewSTM(c v3.Client, apply func(STM) error, so …stmOption) (v3.TxnResponse, error)

func NewSTMReadCommitted(ctx context.Context, c v3.Client, apply func(STM) error) (v3.TxnResponse, error)

func NewSTMRepeatable(ctx context.Context, c v3.Client, apply func(STM) error) (v3.TxnResponse, error)

func NewSTMSerializable(ctx context.Context, c v3.Client, apply func(STM) error) (v3.TxnResponse, error)

func WithAbortContext(ctx context.Context) stmOption

func WithIsolation(lvl Isolation) stmOption

func WithPrefetch(keys …string) stmOption

type Election func NewElection(s Session, pfx string) Election func ResumeElection(s Session, pfx string, leaderKey string, leaderRev int64) Election func (e Election) Campaign(ctx context.Context, val string) error func (e Election) Header() pb.ResponseHeader func (e Election) Key() string func (e Election) Leader(ctx context.Context) (v3.GetResponse, error) func (e Election) Observe(ctx context.Context) <-chan v3.GetResponse func (e Election) Proclaim(ctx context.Context, val string) error func (e Election) Resign(ctx context.Context) (err error) func (e Election) Rev() int64

type Isolation

type Mutex func NewMutex(s Session, pfx string) Mutex func (m Mutex) Header() pb.ResponseHeader func (m Mutex) IsOwner() v3.Cmp func (m Mutex) Key() string func (m Mutex) Lock(ctx context.Context) error func (m Mutex) Unlock(ctx context.Context) error

type STM

type Session func NewSession(client v3.Client, opts …SessionOption) (Session, error) func (s Session) Client() v3.Client func (s Session) Close() error func (s Session) Done() <-chan struct{} func (s Session) Lease() v3.LeaseID func (s Session) Orphan()

type SessionOption func WithContext(ctx context.Context) SessionOption func WithLease(leaseID v3.LeaseID) SessionOption func WithTTL(ttl int) SessionOption

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

官方示例

  1. //先连接,config配置里可以加很多东西,比如自己的证书
  2. cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
  3. if err != nil {
  4.     log.Fatal(err)
  5. }
  6. defer cli.Close()
  8. // 搞两个会话,第一个
  9. // create two separate sessions for election competition
  10. s1, err := concurrency.NewSession(cli)
  11. if err != nil {
  12.     log.Fatal(err)
  13. }
  14. defer s1.Close()
  15. e1 := concurrency.NewElection(s1, "/my-election/")
  17. // 搞两个会话,第二个
  18. s2, err := concurrency.NewSession(cli)
  19. if err != nil {
  20.     log.Fatal(err)
  21. }
  22. defer s2.Close()
  23. e2 := concurrency.NewElection(s2, "/my-election/")
  25. // create competing candidates, with e1 initially losing to e2
  26. var wg sync.WaitGroup
  27. wg.Add(2)
  28. electc := make(chan *concurrency.Election, 2)
  29. go func() {
  30.     defer wg.Done()
  31.     // delay candidacy so e2 wins first
  32.     time.Sleep(3 * time.Second)
  33.     if err := e1.Campaign(context.Background(), "e1"); err != nil {
  34.         log.Fatal(err)
  35.     }
  36.     electc <- e1
  37. }()
  38. go func() {
  39.     defer wg.Done()
  40.     if err := e2.Campaign(context.Background(), "e2"); err != nil {
  41.         log.Fatal(err)
  42.     }
  43.     electc <- e2
  44. }()
  46. cctx, cancel := context.WithCancel(context.TODO())
  47. defer cancel()
  49. e := <-electc
  50. fmt.Println("completed first election with", string((<-e.Observe(cctx)).Kvs[0].Value))
  52. // resign so next candidate can be elected
  53. if err := e.Resign(context.TODO()); err != nil {
  54.     log.Fatal(err)
  55. }
  57. e = <-electc
  58. fmt.Println("completed second election with", string((<-e.Observe(cctx)).Kvs[0].Value))
  60. wg.Wait()

输出:

  1. Output:
  3. completed first election with e2
  4. completed second election with e1
  1. cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
  2. if err != nil {
  3.     log.Fatal(err)
  4. }
  5. defer cli.Close()
  7. // create two separate sessions for lock competition
  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. // acquire lock for 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.     // wait until s1 is locks /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")
  46. Output:
  48. acquired lock for s1
  49. released lock for s1
  50. acquired lock for s2
  1. cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
  2. if err != nil {
  3.     log.Fatal(err)
  4. }
  5. defer cli.Close()
  7. // set up "accounts"
  8. totalAccounts := 5
  9. for i := 0; i < totalAccounts; i++ {
  10.     k := fmt.Sprintf("accts/%d", i)
  11.     if _, err = cli.Put(context.TODO(), k, "100"); err != nil {
  12.         log.Fatal(err)
  13.     }
  14. }
  16. exchange := func(stm concurrency.STM) error {
  17.     from, to := rand.Intn(totalAccounts), rand.Intn(totalAccounts)
  18.     if from == to {
  19.         // nothing to do
  20.         return nil
  21.     }
  22.     // read values
  23.     fromK, toK := fmt.Sprintf("accts/%d", from), fmt.Sprintf("accts/%d", to)
  24.     fromV, toV := stm.Get(fromK), stm.Get(toK)
  25.     fromInt, toInt := 0, 0
  26.     fmt.Sscanf(fromV, "%d", &fromInt)
  27.     fmt.Sscanf(toV, "%d", &toInt)
  29.     // transfer amount
  30.     xfer := fromInt / 2
  31.     fromInt, toInt = fromInt-xfer, toInt+xfer
  33.     // write back
  34.     stm.Put(fromK, fmt.Sprintf("%d", fromInt))
  35.     stm.Put(toK, fmt.Sprintf("%d", toInt))
  36.     return nil
  37. }
  39. // concurrently exchange values between accounts
  40. var wg sync.WaitGroup
  41. wg.Add(10)
  42. for i := 0; i < 10; i++ {
  43.     go func() {
  44.         defer wg.Done()
  45.         if _, serr := concurrency.NewSTM(cli, exchange); serr != nil {
  46.             log.Fatal(serr)
  47.         }
  48.     }()
  49. }
  50. wg.Wait()
  52. // confirm account sum matches sum from beginning.
  53. sum := 0
  54. accts, err := cli.Get(context.TODO(), "accts/", clientv3.WithPrefix())
  55. if err != nil {
  56.     log.Fatal(err)
  57. }
  58. for _, kv := range accts.Kvs {
  59.     v := 0
  60.     fmt.Sscanf(string(kv.Value), "%d", &v)
  61.     sum += v
  62. }
  64. fmt.Println("account sum is", sum)
  67. Output:
  69. account sum is 500

四、注意

  1. 计算保护状态：

分配集群：删除保护状态，按照灾备端 —— 重新计算 保护状态

收回集群：修改保护状态，修改为已删除
集群收回之后 如果不是未保护，状态更改为已删除
获取全部 — 直接全部删除
image.png