ETCD

etc在linux系统中是配置文件目录名；etcd就是配置服务，用于提高高可用、强一致性的小型kv数据存储服务，用于解决集群管理系统中os升级时的分布式并发控制、配置文件的存储与分发等问题。
注意ETCD不是用来存储数据的，它只是同来提供一个高可用、强一致性的小型存储重要数据的kv系统。
体系结构

提升写磁盘的效率：wal预写式日志实现，执行写操作前先写日志，跟mysql中redo类似，wal实现的是顺序写，若直接按照B+数写，则涉及到多次io及随机写，效率低下。所以先写wal日志，再异步落盘。
保证一致：snapshot快照数据，用于其他节点同步主节点数据从而达到一致地状态；类似redis中主从复制中rdb数据恢复。1. leader生成snapshot；2. leader向follower发送snapshot；3. follower接收并应用snapshot；
实现KV存储：boltdb为每一个key都创建一个索引（B+树），该B+树存储了key所对应的版本数据。

Go操作ETCD

基础操作

package main

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

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

func main() {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"}, // 如果连接集群，需要指定多个
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer cli.Close()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	_, err = cli.Put(ctx, "key", "lin") // 相当于命令 etcdctl put key lin
	cancel()
	if err != nil {
		log.Fatal(err)
	}
	ctx, cancel = context.WithTimeout(context.Background(), time.Second)
	resp, err := cli.Get(ctx, "key") // 相当于命令 etcdctl get key
	cancel()
	if err != nil {
		log.Fatal(err)
	}
	for _, ev := range resp.Kvs {
		fmt.Printf("%s:%s\n", ev.Key, ev.Value)
	}
}

Watch

package main

