背景

最近想找一些用Go语言实现的优秀开源项目学习一下，etcd作为一个被广泛应用的高可用、强一致性服务发现存储仓库，非常值得分析学习。
本篇文章主要是对etcd的后台存储源码做一解析，希望可以从中学到一些东西。

etcd大版本区别

目前etcd常用的是v2和v3两个大版本。两个版本不同之处主要在于：

1. v2版本仅在内存中对数据进行了存储，没有做持久化存储。而v3版本做了持久化存储，且还使用了缓存机制加快查询速度。
2. v2版本和v3版本对外提供的接口做了一些改变。在命令行界面中，可以使用环境变量ETCDCTL_API来设置对外接口。

我们在这里主要是介绍v3版本的后台存储部分实现。 并且这里仅涉及到底层的读写操作接口，并不涉及到更上层的读写步骤（键值的revision版本选择等）。

etcd的后端存储接口

分析思路：

1. 查看etcd封装的后端存储接口
2. 查看etcd实现了后端存储接口的结构体
3. 查看上述结构体的初始化方法
4. 查看上述结构体的初始化值
5. 查看上述结构体初始化方法的具体初始化过程

首先，我们先来看下etcd封装的后端存储接口：
路径：https://github.com/etcd-io/etcd/blob/master/mvcc/backend/backend.go

type Backend interface {
    // ReadTx returns a read transaction. It is replaced by ConcurrentReadTx in the main data path, see #10523.
    ReadTx() ReadTx
    BatchTx() BatchTx
    // ConcurrentReadTx returns a non-blocking read transaction.
    ConcurrentReadTx() ReadTx

    Snapshot() Snapshot
    Hash(ignores map[IgnoreKey]struct{}) (uint32, error)
    // Size returns the current size of the backend physically allocated.
    // The backend can hold DB space that is not utilized at the moment,
    // since it can conduct pre-allocation or spare unused space for recycling.
    // Use SizeInUse() instead for the actual DB size.
    Size() int64
    // SizeInUse returns the current size of the backend logically in use.
    // Since the backend can manage free space in a non-byte unit such as
    // number of pages, the returned value can be not exactly accurate in bytes.
    SizeInUse() int64
    // OpenReadTxN returns the number of currently open read transactions in the backend.
    OpenReadTxN() int64
    Defrag() error
    ForceCommit()
    Close() error
}

Backend接口封装了etcd后端所提供的接口，最主要的是：
ReadTx()，提供只读事务的接口，以及BatchTx()，提供读写事务的接口。
Backend作为后端封装好的接口，而backend结构体则实现了Backend接口。
路径：https://github.com/etcd-io/etcd/blob/master/mvcc/backend/backend.go

type backend struct {
    // size and commits are used with atomic operations so they must be
    // 64-bit aligned, otherwise 32-bit tests will crash

    // size is the number of bytes allocated in the backend
    // size字段用于存储给后端分配的字节大小
    size int64
    // sizeInUse is the number of bytes actually used in the backend
    // sizeInUse字段是后端实际上使用的内存大小
    sizeInUse int64
    // commits counts number of commits since start
    // commits字段用于记录启动以来提交的次数
    commits int64
    // openReadTxN is the number of currently open read transactions in the backend
    // openReadTxN存储目前读取事务的开启次数
    openReadTxN int64

    // mu是互斥锁
    mu sync.RWMutex
    // db表示一个boltDB实例，此处可以看到，Etcd默认使用Bolt数据库作为底层存储数据库
    db *bolt.DB

    // 用于读写操作
    batchInterval time.Duration
    batchLimit    int
    batchTx       *batchTxBuffered

    // 该结构体用于只读操作,Tx表示transaction
    readTx *readTx

    stopc chan struct{}
    donec chan struct{}

    // 日志信息
    lg *zap.Logger
}

通过19行 db *bolt.DB 我们可以看到，etcd的底层存储数据库为BoltDB。
好了，接下来我们就看一下这个backend结构体是如何初始化的。
还是在该路径下，我们可以看到New函数

// 创建一个新的backend实例
func New(bcfg BackendConfig) Backend {
    return newBackend(bcfg)
}

该函数传入了参数bcfg，类型为BackendConfig，这是后端存储的配置信息。
我们先看下这个配置信息中包含了什么
依然在该路径下，找到BackendConfig结构体

