高可用分布式存储etcd的实现原理（一）

draveness2018年11月09日约 20 分钟

选择“ 置顶公众号 ”，精品文章第一时间送达！

在上一篇文章详解分布式协调服务 ZooKeeper 中，我们介绍过分布式协调服务 Zookeeper 的实现原理以及应用，今天想要介绍的etcd其实也是在生产环境中经常被使用的协调服务，它与Zookeeper 一样，也能够为整个集群提供服务发现、配置以及分布式协调的功能。

这篇文章将会介绍 etcd 的实现原理，其中包括 Raft 协议、存储两大模块，在最后我们也会简单介绍 etcd 一些具体应用场景。

简介

etcd 的官方将它定位成一个可信赖的分布式键值存储服务，它能够为整个分布式集群存储一些关键数据，协助分布式集群的正常运转。

我们可以简单看一下 etcd 和 Zookeeper 在定义上有什么不同：

etcd is a distributed reliable key-value store for the most critical data of a distributed system…
ZooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and providing group services.

其中前者是一个用于存储关键数据的键值存储，后者是一个用于管理配置等信息的中心化服务。

etcd 的使用其实非常简单，它对外提供了 gRPC 接口，我们可以通过 Protobuf 和 gRPC 直接对 etcd 中存储的数据进行管理，也可以使用官方提供的 etcdctl 操作存储的数据。

service KV {  rpc Range(RangeRequest) returns (RangeResponse) {      option (google.api.http) = {        post: "/v3beta/kv/range"        body: "*"      };  }  rpc Put(PutRequest) returns (PutResponse) {    option (google.api.http) = {      post: "/v3beta/kv/put"      body: "*"    };  }}

文章并不会展开介绍 etcd 的使用方法，这一小节将逐步介绍几大核心模块的实现原理，包括 etcd 使用 Raft 协议同步各个节点数据的过程以及 etcd 底层存储数据使用的结构。

Raft

在每一个分布式系统中，etcd 往往都扮演了非常重要的地位，由于很多服务配置发现以及配置的信息都存储在 etcd 中，所以整个集群可用性的上限往往就是 etcd 的可用性，而使用 3 ~ 5 个 etcd 节点构成高可用的集群往往都是常规操作。

正是因为 etcd 在使用的过程中会启动多个节点，如何处理几个节点之间的分布式一致性就是一个比较有挑战的问题了。

解决多个节点数据一致性的方案其实就是共识算法，在之前的文章中我们简单介绍过 Zookeeper 使用的 Zab 协议以及常见的共识算法 Paxos 和 Raft，etcd 使用的共识算法就是 Raft，这一节我们将详细介绍 Raft 以及 etcd 中 Raft 的一些实现细节。

介绍

Raft 从一开始就被设计成一个易于理解和实现的共识算法，它在容错和性能上与 Paxos 协议比较类似，区别在于它将分布式一致性的问题分解成了几个子问题，然后一一进行解决。

每一个 Raft 集群中都包含多个服务器，在任意时刻，每一台服务器只可能处于 Leader、Follower 以及 Candidate 三种状态；在处于正常的状态时，集群中只会存在一个 Leader，其余的服务器都是 Follower。

上述图片修改自 In Search of an Understandable Consensus Algorithm 一文 5.1 小结中图四。

所有的 Follower 节点都是被动的，它们不会主动发出任何的请求，只会响应 Leader 和 Candidate 发出的请求，对于每一个用户的可变操作，都会被路由给 Leader 节点进行处理，除了 Leader 和 Follower 节点之外，Candidate 节点其实只是集群运行过程中的一个临时状态。

Raft 集群中的时间也被切分成了不同的几个任期（Term），每一个任期都会由 Leader 的选举开始，选举结束后就会进入正常操作的阶段，直到 Leader 节点出现问题才会开始新一轮的选择。

每一个服务器都会存储当前集群的最新任期，它就像是一个单调递增的逻辑时钟，能够同步各个节点之间的状态，当前节点持有的任期会随着每一个请求被传递到其他的节点上。

Raft 协议在每一个任期的开始时都会从一个集群中选出一个节点作为集群的 Leader 节点，这个节点会负责集群中的日志的复制以及管理工作。