import (
	"context"
	"fmt"
	"log"

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

func main() {
	cli, err := clientv3.NewFromURL("127.0.0.1:2379")
	if err != nil {
		log.Fatal(err)
	}
	defer cli.Close()

	// watch := cli.Watch(context.Background(),  clientv3.WithPrefix()) // 相当于命令 etcdctl watch key --prefix，只要是有key前缀都观察
	watch := cli.Watch(context.Background(), "key") // 相当于命令 etcdctl watch key
	for resp := range watch {                       // 会阻塞当前协程
		for _, ev := range resp.Events { // 起另一个客户端 etcdctl put key lin2
			fmt.Printf("Type :%s Key :%s Value :%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value) // 将观察到key的变动
		}
	}

}

Lease

package main

import (
	"context"
	"fmt"
	"log"

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

// 为什么使用租约？使大量相同过期时间的key同属于一个实体，方便管理。

func main() {
	cli, err := clientv3.NewFromURL("127.0.0.1:2379")
	if err != nil {
		log.Fatal(err)
	}
	defer cli.Close()
	// 相当于命令 etcdctl lease grant 1000 创建一个租约
	lease, err := cli.Grant(context.Background(), 5) // 以秒为单位
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("lease id", lease.ID)
	// etcdctl put key val --lease=216542dfsdf56dfe 创建一个key-value并将其绑定在租约上
	// lease revoke 216542dfsdf56dfe 删除该租约，即删除所有绑定在该租约上的kv对
	// lease 216542dfsdf56dfe ttl 可以观察到被删除的租约已经过期了
	_, err = cli.Put(context.Background(), "key", "mark", clientv3.WithLease(lease.ID))
	if err != nil {
		log.Fatal(err)
	}

	if false {
		// 给指定lease续租 etcdctl lease keep-alive 216542dfsdf56dfe
		// 这个接口会阻塞住，即长期不停的续租
		ch, err := cli.KeepAlive(context.Background(), lease.ID)
		if err != nil {
			log.Fatal(err)
		}
		for {
			recv := <-ch
			fmt.Println("time to live", recv.TTL)
		}
	}
	if false {
		// 只续约一次
		res, err := cli.KeepAliveOnce(context.Background(), lease.ID)
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println("time to live", res.TTL)
	}
}

Lock

package main

import (
	"context"
	"fmt"
	"log"

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

// 分布式锁的使用
func main() {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints: []string{"127.0.0.1:2379"},
	})
	if err != nil {
		log.Fatal(err)
	}
	defer cli.Close()

	s1, err := concurrency.NewSession(cli, concurrency.WithContext(context.Background()))
	if err != nil {
		log.Fatal(err)
	}
	defer s1.Close()
	m1 := concurrency.NewMutex(s1, "lock")
	s2, err := concurrency.NewSession(cli, concurrency.WithContext(context.Background()))
	if err != nil {
		log.Fatal(err)
	}
	defer s2.Close()
	m2 := concurrency.NewMutex(s1, "lock")
	if err := m1.Lock(context.Background()); err != nil {
		log.Fatal(err)
	}
	fmt.Println("s1 acquired lock")

	m2ch := make(chan struct{})
	go func() {
		defer close(m2ch)
		if err := m2.Lock(context.Background()); err != nil {
			log.Fatal(err)
		}
	}()
	if err := m1.Unlock(context.Background()); err != nil {
		log.Fatal(err)
	}
	fmt.Println("s1 released lock")
	<-m2ch
	fmt.Println("s2 acquired lock")
}

etcd实现的分布式锁是基于什么实现的？
1. 租约 + keepalive + etcd事务

存储原理

etcd 为每个key创建一个索引；一个索引对应着一个B+树；B+树key为 revision，B+节点存储的值为value；B+树存储着key的版本信息从而实现了etcd的mvcc；etcd不会任由版本信息膨胀，通过定期的compaction 来清理历史数据；
为了加速索引数据，在内存中维持着一个B树；B树key为 key,value 为该key的 revision;
思考：为什么是B树而不是hash，因为etcd经常进行前缀查询，b树有助于加快该过程。
思考：mysql的mvcc 是通过什么实现的(undolog)?mysql B+树存储什么内容？
思考：mysql为了加快索引数据，采用什么数据结构？MySQL采用自适应hash来加速索引

读写机制

etcd 是串行写，并发读；
并发读写时（读写同时进行），读操作是通过B+树mmap访问磁盘数据；写操作走日志复制流程；可以得知如果此时读操作走B树出现脏读幻读问题；通过B+树访问磁盘数据其实访问的事务开始前的数据，由mysql 可重复读隔离级别下 MVCC 读取规则可知能避免脏读和幻读问题；
并发读时，可走内存B树；
思考：由上面的信息展开etcd的acid分析；

Raft共识算法

两个过程
1. leader选举
2. 日志复制
演示网站：thesecretlivesofdata.com/raft/

leader选举

有两个超时参数（选举超时、心跳超时）
三种节点状态，follower、candidate、leader
选举任期Term值
初始选举流程：
1. 最开始时，每个节点都是follower、选举任期Term==0并都随机(150ms~300ms)一个选举超时时间，第一个超时的节点，成为candidate状态，其Term+=1，并给自己投一票，其选举超时重新随机；
2. candidate节点向其他节点发送数据，拉票，每个节点手里只有一张选票；所有节点只能向candidate状态的节点投票。follower节点收到拉票信息后其Term+=1;follower节点投票后，返回一个信息给candidate，且follower的选举超时时间也重新随机；
3. candidate节点收到大多数选票（一般为>=N+1/2），成为leader状态。并开始发送心跳包（两个作用：1.确保网络畅通；2.同步数据；）给followers；
4. follower每次收到心跳包，都会重置选举超时和心跳超时数据；
leader宕机后选举流程：
1. follower收不到心跳包，那么其中有一个节点的选举超时最先超时，成为candidate状态，开始自动拉票；
2. 重复初始选举的流程；
当两个节点同时成为候选者且拉票平票or都没有达到大多数选票：
1. 重新等待新一轮的选举，即再等待下一次成为candidate的节点拉票。

日志复制

我们的所有数据变更都必须通过leader进行传播

客户端发送变更给leader，leader先记录日志，然后发送心跳包（含数据变更）给followers；
follower收到数据变更心跳包后，返回一个response给leader；
leader收到大多数的回复后，日志进行提交，并异步落盘；leader发送一个已经提交的心跳包同步包给followers；

通过任期机制确保网络分区下的强一致性

假设有5台机器ABCDE，其中AB和CDE进行了分隔，即AB间能正常通信，CDE能通信，而之间不能通信；
假设AB间选举出A为leader，CDE间选举出C为leader；
由于AB在被分割的情况下，能收到的大多数同步确认吧包不能满足要求，所以所有的客户端请求A的更改操作，都不能被A落盘；
在ABCDE联通后，A的任期小于C的任期，A将自动退化为Followers，且AB会收到C的同步心跳包，发生日志回滚，仍然确保了整个系统的强一致性；

2024-12-16 该篇文章被 Cleofwine 归为分类: Go云原生