type BackendConfig struct {
    // Path is the file path to the backend file.
    Path string
    // BatchInterval is the maximum time before flushing the BatchTx.
    // BatchInterval表示提交事务的最长间隔时间
    BatchInterval time.Duration
    // BatchLimit is the maximum puts before flushing the BatchTx.
    BatchLimit int
    // BackendFreelistType is the backend boltdb's freelist type.
    BackendFreelistType bolt.FreelistType
    // MmapSize is the number of bytes to mmap for the backend.
    // MmapSize表示分配的内存大小
    MmapSize uint64
    // Logger logs backend-side operations.
    Logger *zap.Logger
    // UnsafeNoFsync disables all uses of fsync.
    UnsafeNoFsync bool `json:"unsafe-no-fsync"`
}

可以看到，有许多backend初始化所需要的信息都在这个结构体中。
既然有这些配置信息，那么一定会有相应的默认配置信息，
我们来看下在默认情况下etcd存储部分会被赋怎样的值。
依然在该目录下，找到DefaultBackendConfig函数。

func DefaultBackendConfig() BackendConfig {
    return BackendConfig{
    BatchInterval: defaultBatchInterval,
    BatchLimit: defaultBatchLimit,
    MmapSize: initialMmapSize,
    }
}

随便查看其中某个全局变量的值，比如defaultBatchInterval，则可以看到默认值：

var (
    defaultBatchLimit = 10000
    defaultBatchInterval = 100 * time.Millisecond
    defragLimit = 10000
    // initialMmapSize is the initial size of the mmapped region. Setting this larger than
    // the potential max db size can prevent writer from blocking reader.
    // This only works for linux.
    initialMmapSize = uint64(10 * 1024 * 1024 * 1024)
    // minSnapshotWarningTimeout is the minimum threshold to trigger a long running snapshot warning.
    minSnapshotWarningTimeout = 30 * time.Second
)

以defaultBatchInterval变量为例，就是说默认情况下，etcd会100秒做一次自动的事务提交。
etcd后端存储默认赋值的部分说完了，就说回对结构体的初始化上。
我们继续看函数New，它调用了函数newBackend，
我们看下函数newBackend做了些什么

func newBackend(bcfg BackendConfig) *backend {
    if bcfg.Logger == nil {
        bcfg.Logger = zap.NewNop()
    }

    // 一些配置载入
    bopts := &bolt.Options{}
    if boltOpenOptions != nil {
        *bopts = *boltOpenOptions
    }
    bopts.InitialMmapSize = bcfg.mmapSize()
    bopts.FreelistType = bcfg.BackendFreelistType
    bopts.NoSync = bcfg.UnsafeNoFsync
    bopts.NoGrowSync = bcfg.UnsafeNoFsync

    // 初始化Bolt数据库
    db, err := bolt.Open(bcfg.Path, 0600, bopts)
    if err != nil {    
        bcfg.Logger.Panic("failed to open database", zap.String("path", bcfg.Path), zap.Error(err))
    }

    // In future, may want to make buffering optional for low-concurrency systems
    // or dynamically swap between buffered/non-buffered depending on workload.
    // 对backend结构体做初始化，包括了readTx只读事务以及batchTx读写事务
    b := &backend{
        db: db,

        batchInterval: bcfg.BatchInterval,
        batchLimit: bcfg.BatchLimit,

        readTx: &readTx{
            baseReadTx: baseReadTx{
                buf: txReadBuffer{
                    txBuffer: txBuffer{make(map[string]*bucketBuffer)},
                },
                buckets: make(map[string]*bolt.Bucket),
                txWg: new(sync.WaitGroup),
                txMu: new(sync.RWMutex),
            },
        },

        stopc: make(chan struct{}),
        donec: make(chan struct{}),

        lg: bcfg.Logger,
    }
    b.batchTx = newBatchTxBuffered(b)
    // 开启一个新的etcd后端存储连接
    go b.run()
    return b
}

我们可以看到6-19行在初始化boltDB的同时载入了一些数据库的配置信息。
23-41行是对backend结构体做了初始化，包括了只读事务readTx、读写事务batchTx结构体的初始化，以及初始化了两个通道stopc、donec，这个后面会用到。
43行开启了一个协程去并发的处理run()函数内的工作。
我们继续看一下run()函数做了什么。依然在该目录下

