etcd (一看就会)

很有前景的一个项目，好多大公司的首选

古城小栈

2767人浏览 · 2022-03-07 11:49:42

古城小栈 · 2022-03-07 11:49:42 发布

简介

“etcd”这个名字源于两个想法，即　unix “/etc” 文件夹和分布式系统”d”istibuted。 “/etc” 文件夹为单个系统存储配置数据的地方，而 etcd 存储大规模分布式系统的配置信息。因此，”d”istibuted　的 “/etc” ，是为 “etcd”。

etcd 是 CoreOS 团队于 2013 年 6 月发起的开源项目，
采用 Go语言开发的一个高可用的分布式键值(key-value)存储系统，可以用于配置共享和服务的注册和发现。
内部采用 raft 协议作为 一致性算法。
etcd 目前默认使用 2379 端口提供 HTTP API 服务，2380 端口和 peer 通信（这两个端口已经被 IANA 官方预留给 etcd ）。
.
etcd具有以下特点：

完全复制：集群中的每个节点都可以使用完整的存档
高可用性：Etcd可用于避免硬件的单点故障或网络问题
一致性：每次读取都会返回跨多主机的最新写入
简单：包括一个定义良好、面向用户的API（gRPC）
安全：实现了带有可选的客户端证书身份验证的自动化TLS
快速：每秒10000次写入的基准速度
可靠：使用Raft算法实现了强一致、高可用的服务存储目录

类似项目有 zookeeper 和 consul.

应用场景

服务发现

服务发现要解决的是分布式系统中最常见的问题之一，
即在同一个分布式集群中的进程或服务，要如何才能找到对方并建立连接。
本质上来说: 服务发现就是要知道集群中是否有进程在监听 udp 或 tcp 端口，并且通过名字就可以查找和连接。

配置中心

将一些配置信息放到 etcd 上进行集中管理。
这类场景的使用方式通常是这样：

应用在启动的时候主动从 etcd 获取一次配置信息，同时，在 etcd 节点上注册一个 Watcher 并等待，
以后每次配置有更新的时候，etcd 都会实时通知订阅者，以此达到获取最新配置信息的目的。

分布式锁

因为 etcd 使用 Raft 算法 保持了数据的强一致性，某次操作存储到集群中的值必然是全局一致的，
所以很容易实现分布式锁。
锁服务有两种使用方式，一是保持独占，二是控制时序。

保持独占
所有获取锁的用户最终只有一个可以得到。
控制时序
所有想要获得锁的用户都会被安排执行，但是获得锁的顺序也是全局唯一的，同时决定了执行顺序。

对比其他

etcd 和 redis

etcd是一种分布式存储，更强调的是各个节点之间的通信，同步，确保各个节点上数据和事务的一致性，
使得服务发现工作更稳定，本身单节点的写入能力并不强。

redis更像是内存型缓存，虽然也有 cluster 做主从同步和读写分离，
但节点间的一致性主要强调的是数据，并不在乎事务，

因此日常使用中，etcd主要还是做一些事务管理类的，基础架构服务用的比较多，容器类的服务部署是其主流。
而redis广泛地使用在 缓存服务器方面，用作mysql的缓存，通常依据请求量，甚至会做成多级缓存。

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。

安装

ubuntu

sudo apt install etcd

测试安装是否成功：

# 测试是否安装成功
etcd --version

# 测试etcd服务是否开启 
curl http://127.0.0.1:2379/version

因为 etcd 是 go 语言编写的，也可以通过下载对应的二进制文件进行安装，该方式以 centos 下举例：

# 1.下载
 
wget https://github.com/etcd-io/etcd/releases/download/v3.4.6/etcd-v3.4.6-linux-amd64.tar.gz
 
# 2.解压
tar -zxf etcd-v3.4.6-linux-amd64.tar.gz
 
# 3.移动到bin目录
mv etcd-v3.4.6-linux-amd64 /usr/local/etcd
 
