etcd 的 go sdk 的使用方法: etcd/clientv3，选举 leader 的方法

// Copyright (C) 2019 lijiaocn <[email protected]> wechat: lijiaocn
//
// Distributed under terms of the GPL license.

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/golang/glog"
    "go.etcd.io/etcd/clientv3"
)

func toString(obj interface{}) (string, error) {
    bytes, err := json.Marshal(obj)
    if err != nil {
        return "", err
    }
    return string(bytes), nil
}

// 写入ETCD
func PUT(cli *clientv3.Client, key, val string) (string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    resp, err := cli.Put(ctx, key, val)
    cancel()
    if err != nil {
        return "", err
    }
    return toString(resp)
}

// 查询ETCD
func GET(cli *clientv3.Client, key string, opts ...clientv3.OpOption) (string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    resp, err := cli.Get(ctx, key, opts...)
    cancel()
    if err != nil {
        return "", err
    }
    return toString(resp)
}

func main() {

    config := clientv3.Config{
        Endpoints:   []string{"localhost:2379", "localhost:22379", "localhost:32379"},
        DialTimeout: 5 * time.Second,
    }

    cli, err := clientv3.New(config)
    if err != nil {
        glog.Fatal(err.Error())
    }

    defer func() {
        if err := cli.Close(); err != nil {
            glog.Error(err.Error())
        }
    }()

    if v, err := PUT(cli, "sample_key", "sample_value123"); err != nil {
        glog.Errorf(err.Error())
    } else {
        fmt.Printf("PUT RESULT: %s\n", v)
    }

    if v, err := GET(cli, "sample_key"); err != nil {
        glog.Errorf(err.Error())
    } else {
        fmt.Printf("GET RESULT: %s\n", v)
    }
}

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
resp, err := cli.Get(ctx, key, opts...)
cancel()
if err != nil {
    return "", err
}

// etcd/clientv3/watch.go
type Watcher interface {
    // Watch watches on a key or prefix. The watched events will be returned
    // through the returned channel. If revisions waiting to be sent over the
    // watch are compacted, then the watch will be canceled by the server, the
    // client will post a compacted error watch response, and the channel will close.
    Watch(ctx context.Context, key string, opts ...OpOption) WatchChan

    // Close closes the watcher and cancels all watch requests.
    Close() error
}

func WATCH(cli *clientv3.Client, key string, opts ...clientv3.OpOption) (clientv3.WatchChan, error) {
    watchChan := cli.Watch(context.Background(), key, opts...)
    return watchChan, nil
}

func main() {

    config := clientv3.Config{
        Endpoints:   []string{"localhost:2379", "localhost:22379", "localhost:32379"},
        DialTimeout: 5 * time.Second,
    }

    cli, err := clientv3.New(config)
    if err != nil {
        glog.Fatal(err.Error())
    }

    defer func() {
        if err := cli.Close(); err != nil {
            glog.Error(err.Error())
        }
    }()

    if watchChan, err := WATCH(cli, "sample_key"); err != nil {
        glog.Errorf(err.Error())
    } else {
        for {
            wr := <-watchChan
            if str, err := toString(wr); err == nil {
                fmt.Printf("%s\n", str)
            } else {
                glog.Error(err.Error())
            }
        }
    }
}

type PutResponse struct {
    Header *ResponseHeader `protobuf:"bytes,1,opt,name=header" json:"header,omitempty"`
    // if prev_kv is set in the request, the previous key-value pair will be returned.
    PrevKv *mvccpb.KeyValue `protobuf:"bytes,2,opt,name=prev_kv,json=prevKv" json:"prev_kv,omitempty"`
}

type RangeResponse struct {
    Header *ResponseHeader `protobuf:"bytes,1,opt,name=header" json:"header,omitempty"`
    // kvs is the list of key-value pairs matched by the range request.
    // kvs is empty when count is requested.
    Kvs []*mvccpb.KeyValue `protobuf:"bytes,2,rep,name=kvs" json:"kvs,omitempty"`
    // more indicates if there are more keys to return in the requested range.
    More bool `protobuf:"varint,3,opt,name=more,proto3" json:"more,omitempty"`
    // count is set to the number of keys within the range when requested.
    Count int64 `protobuf:"varint,4,opt,name=count,proto3" json:"count,omitempty"`
}

type WatchResponse struct {
    Header pb.ResponseHeader
    Events []*Event

    // CompactRevision is the minimum revision the watcher may receive.
    CompactRevision int64

    // Canceled is used to indicate watch failure.
    // If the watch failed and the stream was about to close, before the channel is closed,
    // the channel sends a final response that has Canceled set to true with a non-nil Err().
    Canceled bool

    // Created is used to indicate the creation of the watcher.
    Created bool

    closeErr error

    // cancelReason is a reason of canceling watch
    cancelReason string
}

    config := clientv3.Config{
        Endpoints:   []string{"localhost:2379", "localhost:22379", "localhost:32379"},
        DialTimeout: 5 * time.Second,
    }

    cli, err := clientv3.New(config)
    if err != nil {
        glog.Fatal(err.Error())
    }

    var s *concurrency.Session
    s, err = concurrency.NewSession(cli, concurrency.WithTTL(1))
    if err != nil {
        log.Fatal(err)
    }

    e := concurrency.NewElection(s, "/testelection")
    //竞选 Leader，直到成为 Leader 函数才返回
    if err = e.Campaign(context.Background(), node); err != nil {
        glog.Fatalf("Campaign() returned non nil err: %s", err)
    }
    fmt.Printf("I'm leader")