func (b *backend) run() {
    // 关闭结构体的donec通道
    defer close(b.donec)
    // 开启一个定时器
    t := time.NewTimer(b.batchInterval)
    // 最后要关闭定时器
    defer t.Stop()
    for {
        select {
            // 当定时器到时间了，则t.C会有值
            case <-t.C:
            case <-b.stopc:
                b.batchTx.CommitAndStop()
                return
            }
        // 定时器到时间了，且数据的偏移量非0，即有数据的情况下，则会进行一次事务的自动提交
        if b.batchTx.safePending() != 0 {
            b.batchTx.Commit()
        }
        // 重新设置定时器的时间
        t.Reset(b.batchInterval)    
    }
}

我在代码中注释的比较详细了，简单的说，就是在初始化backend结构体时，开启了一个协程用于事务的自动提交，事务自动提交的时间间隔为batchInterval，这个默认值为100秒。
注意12-14行，这段代码表示，如果是停止信号进来的话，则事务会立即提交并且停止。
到这里，backend结构体就初始化完成了，接下来我们看一下用于读操作的只读事务接口ReadTx

etcd后端存储的读操作

路径：https://github.com/etcd-io/etcd/blob/master/mvcc/backend/read_tx.go

type ReadTx interface {
    Lock()
    Unlock()
    RLock()
    RUnlock()

    UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte)
    UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error
}

该接口是用结构体baseReadTx实现的，来看一下baseReadTx结构体，文件路径与ReadTx接口一样

type baseReadTx struct {
    // buf与buckets都是用于增加读效率的缓存
    // mu用于保护txReadBuffer缓存的操作
    mu  sync.RWMutex
    buf txReadBuffer

    // txMu用于保护buckets缓存和tx的操作
    txMu    *sync.RWMutex
    tx      *bolt.Tx
    buckets map[string]*bolt.Bucket
    // txWg可以防止tx在批处理间隔结束时回滚，直到使用该tx完成所有读取为止
    txWg *sync.WaitGroup
}

只读事务ReadTx的读取数据的接口有两个，分别是UnsafeRange以及UnsafeForEach。我们以UnsafeRange接口为例进行代码分析。
UnsafeRange接口的实现依然在上述路径中

// 该方法用于底层数据的只读操作
func (baseReadTx *baseReadTx) UnsafeRange(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
    // 不使用范围查询
    if endKey == nil {
        // forbid duplicates for single keys
        limit = 1
    }
    // 当范围值异常时，则传入最大范围
    if limit <= 0 {
        limit = math.MaxInt64
    }
    if limit > 1 && !bytes.Equal(bucketName, safeRangeBucket) {
        panic("do not use unsafeRange on non-keys bucket")
    }
    // 将buf缓存中的数据读取出来
    keys, vals := baseReadTx.buf.Range(bucketName, key, endKey, limit)
    // 如果取出的数据满足了需求，那么则直接返回数据
    if int64(len(keys)) == limit {
        return keys, vals
    }

    // find/cache bucket
    // 从bucket缓存中查询bucket实例，查询到了则返回缓存中的实例，查询不到，则在BoltDB中查找
    bn := string(bucketName)
    baseReadTx.txMu.RLock()
    bucket, ok := baseReadTx.buckets[bn]
    baseReadTx.txMu.RUnlock()
    lockHeld := false
    // 缓存中取不到bucket的话，会从bolt中查找，并写入缓存中
    if !ok {
        baseReadTx.txMu.Lock()
        lockHeld = true
        bucket = baseReadTx.tx.Bucket(bucketName)
        baseReadTx.buckets[bn] = bucket
    }

    // ignore missing bucket since may have been created in this batch
    if bucket == nil {
        if lockHeld {
            baseReadTx.txMu.Unlock()
        }
        return keys, vals
    }
    if !lockHeld {
        baseReadTx.txMu.Lock()
        lockHeld = true
    }
    c := bucket.Cursor()
    baseReadTx.txMu.Unlock()

    // 从bolt的该bucket中查找键值对
    k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys)))
    return append(k2, keys...), append(v2, vals...)
}

4-14行是一些前置的判断步骤，16-20则是从buf缓存中读取数据，前面提到过，buf是etcd用于提高读取效率的缓存。
我们看下具体的从buf读取数据的过程。
Range函数在路径：https://github.com/etcd-io/etcd/blob/master/mvcc/backend/tx_buffer.go