# 4.开启端口 2379 -需要外部访问的话可以
firewall-cmd --zone=public --add-port=2379/tcp --permanent
firewall-cmd --reload
firewall-cmd --list-all
 
# 5.启动
cd /usr/local/etcd
./etcd --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://0.0.0.0:2379

基本使用

HTTP API

etcd 对外通过 HTTP API 对外提供服务，这种方式方便测试（通过 curl工具就能和 etcd 交互），
也很容易集成到各种语言中（每个语言封装 HTTP API 实现自己的 client 就行）。
这个部分，我们就介绍 etcd 通过 HTTP API 提供了哪些功能.

增 / 改

#  键message 值"Hello word"
curl http://127.0.0.1:2379/v2/keys/message -XPUT -d value="Hello world"

#  键message 值"Hello word"
curl http://127.0.0.1:2379/v2/keys/message -XDELETE

#  键message 值"Hello word"
curl http://127.0.0.1:2379/v2/keys/message

短期存储：

#  5秒后 foo将被销除
curl http://127.0.0.1:2379/v2/keys/foo -XPUT -d value=bar -d ttl=5

etcdctl 命令工具

使用 etcd 自带的工具能方便地对数据进行相应操作

# 初次使用
hero@hlly:~$ etcdctl
NAME:
   etcdctl - A simple command line client for etcd.

# 这里有一个警告：默认使用的是 API 版本是2， 推荐使用3， 根据警告进行 如下操作
WARNING:
   Environment variable ETCDCTL_API is not set; defaults to etcdctl v2.
   Set environment variable ETCDCTL_API=3 to use v3 API or ETCDCTL_API=2 to use v2 API.

USAGE:
   etcdctl [global options] command [command options] [arguments...]

VERSION:
   3.2.26
......

# 使用 API 版本3 的操作：
vim /etc/profile 
# 将一下内容加入该文件
export ETCDCTL_API=3
# source 一下， 
source /etc/profile 

# 二次使用
root@hlly:/home/hero# etcdctl
NAME:
        etcdctl - A simple command line client for etcd3.

USAGE:
        etcdctl [flags]

VERSION:
        3.2.26
# 这里可以看到， 使用 API 版本3
API VERSION:
        3.2
......


# 常用操作：
hero@hlly:~$ etcdctl get name
tom

hero@hlly:~$ etcdctl update name "cat"
cat

hero@hlly:~$ etcdctl del name

etcd集群

每个 etcd cluster 都是有若干个 member 组成的，每个 member 是一个独立运行的 etcd 实例，
单台机器上可以运行多个 member。在正常运行的状态下，集群中会有一个 leader，其余的 member 都是 followers。
leader 向 followers 同步日志，保证数据在各个 member 都有副本。leader 还会定时向所有的 member 发送心跳报文，
如果在规定的时间里 follower 没有收到心跳，就会重新选举。
客户端所有的请求都会先发送给 leader，leader 向所有的 followers 同步日志，
等收到超过半数的确认后就把该日志存储到磁盘，并返回响应客户端。

举例说明

搭建一个3节点集群示例, 为了区分不同的集群最好同时配置一个独一无二的token,
和另外两个数据，我们将这三个定义为 shell 变量：

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集群就搭建起来了，可以使用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 list

Go 操作 etcd

我们不使用 etcd/clientv3，因为它与grpc 最新版本不兼容，
这里我们使用官方最新推荐的方式 etcd/client/v3

go get go.etcd.io/etcd/client/v3


# go.mod
module go_etcd_test

go 1.18

require (
	go.etcd.io/etcd v2.3.8+incompatible
	go.etcd.io/etcd/client/v3 v3.5.5
)

Put 和 Get 操作

put 设置键值对数据，
get 根据key获取值。

package main

import (
	"context"
	"fmt"
	"go.etcd.io/etcd/client/v3"
	"log"
	"time"
)

