etcd从入门到入睡(Go语言客户端)

安装etcd

这里我使用的Docker安装，命令如下：

docker pull quay.io/coreos/etcd:v3.5.1

或者参考此篇使用预编译好的二进制执行文件，指路：etcd安装。

搭建etcd集群

这里我使用docker-compose来搭建etcd集群。yaml文件如下：

services:
  etcd0:
    image: "quay.io/coreos/etcd:v3.5.1"
    container_name: etcd0
    ports:
      - "23800:2380"
      - "23790:2379"
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd0
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://10.0.18.154:23790
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd0:2380
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd0=http://etcd0:2380,etcd1=http://etcd1:2380,etcd2=http://etcd2:2380
      - ETCD_INITIAL_CLUSTER_STATE=new

  etcd1:
    image: "quay.io/coreos/etcd:v3.5.1"
    container_name: etcd1
    ports:
      - "23801:2380"
      - "23791:2379"
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd1
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://10.0.18.154:23791
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd1:2380
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd0=http://etcd0:2380,etcd1=http://etcd1:2380,etcd2=http://etcd2:2380
      - ETCD_INITIAL_CLUSTER_STATE=new

  etcd2:
    image: "quay.io/coreos/etcd:v3.5.1"
    container_name: etcd2
    ports:
      - "23802:2380"
      - "23792:2379"
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd2
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://10.0.18.154:23792
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd2:2380
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd0=http://etcd0:2380,etcd1=http://etcd1:2380,etcd2=http://etcd2:2380
      - ETCD_INITIAL_CLUSTER_STATE=new

 etcd的Put和Get示例

package main

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

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

var client *clientv3.Client

func NewClient() error {
	var err error
	client, err = clientv3.New(clientv3.Config{
		Endpoints:         []string{"localhost:23790", "localhost:23791", "localhost:23792"},
		DialTimeout:       5 * time.Second,
		DialKeepAliveTime: 1 * time.Minute,
	})
	if err != nil {
		fmt.Println("etcd new fail ", err)

		return err
	}

	return nil
}
func getKeyValByPrefix(prefix string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	resp, err := client.Get(ctx, prefix, clientv3.WithPrefix(),
		clientv3.WithSort(clientv3.SortByKey, clientv3.SortAscend))
	if err != nil {
		fmt.Printf("get from etcd failed, err:%v\n", err)
		return err
	}
	fmt.Println(len(resp.Kvs))
	for _, ev := range resp.Kvs {
		fmt.Println(string(ev.Key), string(ev.Value))
	}
	return nil
}

func setEtcdKeyVal(key, val string) {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	_, err := client.Put(ctx, key, val)
	if err != nil {
		panic(err)
	}
}

type Info struct {
	Name string `json:"name"`
	Age  int    `json:"age"`
}

func main() {
	NewClient()
	setEtcdKeyVal("/app/bussiness1/app_id", "afadfjal")
	setEtcdKeyVal("/app/bussiness2/app_id", "dfjfdalj")
	info := &Info{"Jack", 23}
	bytes, err := json.Marshal(info)
	if err != nil {
		panic(err)
	}
	setEtcdKeyVal("/app/bussiness1/user_info", string(bytes))
	getKeyValByPrefix("/app/bussiness1")
	getKeyValByPrefix("/app/bussiness2")
	getKeyValByPrefix("unexistKey")
}

etcd的lease使用

func getKeyVal(key string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	resp, err := client.Get(ctx, key)
	if err != nil {
		fmt.Printf("getKeyVal failed, err:%v\n", err)
		return err
	}
	fmt.Println(len(resp.Kvs))
	for _, ev := range resp.Kvs {
		fmt.Println(string(ev.Key), string(ev.Value))
	}
	return nil
}

func setKeyValWithLease(key, val string) {
	lease, err := client.Grant(context.Background(), 5)
	if err != nil {
		panic(err)
	}
	fmt.Println(lease.ID)
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	_, err = client.Put(ctx, key, val, clientv3.WithLease(lease.ID))
	if err != nil {
		panic(err)
	}
    // 可以正常获取到指定key-value
	getKeyVal(key)
    // lease过期，key被删除，无法获取到指定key-value
	time.Sleep(6 * time.Second)
	getKeyVal(key)
}

etcd的lease保活

永久保活

func keepAliveKey(key, val string) {
	lease, err := client.Grant(context.Background(), 5)
	if err != nil {
		panic(err)
	}
	fmt.Println(lease.ID)
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	_, err = client.Put(ctx, key, val, clientv3.WithLease(lease.ID))
	if err != nil {
		panic(err)
	}
	kaCh, err := client.KeepAlive(context.Background(), lease.ID)
	if err != nil {
		panic(err)
	}
	kaResp := <-kaCh
	fmt.Println("ttl 1:", kaResp.TTL)
	getKeyVal(key)
	time.Sleep(6 * time.Second)
	fmt.Println("ttl 2:", kaResp.TTL)
	getKeyVal(key)
	kaResp = <-kaCh
	fmt.Println("ttl 3:", kaResp.TTL)
	getKeyVal(key)
	for kaResp := range kaCh {
		fmt.Println("loop ttl: ", kaResp.TTL)
		getKeyVal(key)
	}
}

保活一次 

func keepAliveOnceKey(key, val string) {
	lease, err := client.Grant(context.Background(), 5)
	if err != nil {
		panic(err)
	}
	fmt.Println(lease.ID)
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	_, err = client.Put(ctx, key, val, clientv3.WithLease(lease.ID))
	if err != nil {
		panic(err)
	}
	getKeyVal(key)
	time.Sleep(4 * time.Second)
	getKeyVal(key)
	kaResp, err := client.KeepAliveOnce(context.Background(), lease.ID)
	if err != nil {
		panic(err)
	}
	fmt.Println("ttl: ", kaResp.TTL)
	time.Sleep(3 * time.Second)
	// key还在，正常输出key-value；如果没有keep alive once,这里不会打印出key
	getKeyVal(key)
	time.Sleep(4 * time.Second)
	// lease过期，key已被删除
	getKeyVal(key)
}

 etcd的watch用法

func watchKey(key string) {
	wch := client.Watch(context.Background(), key)
	for wchResp := range wch {
		for _, item := range wchResp.Events {
			fmt.Println(item.Type, string(item.Kv.Key), string(item.Kv.Value))
		}
	}
}

func watchPrefix(prefix string) {

	for i := 1; i <= 3; i++ {
		go func(i int) {
			for j := 0; j < 5; j++ {
				setKeyVal(fmt.Sprintf("name%v", i), fmt.Sprintf("value%v", j))
				time.Sleep(800 * time.Microsecond)
			}
		}(i)
	}

	wch := client.Watch(context.Background(), prefix, clientv3.WithPrefix())
	for wchResp := range wch {
		for _, item := range wchResp.Events {
			fmt.Println(item.Type, string(item.Kv.Key), string(item.Kv.Value))
		}
		fmt.Println("-----")
	}
}