func (txr *txReadBuffer) Range(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
    if b := txr.buckets[string(bucketName)]; b != nil {
        return b.Range(key, endKey, limit)
    }
    return nil, nil
}

可以看到，该方法就是实例化了名为bucketName的桶，然后从该桶中按照范围读取键值数据。
我们可以看到，bucket的实例为结构体bucketBuffer

type bucketBuffer struct {
    buf []kv
    // used字段记录了正在使用的元素个数，这样buf无需重新分配内存就可以覆盖写入
    used int
}

看回到Range方法代码的第3行，我们来看一下b.Range方法的代码。b.Range与buf.Range方法不同，b.Range是结构体bucketBuffer实现的方法。
依然与Range方法相同路径

func (bb *bucketBuffer) Range(key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
    // 查找到key在buf中的索引idx
    f := func(i int) bool { return bytes.Compare(bb.buf[i].key, key) >= 0 }
    // sort.Search用于从某个切片中查找某个值的索引
    idx := sort.Search(bb.used, f)
    if idx < 0 {
        return nil, nil
    }
    // 只查找一个key值，而非范围查找
    if len(endKey) == 0 {
        if bytes.Equal(key, bb.buf[idx].key) {
            keys = append(keys, bb.buf[idx].key)
            vals = append(vals, bb.buf[idx].val)
        }
        return keys, vals
    }
    // 根据字节的值来比较字节切片的大小
    // 如果endKey比key值小，则返回nil
    if bytes.Compare(endKey, bb.buf[idx].key) <= 0 {
        return nil, nil
    }
    // 在个数限制limit内，且小于endKey的所有键值对都取出来
    for i := idx; i < bb.used && int64(len(keys)) < limit; i++ {
        if bytes.Compare(endKey, bb.buf[i].key) <= 0 {
            break
        }
        keys = append(keys, bb.buf[i].key)
        vals = append(vals, bb.buf[i].val)
    }
    return keys, vals
}

3-5行代码表示要从buf结构体中找到第一个满足包含key的索引值。该两行代码一般结合使用，是一种常见的用于查找值对应索引值的方式。

10-16行代码表示，如果endKey为0，即不使用范围查找，只查找key这一个精确值，那么就需要判断3-5代码找到的值是否与该key完全相等，只有完全相等了才会返回keys与vals。

19-21行代码表示，如果输入的endKey比key值还要小，那么就认为是输入的问题，则返回nil值。

最后19-30行代码表示，key与endKey都输入正常的情况下，则将limit内，大于等于key且小于endKey的键值对都取出来，并返回keys、vals结果。

到此，从buf缓存中就可以读取所需要的数据了，那么，我们回过头接着看UnsafeRange方法的实现，该方法在前面有提到。
我再次把代码贴在这里：

// 该方法用于底层数据的只读操作
func (baseReadTx *baseReadTx) UnsafeRange(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
    // 不使用范围查询
    if endKey == nil {
        // forbid duplicates for single keys
        limit = 1
    }
    // 当范围值异常时，则传入最大范围
    if limit <= 0 {
        limit = math.MaxInt64
    }
    if limit > 1 && !bytes.Equal(bucketName, safeRangeBucket) {
        panic("do not use unsafeRange on non-keys bucket")
    }
    // 将buf缓存中的数据读取出来
    keys, vals := baseReadTx.buf.Range(bucketName, key, endKey, limit)
    // 如果取出的数据满足了需求，那么则直接返回数据
    if int64(len(keys)) == limit {
        return keys, vals
    }

    // find/cache bucket
    // 从bucket缓存中查询bucket实例，查询到了则返回缓存中的实例，查询不到，则在BoltDB中查找
    bn := string(bucketName)
    baseReadTx.txMu.RLock()
    bucket, ok := baseReadTx.buckets[bn]
    baseReadTx.txMu.RUnlock()
    lockHeld := false
    // 缓存中取不到bucket的话，会从bolt中查找，并写入缓存中
    if !ok {
        baseReadTx.txMu.Lock()
        lockHeld = true
        bucket = baseReadTx.tx.Bucket(bucketName)
        baseReadTx.buckets[bn] = bucket
    }

    // ignore missing bucket since may have been created in this batch
    if bucket == nil {
        if lockHeld {
            baseReadTx.txMu.Unlock()
        }
        return keys, vals
    }
    if !lockHeld {
        baseReadTx.txMu.Lock()
        lockHeld = true
    }
    c := bucket.Cursor()
    baseReadTx.txMu.Unlock()

    // 从bolt的该bucket中查找键值对
    k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys)))
    return append(k2, keys...), append(v2, vals...)
}

