Go操作etcd的實現示例

Posted on 2021-09-02 by WalkonNet

etcdetcd介紹

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

類似項目有zookeeper和consul。

etcd具有以下特點：

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

etcd應用場景

服務發現

服務發現要解決的也是分佈式系統中最常見的問題之一，即在同一個分佈式集群中的進程或服務，要如何才能找到對方並建立連接。本質上來說，服務發現就是想要瞭解集群中是否有進程在監聽 udp 或 tcp 端口，並且通過名字就可以查找和連接。

配置中心

將一些配置信息放到 etcd 上進行集中管理。

這類場景的使用方式通常是這樣：應用在啟動的時候主動從 etcd 獲取一次配置信息，同時，在 etcd 節點上註冊一個 Watcher 並等待，以後每次配置有更新的時候，etcd 都會實時通知訂閱者，以此達到獲取最新配置信息的目的。

分佈式鎖

因為 etcd 使用 Raft 算法保持瞭數據的強一致性，某次操作存儲到集群中的值必然是全局一致的，所以很容易實現分佈式鎖。鎖服務有兩種使用方式，一是保持獨占，二是控制時序。

保持獨占即所有獲取鎖的用戶最終隻有一個可以得到。etcd 為此提供瞭一套實現分佈式鎖原子操作 CAS（CompareAndSwap）的 API。通過設置prevExist值，可以保證在多個節點同時去創建某個目錄時，隻有一個成功。而創建成功的用戶就可以認為是獲得瞭鎖。

控制時序，即所有想要獲得鎖的用戶都會被安排執行，但是獲得鎖的順序也是全局唯一的，同時決定瞭執行順序。etcd 為此也提供瞭一套 API（自動創建有序鍵），對一個目錄建值時指定為POST動作，這樣 etcd 會自動在目錄下生成一個當前最大的值為鍵，存儲這個新的值（客戶端編號）。同時還可以使用 API 按順序列出所有當前目錄下的鍵值。此時這些鍵的值就是客戶端的時序，而這些鍵中存儲的值可以是代表客戶端的編號。

為什麼用 etcd 而不用ZooKeeper？

etcd 實現的這些功能，ZooKeeper都能實現。那麼為什麼要用 etcd 而非直接使用ZooKeeper呢？

為什麼不選擇ZooKeeper？

部署維護復雜，其使用的Paxos強一致性算法復雜難懂。官方隻提供瞭Java和C兩種語言的接口。
使用Java編寫引入大量的依賴。運維人員維護起來比較麻煩。
最近幾年發展緩慢，不如etcd和consul等後起之秀。

為什麼選擇etcd？

簡單。使用 Go 語言編寫部署簡單；支持HTTP/JSON API,使用簡單；使用 Raft 算法保證強一致性讓用戶易於理解。
etcd 默認數據一更新就進行持久化。
etcd 支持 SSL 客戶端安全認證。

最後，etcd 作為一個年輕的項目，正在高速迭代和開發中，這既是一個優點，也是一個缺點。優點是它的未來具有無限的可能性，缺點是無法得到大項目長時間使用的檢驗。然而，目前 CoreOS、Kubernetes和CloudFoundry等知名項目均在生產環境中使用瞭etcd，所以總的來說，etcd值得你去嘗試。

etcd集群

etcd 作為一個高可用鍵值存儲系統，天生就是為集群化而設計的。由於 Raft 算法在做決策時需要多數節點的投票，所以 etcd 一般部署集群推薦奇數個節點，推薦的數量為 3、5 或者 7 個節點構成一個集群。

搭建一個3節點集群示例：

在每個etcd節點指定集群成員，為瞭區分不同的集群最好同時配置一個獨一無二的token。

下面是提前定義好的集群信息，其中n1、n2和n3表示3個不同的etcd節點。

TOKEN=token-01
CLUSTER_STATE=new
CLUSTER=n1=http://10.240.0.17:2380,n2=http://10.240.0.18:2380,n3=http://10.240.0.19:2380

在n1這臺機器上執行以下命令來啟動etcd：

etcd --data-dir=data.etcd --name n1 \
    --initial-advertise-peer-urls http://10.240.0.17:2380 --listen-peer-urls http://10.240.0.17:2380 \
    --advertise-client-urls http://10.240.0.17:2379 --listen-client-urls http://10.240.0.17:2379 \
    --initial-cluster ${CLUSTER} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

在n2這臺機器上執行以下命令啟動etcd：

etcd --data-dir=data.etcd --name n2 \
    --initial-advertise-peer-urls http://10.240.0.18:2380 --listen-peer-urls http://10.240.0.18:2380 \
    --advertise-client-urls http://10.240.0.18:2379 --listen-client-urls http://10.240.0.18:2379 \
    --initial-cluster ${CLUSTER} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

在n3這臺機器上執行以下命令啟動etcd：

