etcd入門 & go操做etcd

1、etcd

一、簡介

etcd是使用Go語言開發的一個開源的、高可用的分佈式key-value存儲系統，能夠用於配置共享和服務的註冊和發現。

相似項目有`zookeeper`和`consul`。

二、特色

`徹底複製`：集羣中的每一個節點均可以使用完整的存檔
`高可用性`：Etcd可用於避免硬件的單點故障或網絡問題
`一致性`：每次讀取都會返回跨多主機的最新寫入
`簡單`：包括一個定義良好、面向用戶的API（gRPC）
`安全`：實現了帶有可選的客戶端證書身份驗證的自動化TLS
`快速`：每秒10000次寫入的基準速度
`可靠`：使用Raft算法實現了強一致、高可用的服務存儲目錄

三、應用場景

服務發現
配置中心
分佈式鎖 - 1.保持獨佔 2.控制時序

注: 奇數節點原則

四、etcd集羣

...

五、go 操做 etcd

go get go.etcd.io/etcd/clientv3

官方文檔

PUT GET

package main

import (
    "context"
    "fmt"
    "time"

    "go.etcd.io/etcd/clientv3"
)

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        // handle error!
        fmt.Printf("connect to etcd failed, err:%v\n", err)
        return
    }
    fmt.Println("connect to etcd success")
    defer cli.Close()
    // put
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    _, err = cli.Put(ctx, "kk", "123")
    cancel()
    if err != nil {
        fmt.Printf("put to etcd failed, err:%v\n", err)
        return
    }
    // get
    ctx, cancel = context.WithTimeout(context.Background(), time.Second)
    resp, err := cli.Get(ctx, "kk")
    cancel()
    if err != nil {
        fmt.Printf("get from etcd failed, err:%v\n", err)
        return
    }
    for _, ev := range resp.Kvs {
        fmt.Printf("%s:%s\n", ev.Key, ev.Value)
    }
}

WATCH

package main

import (
    "context"
    "fmt"
    "time"

    "go.etcd.io/etcd/clientv3"
)

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        fmt.Printf("connect to etcd failed, err:%v\n", err)
        return
    }
    fmt.Println("connect to etcd success")
    defer cli.Close()
    // watch key:kk change
    rch := cli.Watch(context.Background(), "kk") // <-chan WatchResponse
    for wresp := range rch {
        for _, ev := range wresp.Events {
            fmt.Printf("Type: %s Key:%s Value:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
        }
    }
}

命令行操做更改鍵值

etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 put kk "123"
OK

etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 del kk
1

etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 put kk "321"
OK

通知

watch> ./watch
connect to etcd success
Type: PUT Key:kk Value:123
Type: DELETE Key:kk Value:
Type: PUT Key:kk Value:321

LEASE

package main

import (
    "fmt"
    "time"
)

// etcd lease

import (
    "context"
    "log"

    "go.etcd.io/etcd/clientv3"
)

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: time.Second * 5,
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("connect to etcd success.")
    defer cli.Close()

    // 建立一個5秒的租約
    resp, err := cli.Grant(context.TODO(), 5)
    if err != nil {
        log.Fatal(err)
    }

    // 5秒鐘以後, /nazha/ 這個key就會被移除
    _, err = cli.Put(context.TODO(), "/kk/", "123", clientv3.WithLease(resp.ID))
    if err != nil {
        log.Fatal(err)
    }
}

KEEPALIVE

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "go.etcd.io/etcd/clientv3"
)

// etcd keepAlive

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: time.Second * 5,
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("connect to etcd success.")
    defer cli.Close()

    resp, err := cli.Grant(context.TODO(), 5)
    if err != nil {
        log.Fatal(err)
    }

    _, err = cli.Put(context.TODO(), "/kk/", "123", clientv3.WithLease(resp.ID))
    if err != nil {
        log.Fatal(err)
    }

    // the key 'foo' will be kept forever
    ch, kaerr := cli.KeepAlive(context.TODO(), resp.ID)
    if kaerr != nil {
        log.Fatal(kaerr)
    }
    for {
        ka := <-ch
        fmt.Println("ttl:", ka.TTL)
    }
}

六、分佈式鎖

cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
if err != nil {
    log.Fatal(err)
}
defer cli.Close()

// 建立兩個單獨的會話用來演示鎖競爭
s1, err := concurrency.NewSession(cli)
if err != nil {
    log.Fatal(err)
}
defer s1.Close()
m1 := concurrency.NewMutex(s1, "/my-lock/")

s2, err := concurrency.NewSession(cli)
if err != nil {
    log.Fatal(err)
}
defer s2.Close()
m2 := concurrency.NewMutex(s2, "/my-lock/")

// 會話s1獲取鎖
if err := m1.Lock(context.TODO()); err != nil {
    log.Fatal(err)
}
fmt.Println("acquired lock for s1")

m2Locked := make(chan struct{})
go func() {
    defer close(m2Locked)
    // 等待直到會話s1釋放了/my-lock/的鎖
    if err := m2.Lock(context.TODO()); err != nil {
        log.Fatal(err)
    }
}()

if err := m1.Unlock(context.TODO()); err != nil {
    log.Fatal(err)
}
fmt.Println("released lock for s1")

<-m2Locked
fmt.Println("acquired lock for s2")

輸出

acquired lock for s1
released lock for s1
acquired lock for s2