看第16-20行，刚才我们已经分析了16行代码的具体实现。

18-19行则表示，如果返回的key的数量与limit相等，则就直接返回缓存中的数据即可。如果不是相等的，一般取出的key的数量小于limit值，也就是说，缓存中的数据不完全满足我们的查询需求，那么则需要继续向下执行，到etcd的底层数据库bolt中查询数据。

注意24-43行，首先etcd会从baseReadTx结构体的buckets缓存中查询查询bucket实例，如果缓存中查询不到该实例，则会从bolt数据库中查询并且将实例写入到缓存中。而如果bolt中也查询不到该bucket，则会直接返回之前从buf中查询到的keys与vals值。

如果从缓存或者bolt中查询到了bucket实例，那么，后续就可以直接从bolt中查询该bucket下的键值对了。

我们看一下52行的具体实现。

func unsafeRange(c *bolt.Cursor, key, endKey []byte, limit int64) (keys [][]byte, vs [][]byte) {
    if limit <= 0 {
        limit = math.MaxInt64
    }
    var isMatch func(b []byte) bool
    // 如果有终止的key,则将找到的key与终止的key比较,是否key小于endkey
    // 否则，将找到的key与自身比较是否相等
    if len(endKey) > 0 {
        isMatch = func(b []byte) bool { return bytes.Compare(b, endKey) < 0 }
    } else {
        isMatch = func(b []byte) bool { return bytes.Equal(b, key) }
        limit = 1
    }

    // 循环查找key所对应的值, 然后与endkey做对比(如果有endkey的话)
    // 直到不满足所需条件
    for ck, cv := c.Seek(key); ck != nil && isMatch(ck); ck, cv = c.Next() {
        vs = append(vs, cv)
        keys = append(keys, ck)
        if limit == int64(len(keys)) {
            break
        }
    }
    return keys, vs
}

先看一下传入unsafeRange方法的参数。

c为cursor实例，key为我们要查询的初始key值，endkey为我们要查询的终止key值。而limit是我们查询的范围值，由于我们之前用缓存已经查询出来了一些数据，因此，该范围其实是我们的总范围减去已经查询到的key值数。

其中，5-13行代码用到了匿名函数，用于判断查询到的key值是否依然满足需求。如果我们给到了endkey，那么就会对查到的key与endkey做比较。如果我们没有给endkey，那么就会直接判断查询到的key值是否等于我们要查询的key。

17-23行则为用于DB查询实现代码。

etcd后端存储的写操作

文章开头部分，我们讲到过etcd后端存储对外的接口Backend，其中包括了两个重要的接口：ReadTx以及BatchTx，ReadTx接口负责只读操作，这个我们在前面已经讲到了。
接下来，我们看一下etcd后端存储的读写接口BatchTx。
路径：https://github.com/etcd-io/etcd/blob/master/mvcc/backend/batch_tx.go

type BatchTx interface {
    ReadTx
    UnsafeCreateBucket(name []byte)
    UnsafePut(bucketName []byte, key []byte, value []byte)
    UnsafeSeqPut(bucketName []byte, key []byte, value []byte)
    UnsafeDelete(bucketName []byte, key []byte)
    // Commit commits a previous tx and begins a new writable one.
    Commit()
    // CommitAndStop commits the previous tx and does not create a new one.
    CommitAndStop()
}

我们可以看到，BatchTx接口也包含了前面讲到的ReadTx接口，以及其他用于写操作的方法。batchTx结构体实现了BatchTx接口。

type batchTx struct {
    sync.Mutex
    tx      *bolt.Tx
    backend *backend

    // 数据的偏移量
    pending int
}

具体接口中方法的实现我们就不一一看了，因为都是直接调用了bolt数据库的接口，比较简单。

总结

本篇文章主要从源码角度分析了etcd后端存储的底层读写操作的具体实现。无论我们是使用命令行操作etcd，还是调用etcd的对外接口。最终在对键值对进行读写操作时，底层都会涉及到今天分析的这两个接口：ReadTx以及BatchTx。

然而，etcd的键值对读写其实还会涉及到许多其他的知识，比如revision的概念。接下来还会有文章继续对这些知识做解析。