etcd 存储：如何实现键值对的读写操作？
etcd 存储：如何实现键值对的读写操作？
你好，我是 aoho，今天我和你分享的主题是 etcd 存储：如何实现键值对的读写操作？
我们在前面课时介绍了 etcd 的整体架构以及 etcd 常用的通信接口。在介绍 etcd 整体架构时，我们梳理了 etcd 的分层架构以及交互概览。本课时将会聚焦于 etcd 存储是如何实现键值对的读写操作。
本课时围绕 etcd 底层读写的实现，首先会简要介绍客户端访问 etcd 服务端读写的整个过程，然后是重点介绍读写的实现细节。
读操作
在 etcd 中读请求占了大部分，是高频的操作。我们使用 etcdctl 命令行工具进行读操作：
$ etcdctl --endpoints http://localhost:2379 get foo
foo
bar
将整个读操作划分成如下几个步骤：
- etcdctl 会创建一个 clientv3 库对象，选取一个合适的 etcd 节点；
- 调用 KVServer 模块的 Range RPC 方法（上一课时有讲解），发送请求；
- 拦截器拦截，主要做一些校验和监控；
- 调用 KVServer 模块的 Range 接口获取数据；
接着就进入了读请求的核心步骤，会经过线性读 ReadIndex 模块、MVCC (包含 treeIndex 和 BlotDB）模块。
这里提一下线性读，线性读是相对串行读来说。集群模式下会有多个 etcd 节点，不同节点之间可能存在一致性问题，串行读直接返回状态数据，不需要与集群中其他节点交互。这种方式速度快，开销小，但是会存在数据不一致的情况。而线性读则需要集群成员之间达成共识，存在开销，响应速度相对慢。但是能够保证数据的一致性，etcd 默认读模式是线性读。我们将在后面的课时重点介绍分布式一致性的实现。
继续往下，看看如何读取 etcd 中的数据。etcd 中查询请求，查询单个键或者一组键，亦或是查询数量，到了底层实际都会调用 rangeKeys 方法，我们来分析下这个方式的实现。range 请求的结构图如下所示：
从上至下，查询键值对的流程包括：
- 在 treeIndex 中根据键利用 BTree 快速查询该键对应的索引项 keyIndex，索引项中包含 Revision；
- 根据查询到的版本号信息 Revision，在 Backend 的缓存 buffer 中利用二分法查找，如果命中则直接返回；
- 若缓存中不符合条件，在 BlotDB 中查找（基于 BlotDB 的索引），查询之后返回键值对信息。
图中的 ReadTx 和 BatchTx 是两个接口，用于负责读写请求。我们在创建 backend 结构时，都会创建 readTx 和 batchTx 结构体，这两个结构体分别实现了 ReadTx 和 BatchTx 接口，一个负责处理只读请求，另一个负责处理读写请求。。
rangeKeys 方法的实现如下所示：
// 位于 mvcc/kvstore_txn.go:117
func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions) (*RangeResult, error) {
rev := ro.Rev
if rev > curRev {
return &RangeResult{KVs: nil, Count: -1, Rev: curRev}, ErrFutureRev
}
if rev <= 0 {
rev = curRev
}
if rev < tr.s.compactMainRev {
return &RangeResult{KVs: nil, Count: -1, Rev: 0}, ErrCompacted
}
// 获取索引项 keyIndex，索引项中包含 Revision
revpairs := tr.s.kvindex.Revisions(key, end, rev)
tr.trace.Step("range keys from in-memory index tree")
// 结果为空，直接返回
if len(revpairs) == 0 {
return &RangeResult{KVs: nil, Count: 0, Rev: curRev}, nil
}
if ro.Count {
return &RangeResult{KVs: nil, Count: len(revpairs), Rev: curRev}, nil
}
limit := int(ro.Limit)
if limit <= 0 || limit > len(revpairs) {
limit = len(revpairs)
}
kvs := make([]mvccpb.KeyValue, limit)
revBytes := newRevBytes()
for i, revpair := range revpairs[:len(kvs)] {
revToBytes(revpair, revBytes)
// UnsafeRange 实现了 ReadTx，查询对应的键值对
_, vs := tr.tx.UnsafeRange(keyBucketName, revBytes, nil, 0)
if len(vs) != 1 {
tr.s.lg.Fatal(
"range failed to find revision pair",
zap.Int64("revision-main", revpair.main),
zap.Int64("revision-sub", revpair.sub),
)
}
if err := kvs[i].Unmarshal(vs[0]); err != nil {
tr.s.lg.Fatal(
"failed to unmarshal mvccpb.KeyValue",
zap.Error(err),
)
}
}
tr.trace.Step("range keys from bolt db")
return &RangeResult{KVs: kvs, Count: len(revpairs), Rev: curRev}, nil
}
在上述代码的实现中，我们需要通过 Revisions 方法从 Btree 中获取范围内所有的 keyIndex，以此才能获取一个范围内的所有键值对。Revisions 方法实现如下：
// 位于 mvcc/index.go:106
func (ti *treeIndex) Revisions(key, end []byte, atRev int64) (revs []revision) {
if end == nil {
rev, _, _, err := ti.Get(key, atRev)
if err != nil {
return nil
}
return []revision{rev}
}
ti.visit(key, end, func(ki *keyIndex) {
// 使用 keyIndex.get 来遍历整颗树
if rev, _, _, err := ki.get(ti.lg, atRev); err == nil {
revs = append(revs, rev)
}
})
return revs
}
如果只是获取一个键对应的版本，使用 treeIndex 方法即可，但是一般会从 Btree 索引中获取多个 Revision 值时，需要调用 keyIndex.get 方法来遍历整颗树并选取合适的版本。这是因为 BoltDB 保存一个 key 的多个历史版本。每一个 Key 的 keyIndex 中其实都存储着多个历史版本，我们需要根据传入的参数返回正确的版本。
对于上层的键值存储来说，它会利用这里返回的 Revision 从真正存储数据的 BoltDB 中查询当前 Key 对应 Revision 的结果。BoltDB 内部使用的也是类似 bucket 的方式存储，其实就是对应 MySQL 中的表结构，用户的 key 数据存放的 bucket 名字的是 key，etcd MVCC 元数据存放的 bucket 是 meta。
写操作
介绍完读请求，我们回忆一下写操作的实现。使用 etcdctl 命令行工具进行写操作：
$ etcdctl --endpoints http://localhost:2379 put foo bar
将整个写操作划分成如下几个步骤：
- 客户端通过负载均衡算法选择一个 etcd 节点，发起 gRPC 调用；
- etcd server 收到客户端请求；
- 经过 gRPC 拦截、Quota 校验，Quota 模块用于校验 etcd db 文件大小是否超过了配额；
- 接着 KVServer 模块将请求发送给本模块中的 raft，这里负责与 etcd raft 模块进行通信。发起一个提案，命令为
put foo bar，即使用 put 方法将 foo 更新为 bar； - 提案经过转发之后，半数节点成功持久化；
- MVCC 模块更新状态机。
我们重点关注最后一步，学习如何更新和插入键值对。与上面一张图相对应，我们来看下 put 接口的执行过程：
调用 put 向 etcd 写入数据时，首先会使用传入的键构建 keyIndex 结构体，基于 currentRevision 自增生成新的 revision 如{1,0}，并从 treeIndex 中获取相关版本 revision 等信息；写事务提交之后，将本次写操作的缓存 buffer 合并到（merge）到读缓存上（图中 ReadTx 中的缓存）。代码实现如下所示：
//位于 mvcc/index.go:53
func (ti *treeIndex) Put(key []byte, rev revision) {
keyi := &keyIndex{key: key}
// 加锁，互斥
ti.Lock()
defer ti.Unlock()
// 获取版本信息
item := ti.tree.Get(keyi)
if item == nil {
keyi.put(ti.lg, rev.main, rev.sub)
ti.tree.ReplaceOrInsert(keyi)
return
}
okeyi := item.(*keyIndex)
okeyi.put(ti.lg, rev.main, rev.sub)
}
treeIndex.Put 在获取 Btree 中的 keyIndex 结构之后，会通过 keyIndex.put 在其中加入新的 revision，方法实现如下：
// 位于 mvcc/key_index.go:77
func (ki *keyIndex) put(lg *zap.Logger, main int64, sub int64) {
rev := revision{main: main, sub: sub}
// 校验版本号
if !rev.GreaterThan(ki.modified) {
lg.Panic(
"'put' with an unexpected smaller revision",
zap.Int64("given-revision-main", rev.main),
zap.Int64("given-revision-sub", rev.sub),
zap.Int64("modified-revision-main", ki.modified.main),
zap.Int64("modified-revision-sub", ki.modified.sub),
)
}
if len(ki.generations) == 0 {
ki.generations = append(ki.generations, generation{})
}
g := &ki.generations[len(ki.generations)-1]
if len(g.revs) == 0 { // 创建一个新的键
keysGauge.Inc()
g.created = rev
}
g.revs = append(g.revs, rev)
g.ver++
ki.modified = rev
}
新的 revision 结构体写入 keyIndex 键值索引时，都会改变 generation 结构体中的创建版本 、修改次数等参数，因此，基于 put 方法，我们就可以知道 generation 结构体中的各个成员如何定义和赋值。
revision{1,0} 是生成的全局版本号，作为 BoltDB 的 key，经过序列化包括 key 名称、key 创建时的版本号（create_revision）、value 值和租约等信息为二进制数据之后，将填充到 BoltDB 的 value 中，同时将该键和 revision 等信息存储到 Btree。
小结
本文主要介绍了 etcd 的底层如何实现读写操作。我们首先简单介绍了客户端与服务端读写操作的流程，之后重点分析了在 etcd 中如何读写数据。
读写操作依赖 MVCC 模块的 treeIndex 和 BoltDB，treeIndex 是一个 内存索引模块，用来保存键的历史版本号信息；BoltDB 是一个基于 Btree 实现的数据库，可以用来保存 etcd 的键值对数据。通过这两个模块之间的协作，实现了 etcd 数据的读取和存储。
学习完本课时，给大家留个小问题，etcd 写事务的提交会涉及 B+ 重新平衡，这部分开销昂贵该如何权衡呢？欢迎你在留言区提出。
当然，本课时仅是介绍了底层的存储，对于如何实现分布式数据一致性并没有展开讲解。我们将在下一讲介绍 etcd-raft 如何实现分布式一致性。