我们将 Raft 协议分成三个子问题：节点选举、日志复制以及安全性，文章会以 etcd 为例介绍 Raft 协议是如何解决这三个子问题的。

节点选举

使用 Raft 协议的 etcd 集群在启动节点时，会遵循 Raft 协议的规则，所有节点一开始都被初始化为 Follower 状态，新加入的节点会在 NewNode 中做一些配置的初始化，包括用于接收各种信息的 Channel：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/node.go#L190-225func StartNode(c *Config, peers []Peer) Node {    r := newRaft(c)    r.becomeFollower(1, None)    r.raftLog.committed = r.raftLog.lastIndex()    for _, peer := range peers {       r.addNode(peer.ID)    }    n := newNode()    go n.run(r)    return &n}

在做完这些初始化的节点和 Raft 配置的事情之后，就会进入一个由 for 和 select 组成的超大型循环，这个循环会从 Channel 中获取待处理的事件：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/node.go#L291-423func (n *node) run(r *raft) {    lead := None    for {        if lead != r.lead {            lead = r.lead        }        select {            case m := <-n.recvc:                r.Step(m)            case <-n.tickc:                r.tick()            case <-n.stop:                close(n.done)                return        }    }}

作者对整个循环内的代码进行了简化，因为当前只需要关心三个 Channel 中的消息，也就是用于接受其他节点消息的 recvc 、用于触发定时任务的 tickc 以及用于暂停当前节点的 stop 。

除了 stop Channel 中介绍到的消息之外， recvc 和 tickc 两个 Channel 中介绍到事件时都会交给当前节点持有 Raft 结构体处理。

定时器与心跳

当节点从任意状态（包括启动）调用 becomeFollower 时，都会将节点的定时器设置为 tickElection ：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643func (r *raft) tickElection() {    r.electionElapsed++    if r.promotable() && r.pastElectionTimeout() {       r.electionElapsed = 0       r.Step(pb.Message{From: r.id, Type: pb.MsgHup})    }}

如果当前节点可以成为 Leader 并且上一次收到 Leader 节点的消息或者心跳已经超过了等待的时间，当前节点就会发送 MsgHup 消息尝试开始新的选举。

但是如果 Leader 节点正常运行，就能够同样通过它的定时器 tickHeartbeat 向所有的 Follower 节点广播心跳请求，也就是 MsgBeat 类型的 RPC 消息：

func (r *raft) tickHeartbeat() {    r.heartbeatElapsed++    r.electionElapsed++    if r.heartbeatElapsed >= r.heartbeatTimeout {       r.heartbeatElapsed = 0       r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})    }}