func main() {

	client, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatalln(err)
	}

	// PUT
	ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)
	defer cancel()
	key := "/ns/service"
	value := "127.0.0.1:8000"
	_, err = client.Put(ctx, key, value)
	if err != nil {
		log.Printf("etcd put error,%v\n", err)
		return
	}

	// GET
	getResponse, err := client.Get(ctx, key)
	if err != nil {
		log.Printf("etcd GET error,%v\n", err)
		return
	}

	for _,kv := range getResponse.Kvs {
		fmt.Printf("Key:%s ===> Val:%s \n", kv.Key, kv.Value)
	}
}

watch 操作

对指定的键的 PUT、DELETE 等操作进行监听，用来获取未来更改的通知

package main

import (
	"context"
	"fmt"
	"go.etcd.io/etcd/client/v3"
	"log"
	"strconv"
	"time"
)

func main() {

	client, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatalln(err)
	}

    // 监听变化
	go watcher(client,key)
	

	key := "/ns/service"
	value := "this value is "
	ctx := context.Background()

	// 每隔2秒重新PUT一次
	for i:=0 ;i<100;i++ {
		time.Sleep(2*time.Second)
		_, err := client.Put(ctx, key, value+strconv.Itoa(i))
		if err != nil {
			log.Printf("put error %v",err)
		}
	}

}


func watcher(client *clientv3.Client,key string) {
	// 监听这个chan
	watchChan := client.Watch(context.Background(), key)

	for watchResponse := range watchChan {

		for _, event := range watchResponse.Events {
			fmt.Printf("操作时间类型:%s,Key:%s,Value:%s\n",event.Type, event.Kv.Key, event.Kv.Value)
		}
	}
}

租约

设置KEY-VALUE的过期时间，这里引入一个概念叫租约，将设置过期时间这一过程称为租约，租约期限到期则KEY-VALUE将删除

package main

import (
	"context"
	"fmt"
	"go.etcd.io/etcd/client/v3"
	"log"
	"time"
)

func main() {

	client, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatalln(err)
	}

	key := "/ns/service"
	value := "127.0.0.1:800"
	ctx := context.Background()

	// 获取一个租约 有效期为5秒
	leaseGrant, err := client.Grant(ctx, 5)
	if err != nil {
		log.Printf("put error %v",err)
		return
	}

	// PUT 租约期限为5秒
	_, err = client.Put(ctx, key, value, clientv3.WithLease(leaseGrant.ID)) // here
	if err != nil {
		log.Printf("put error %v",err)
		return
	}

	// 监听变化 5秒后将监听到DELETE事件
	watcher(client,key)

}


func watcher(client *clientv3.Client,key string) {

	// 监听这个chan
	watchChan := client.Watch(context.Background(), key)

	for watchResponse := range watchChan {
		for _, event := range watchResponse.Events {
			fmt.Printf("Type:%s,Key:%s,Value:%s\n",event.Type,event.Kv.Key,event.Kv.Value)
			// Type:DELETE,Key:/ns/service,Value:
		}
	}

}

KeepAlive 续租

package main

import (
	"context"
	"fmt"
	"go.etcd.io/etcd/client/v3"
	"log"
	"time"
)

func main() {

	client, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatalln(err)
	}

	key := "/ns/service"
	value := "127.0.0.1:800"
	ctx := context.Background()

	// 获取一个租约 有效期为5秒
	leaseGrant, err := client.Grant(ctx, 5)
	if err != nil {
		log.Printf("grant error %v",err)
		return
	}

	// PUT 租约期限为5秒
	_, err = client.Put(ctx, key, value, clientv3.WithLease(leaseGrant.ID))
	if err != nil {
		log.Printf("put error %v",err)
		return
	}

	// 续租
	keepaliveResponseChan, err := client.KeepAlive(ctx, leaseGrant.ID)  // here
	if err != nil {
		log.Printf("KeepAlive error %v",err)
		return
	}
	
	for {
		ka := <-keepaliveResponseChan
		fmt.Println("ttl:", ka.TTL) // ttl: 5
		
	}
}

基于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")