etcd --data-dir=data.etcd --name n3 \
    --initial-advertise-peer-urls http://10.240.0.19:2380 --listen-peer-urls http://10.240.0.19:2380 \
    --advertise-client-urls http://10.240.0.19:2379 --listen-client-urls http://10.240.0.19:2379 \
    --initial-cluster ${CLUSTER} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

etcd 官網提供瞭一個可以公網訪問的 etcd 存儲地址。你可以通過如下命令得到 etcd 服務的目錄，並把它作為-discovery參數使用。

curl https://discovery.etcd.io/new?size=3
https://discovery.etcd.io/a81b5818e67a6ea83e9d4daea5ecbc92
 
# grab this token
TOKEN=token-01
CLUSTER_STATE=new
DISCOVERY=https://discovery.etcd.io/a81b5818e67a6ea83e9d4daea5ecbc92
 
 
etcd --data-dir=data.etcd --name n1 \
    --initial-advertise-peer-urls http://10.240.0.17:2380 --listen-peer-urls http://10.240.0.17:2380 \
    --advertise-client-urls http://10.240.0.17:2379 --listen-client-urls http://10.240.0.17:2379 \
    --discovery ${DISCOVERY} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}
 
 
etcd --data-dir=data.etcd --name n2 \
    --initial-advertise-peer-urls http://10.240.0.18:2380 --listen-peer-urls http://10.240.0.18:2380 \
    --advertise-client-urls http://10.240.0.18:2379 --listen-client-urls http://10.240.0.18:2379 \
    --discovery ${DISCOVERY} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}
 
 
etcd --data-dir=data.etcd --name n3 \
    --initial-advertise-peer-urls http://10.240.0.19:2380 --listen-peer-urls http://10.240.0.19:2380 \
    --advertise-client-urls http://10.240.0.19:2379 --listen-client-urls http:/10.240.0.19:2379 \
    --discovery ${DISCOVERY} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

到此etcd集群就搭建起來瞭，可以使用etcdctl來連接etcd。

export ETCDCTL_API=3
HOST_1=10.240.0.17
HOST_2=10.240.0.18
HOST_3=10.240.0.19
ENDPOINTS=$HOST_1:2379,$HOST_2:2379,$HOST_3:2379
 
etcdctl --endpoints=$ENDPOINTS member lis

Go語言操作etcd

這裡使用官方的etcd/clientv3包來連接etcd並進行相關操作。

安裝

go get go.etcd.io/etcd/clientv3

put和get操作

put命令用來設置鍵值對數據，get命令用來根據key獲取值。

package main
 
import (
    "context"
    "fmt"
    "time"
 
    "go.etcd.io/etcd/clientv3"
)
 
// etcd client put/get demo
// use 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, "q1mi", "dsb")
    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, "q1mi")
    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操作

watch用來獲取未來更改的通知。

package main
 
import (
    "context"
    "fmt"
    "time"
 
    "go.etcd.io/etcd/clientv3"
)
 
// watch demo
 
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:q1mi change
    rch := cli.Watch(context.Background(), "q1mi") // <-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中q1mi這個key的變化。

例如：我們打開終端執行以下命令修改、刪除、設置q1mi這個key。

etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 put q1mi "dsb2"
OK
 
etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 del q1mi
1
 
etcd> etcdctl.exe --endpoints=http://127.0.0.1:2379 put q1mi "dsb3"
OK

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

watch>watch.exe
connect to etcd success
Type: PUT Key:q1mi Value:dsb2
Type: DELETE Key:q1mi Value:
Type: PUT Key:q1mi Value:dsb3

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(), "/nazha/", "dsb", 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(), "/nazha/", "dsb", 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)
    }
}

基於etcd實現分佈式鎖

go.etcd.io/etcd/clientv3/concurrency在etcd之上實現並發操作，如分佈式鎖、屏障和選舉。

導入該包：

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

基於etcd實現的分佈式鎖示例：

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

查看文檔瞭解更多

其他操作

其他操作請查看etcd/clientv3官方文檔。

參考鏈接：

https://etcd.io/docs/v3.3.12/demo/

https://www.infoq.cn/article/etcd-interpretation-application-scenario-implement-principle/ 代碼改變世界，腳踏實地，python、Golang。

到此這篇關於Go操作etcd的實現示例的文章就介紹到這瞭,更多相關Go操作etcd內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

Go操作etcd的實現示例

目錄

etcdetcd介紹

etcd應用場景

服務發現

配置中心

分佈式鎖

為什麼用 etcd 而不用ZooKeeper？

為什麼不選擇ZooKeeper？

為什麼選擇etcd？

etcd集群

搭建一個3節點集群示例：

Go語言操作etcd

安裝

put和get操作

watch操作

基於etcd實現分佈式鎖

參考鏈接：

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

etcdetcd介紹

etcd應用場景

服務發現

配置中心

分佈式鎖

為什麼用 etcd 而不用ZooKeeper？

為什麼不選擇ZooKeeper？

為什麼選擇etcd？

etcd集群

搭建一個3節點集群示例：

Go語言操作etcd

安裝

put和get操作

watch操作

基於etcd實現分佈式鎖

參考鏈接：

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