上述代码段 Leader 节点中调用的 Step 函数，最终会调用 stepLeader 方法，该方法会根据消息的类型进行不同的处理：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L931-1142func stepLeader(r *raft, m pb.Message) error {    switch m.Type {        case pb.MsgBeat:           r.bcastHeartbeat()           return nil        // ...    }    //...}

bcastHeartbeat 方法最终会向所有的 Follower 节点发送 MsgHeartbeat 类型的消息，通知它们目前 Leader 的存活状态，重置所有 Follower 持有的超时计时器。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L518-534func (r *raft) sendHeartbeat(to uint64, ctx []byte) {    commit := min(r.getProgress(to).Match, r.raftLog.committed)m := pb.Message{        To:      to,        Type:    pb.MsgHeartbeat,        Commit:  commit,        Context: ctx,    }    r.send(m)}

作为集群中的 Follower，它们会在 stepFollower 方法中处理接收到的全部消息，包括 Leader 节点发送的心跳 RPC 消息：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1191-1247func stepFollower(r *raft, m pb.Message) error {    switch m.Type {       case pb.MsgHeartbeat:           r.electionElapsed = 0           r.lead = m.From           r.handleHeartbeat(m)       // ...    }    return nil}

当 Follower 接受到了来自 Leader 的 RPC 消息 MsgHeartbeat 时，会将当前节点的选举超时时间重置并通过 handleHeartbeat 向 Leader 节点发出响应 —— 通知 Leader 当前节点能够正常运行。

而 Candidate 节点对于 MsgHeartBeat 消息的处理会稍有不同，它会先执行 becomeFollower 设置当前节点和 Raft 协议的配置：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1146-1189func stepCandidate(r *raft, m pb.Message) error {    // ...    switch m.Type {        case pb.MsgHeartbeat:            r.becomeFollower(m.Term, m.From) // always m.Term == r.Term            r.handleHeartbeat(m)    }    // ...    return nil}

Follower 与 Candidate 会根据节点类型的不同做出不同的响应，两者收到心跳请求时都会重置节点的选举超时时间，不过后者会将节点的状态直接转变成 Follower：

当 Leader 节点收到心跳的响应时就会将对应节点的状态设置为 Active ，如果 Follower 节点在一段时间内没有收到来自 Leader 节点的消息就会尝试发起竞选。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643func (r *raft) tickElection() {    r.electionElapsed++    if r.promotable() && r.pastElectionTimeout() {       r.electionElapsed = 0       r.Step(pb.Message{From: r.id, Type: pb.MsgHup})    }}

到了这里，心跳机制就起到了作用开始发送 MsgHup 尝试重置整个集群中的 Leader 节点，接下来我们就会开始分析 Raft 协议中的竞选流程了。

竞选流程

如果集群中的某一个 Follower 节点长时间内没有收到来自 Leader 的心跳请求，当前节点就会通过 MsgHup 消息进入预选举或者选举的流程。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L785-927func (r *raft) Step(m pb.Message) error {    // ...    switch m.Type {        case pb.MsgHup:            if r.state != StateLeader {               if r.preVote {                  r.campaign(campaignPreElection)               } else {                  r.campaign(campaignElection)               }            } else {               r.logger.Debugf("%x ignoring MsgHup because already leader", r.id)            }        }        // ...        return nil}

如果收到 MsgHup 消息的节点不是 Leader 状态，就会根据当前集群的配置选择进入 PreElection 或者 Election 阶段， PreElection 阶段并不会真正增加当前节点的 Term ，它的主要作用是得到当前集群能否成功选举出一个 Leader 的答案，如果当前集群中只有两个节点而且没有预选举阶段，那么这两个节点的 Term 会无休止的增加，预选举阶段就是为了解决这一问题而出现的。

在这里不会讨论预选举的过程，而是将目光主要放在选举阶段，具体了解一下使用 Raft 协议的 etcd 集群是如何从众多节点中选出 Leader 节点的。

我们可以继续来分析 campaign 方法的具体实现，下面就是删去预选举相关逻辑后的代码：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L730-766func (r *raft) campaign(t CampaignType) {      r.becomeCandidate()      if r.quorum() == r.poll(r.id, voteRespMsgType(voteMsg), true) {         r.becomeLeader()         return      }      for id := range r.prs {           if id == r.id {              continue           }           r.send(pb.Message{Term: r.Term, To: id, Type: pb.MsgVote, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})      }}

当前节点会立刻调用 becomeCandidate 将当前节点的 Raft 状态变成候选人；在这之后，它会将票投给自己，如果当前集群只有一个节点，该节点就会直接成为集群中的 Leader 节点。

如果集群中存在了多个节点，就会向集群中的其他节点发出 MsgVote 消息，请求其他节点投票，在 Step 函数中包含不同状态的节点接收到消息时的响应：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L785-927func (r *raft) Step(m pb.Message) error {     // ...     switch m.Type {         case pb.MsgVote, pb.MsgPreVote:              canVote := r.Vote == m.From || (r.Vote == None && r.lead == None)              if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {                 r.send(pb.Message{To: m.From, Term: m.Term, Type: pb.MsgVoteResp})                 r.electionElapsed = 0                 r.Vote = m.From              } else {                 r.send(pb.Message{To: m.From, Term: r.Term, Type: pb.MsgVoteResp, Reject: true})              }     }     // ...     return nil}

如果当前节点投的票就是消息的来源或者当前节点没有投票也没有 Leader，那么就会向来源的节点投票，否则就会通知该节点当前节点拒绝投票。

在 stepCandidate 方法中，候选人节点会处理来自其他节点的投票响应消息，也就是 MsgVoteResp ：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1146-1189func stepCandidate(r *raft, m pb.Message) error {     switch m.Type {          // ...          case pb.MsgVoteResp:               gr := r.poll(m.From, m.Type, !m.Reject)               switch r.quorum() {                   case gr:                       r.becomeLeader()                       r.bcastAppend()               // ...               }     }     return nil}

每当收到一个 MsgVoteResp 类型的消息时，就会设置当前节点持有的 votes 数组，更新其中存储的节点投票状态并返回投『同意』票的人数，如果获得的票数大于法定人数 quorum ，当前节点就会成为集群的 Leader 并向其他的节点发送当前节点当选的消息，通知其余节点更新 Raft 结构体中的 Term 等信息。

节点状态

对于每一个节点来说，它们根据不同的节点状态会对网络层发来的消息做出不同的响应，我们会分别介绍下面的四种状态在 Raft 中对于配置和消息究竟是如何处理的。

对于每一个 Raft 的节点状态来说，它们分别有三个比较重要的区别，其中一个是在改变状态时调用 becomeLeader 、 becomeCandidate 、 becomeFollower 和 becomePreCandidate 方法改变 Raft 状态有比较大的不同，第二是处理消息时调用 stepLeader 、 stepCandidate 和 stepFollower 时有比较大的不同，最后是几种不同状态的节点具有功能不同的定时任务。

对于方法的详细处理，我们在这一节中不详细介绍和分析，如果一个节点的状态是 Follower，那么当前节点切换到 Follower 一定会通过 becomeFollower 函数，在这个函数中会重置节点持有任期，并且设置处理消息的函数为 stepFollower ：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L671-678func (r *raft) becomeFollower(term uint64, lead uint64) {     r.step = stepFollower     r.reset(term)     r.tick = r.tickElection     r.lead = lead     r.state = StateFollower}// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643func (r *raft) tickElection() {     r.electionElapsed++     if r.promotable() && r.pastElectionTimeout() {        r.electionElapsed = 0        r.Step(pb.Message{From: r.id, Type: pb.MsgHup})     }}

除此之外，它还会设置一个用于在 Leader 节点宕机时触发选举的定时器 tickElection 。

Candidate 状态的节点与 Follower 的配置差不了太多，只是在消息处理函数 step 、任期以及状态上的设置有一些比较小的区别：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L680-691func (r *raft) becomeCandidate() {      r.step = stepCandidate      r.reset(r.Term + 1)      r.tick = r.tickElection      r.Vote = r.id      r.state = StateCandidate}

最后的 Leader 就与这两者有其他的区别了，它不仅设置了处理消息的函数 step 而且设置了与其他状态完全不同的 tick 函数：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L708-728func (r *raft) becomeLeader() {     r.step = stepLeader     r.reset(r.Term)     r.tick = r.tickHeartbeat     r.lead = r.id     r.state = StateLeader     r.pendingConfIndex = r.raftLog.lastIndex()     r.appendEntry(pb.Entry{Data: nil})}

这里的 tick 函数 tickHeartbeat 每隔一段时间会通过 Step 方法向集群中的其他节点发送 MsgBeat 消息：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L646-669func (r *raft) tickHeartbeat() {    r.heartbeatElapsed++    r.electionElapsed++    if r.electionElapsed >= r.electionTimeout {       r.electionElapsed = 0       if r.checkQuorum {          r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum})       }    }     if r.heartbeatElapsed >= r.heartbeatTimeout {       r.heartbeatElapsed = 0       r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})    }}

上述代码中的 MsgBeat 消息会在 Step 中被转换成 MsgHeartbeat 最终发送给其他的节点，Leader 节点超时之后的选举流程我们在前两节中也已经介绍过了，在这里就不再重复了。

存储

etcd 目前支持 V2 和 V3 两个大版本，这两个版本在实现上有比较大的不同，一方面是对外提供接口的方式，另一方面就是底层的存储引擎，V2 版本的实例是一个纯内存的实现，所有的数据都没有存储在磁盘上，而 V3 版本的实例就支持了数据的持久化。

在这一节中，我们会介绍 V3 版本的 etcd 究竟是通过什么样的方式存储用户数据的。

后端

在 V3 版本的设计中，etcd 通过 backend 后端这一设计，很好地封装了存储引擎的实现细节，为上层提供一个更一致的接口，对于 etcd 的其他模块来说，它们可以将更多注意力放在接口中的约定上，不过在这里，我们更关注的是 etcd 对 Backend 接口的实现。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L51-69type Backend interface {    ReadTx() ReadTx    BatchTx() BatchTx    Snapshot() Snapshot    Hash(ignores map[IgnoreKey]struct{}) (uint32, error)    Size() int64    SizeInUse() int64    Defrag() error    ForceCommit()    Close() error}

etcd 底层默认使用的是开源的嵌入式键值存储数据库 bolt，但是这个项目目前的状态已经是归档不再维护了，如果想要使用这个项目可以使用 CoreOS 的 bbolt 版本。

这一小节中，我们会简单介绍 etcd 是如何使用 BoltDB 作为底层存储的，首先可以先来看一下 pacakge 内部的 backend 结构体，这是一个实现了 Backend 接口的结构：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L80-104type backend struct {     size int64     sizeInUse int64     commits int64     mu sync.RWMutex     db *bolt.DB     batchInterval time.Duration     batchLimit    int     batchTx       *batchTxBuffered     readTx *readTx     stopc chan struct{}     donec chan struct{}     lg *zap.Logger}

从结构体的成员 db 我们就可以看出，它使用了 BoltDB 作为底层存储，另外的两个 readTx 和 batchTx 分别实现了 ReadTx 和 BatchTx 接口：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/read_tx.go#L30-36type ReadTx interface {     Lock()     Unlock()     UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte)     UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error}// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L28-38type 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()     CommitAndStop()}

从这两个接口的定义，我们不难发现它们能够对外提供数据库的读写操作，而 Backend 就能对这两者提供的方法进行封装，为上层屏蔽存储的具体实现：

每当我们使用 newBackend 创建一个新的 backend 结构时，都会创建一个 readTx 和 batchTx 结构体，这两者一个负责处理只读请求，一个负责处理读写请求：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L137-176func newBackend(bcfg BackendConfig) *backend {    bopts := &bolt.Options{}    bopts.InitialMmapSize = bcfg.mmapSize()db, _ := bolt.Open(bcfg.Path, 0600, bopts)    b := &backend{       db: db,       batchInterval: bcfg.BatchInterval,       batchLimit:    bcfg.BatchLimit,       readTx: &readTx{            buf: txReadBuffer{                 txBuffer: txBuffer{make(map[string]*bucketBuffer)},            },            buckets: make(map[string]*bolt.Bucket),       },       stopc: make(chan struct{}),       donec: make(chan struct{}),    }    b.batchTx = newBatchTxBuffered(b)    go b.run()    return b}

当我们在 newBackend 中进行了初始化 BoltDB、事务等工作后，就会开一个 goroutine 异步的对所有批量读写事务进行定时提交：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L289-305func (b *backend) run() {    defer close(b.donec)    t := time.NewTimer(b.batchInterval)    defer t.Stop()    for {        select {            case <-t.C:            case <-b.stopc:               b.batchTx.CommitAndStop()               return        }        if b.batchTx.safePending() != 0 {           b.batchTx.Commit()        }        t.Reset(b.batchInterval)    }}

对于上层来说， backend 其实只是对底层存储的一个抽象，很多时候并不会直接跟它打交道，往往都是使用它持有的 ReadTx 和 BatchTx 与数据库进行交互。

只读事务

目前大多数的数据库对于只读类型的事务并没有那么多的限制，尤其是在使用了 MVCC 之后，所有的只读请求几乎不会被写请求锁住，这大大提升了读的效率，由于在 BoltDB 的同一个 goroutine 中开启两个相互依赖的只读事务和读写事务会发生死锁，为了避免这种情况我们还是引入了 sync.RWLock 保证死锁不会出现：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/read_tx.go#L38-47type readTx struct {    mu  sync.RWMutex    buf txReadBuffer    txmu    sync.RWMutex    tx      *bolt.Tx    buckets map[string]*bolt.Bucket}

你可以看到在整个结构体中，除了用于保护 tx 的 txmu 读写锁之外，还存在另外一个 mu 读写锁，它的作用是保证 buf 中的数据不会出现问题， buf 和结构体中的 buckets 都是用于加速读效率的缓存。

对于一个只读事务来说，它对上层提供了两个获取存储引擎中数据的接口，分别是 UnsafeRange 和 UnsafeForEach ，在这里会重点介绍前面方法的实现细节：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/read_tx.go#L52-90func (rt *readTx) UnsafeRange(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {    if endKey == nil {        limit = 1    }    keys, vals := rt.buf.Range(bucketName, key, endKey, limit)    if int64(len(keys)) == limit {        return keys, vals    }    bn := string(bucketName)    bucket, ok := rt.buckets[bn]    if !ok {        bucket = rt.tx.Bucket(bucketName)        rt.buckets[bn] = bucket    }    if bucket == nil {        return keys, vals    }    c := bucket.Cursor()    k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys)))    return append(k2, keys...), append(v2, vals...)}

上述代码中省略了加锁保护读缓存以及 Bucket 中存储数据的合法性，也省去了一些参数的检查，不过方法的整体接口还是没有太多变化， UnsafeRange 会先从自己持有的缓存 txReadBuffer 中读取数据，如果数据不能够满足调用者的需求，就会从 buckets 缓存中查找对应的 BoltDB bucket 并从 BoltDB 数据库中读取。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L121-141func unsafeRange(c *bolt.Cursor, key, endKey []byte, limit int64) (keys [][]byte, vs [][]byte) {    var isMatch func(b []byte) bool    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    }    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 实际上通过 BoltDB 中的游标来遍历满足查询条件的键值对。

到这里为止，整个只读事务提供的接口就基本介绍完了，在 etcd 中无论我们想要后去单个 Key 还是一个范围内的 Key 最终都是通过 Range 来实现的，这其实也是只读事务的最主要功能。

读写事务

只读事务只提供了读数据的能力，包括 UnsafeRange 和 UnsafeForeach ，而读写事务 BatchTx 提供的就是读和写数据的能力了：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L40-46type batchTx struct {    sync.Mutex    tx      *bolt.Tx    backend *backend    pending int}

读写事务同时提供了不带缓存的 batchTx 实现以及带缓存的 batchTxBuffered 实现，后者其实『继承了』前者的结构体，并额外加入了缓存 txWriteBuffer 加速读请求：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L243-246type batchTxBuffered struct {    batchTx    buf txWriteBuffer}

后者在实现接口规定的方法时，会直接调用 batchTx 的同名方法，并将操作造成的副作用的写入的缓存中，在这里我们并不会展开介绍这一版本的实现，还是以分析 batchTx 的方法为主。

当我们向 etcd 中写入数据时，最终都会调用 batchTx 的 unsafePut 方法将数据写入到 BoltDB 中：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L65-67func (t *batchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {    t.unsafePut(bucketName, key, value, false)}// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L74-103func (t *batchTx) unsafePut(bucketName []byte, key []byte, value []byte, seq bool) {    bucket := t.tx.Bucket(bucketName)    if err := bucket.Put(key, value); err != nil {       plog.Fatalf("cannot put key into bucket (%v)", err)    }    t.pending++}

这两个方法的实现非常清晰，作者觉得他们都并不值得展开详细介绍，只是调用了 BoltDB 提供的 API 操作一下 bucket 中的数据，而另一个删除方法的实现与这个也差不多：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L144-169func (t *batchTx) UnsafeDelete(bucketName []byte, key []byte) {    bucket := t.tx.Bucket(bucketName)    err := bucket.Delete(key)    if err != nil {       plog.Fatalf("cannot delete key from bucket (%v)", err)    }    t.pending++}

它们都是通过 Bolt.Tx 找到对应的 Bucket ，然后做出相应的增删操作，但是这写请求在这两个方法执行后其实并没有提交，我们还需要手动或者等待 etcd 自动将请求提交：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L184-188func (t *batchTx) Commit() {    t.Lock()    t.commit(false)    t.Unlock()}// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L210-241func (t *batchTx) commit(stop bool) {    if t.tx != nil {       if t.pending == 0 && !stop {          return       }       start := time.Now()       err := t.tx.Commit()       rebalanceSec.Observe(t.tx.Stats().RebalanceTime.Seconds())       spillSec.Observe(t.tx.Stats().SpillTime.Seconds())       writeSec.Observe(t.tx.Stats().WriteTime.Seconds())       commitSec.Observe(time.Since(start).Seconds())       atomic.AddInt64(&t.backend.commits, 1)       t.pending = 0    }    if !stop {       t.tx = t.backend.begin(true)    }}