分布式键值存储 etcd 原理与实现 · Analyze

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// etcd/raft/raft.go（代码已删减）
func (r *raft) Step(m pb.Message) error {
    switch m.Type {
    case pb.MsgHup:        // 选举计时器超时会创建 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
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// etcd/raft/raft.go（代码已删减）
func (r *raft) becomeLeader() {
    // 省略初始化 raft 结构体步骤
    emptyEnt := pb.Entry{Data: nil}
    if !r.appendEntry(emptyEnt) {
        r.logger.Panic("empty entry was dropped")
    }
    r.reduceUncommittedSize([]pb.Entry{emptyEnt})
    r.logger.Infof("%x became leader at term %d", r.id, r.Term)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// etcd/wal/walpb/raft.pd.go
type Entry struct {
    Term             uint64
    Index            uint64
    Type             EntryType
    Data             []byte
}

const (
    EntryNormal       EntryType = 0
    EntryConfChange   EntryType = 1
    EntryConfChangeV2 EntryType = 2 // 用于 etcd V2 版本
)

1
2
3
4
5
6
7
8
9

// etcd/raft/raft.go（代码已删减）
func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) {
    li := r.raftLog.lastIndex()  // 获取最后一条日志的索引值
    for i := range es {
        es[i].Term = r.Term
        es[i].Index = li + 1 + uint64(i)
    }
    li = r.raftLog.append(es...) // 向 raftLog 中追加 Entry 日志
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// etcd/raft/log_unstable.go（代码已删减）
type unstable struct {
    snapshot *pb.Snapshot // 未写入 Storage 的快照数据
    entries []pb.Entry    // 未写入 Storage 中的 Entry 日志。
    offset uint64         // entries 中第一条 Entry 日志的索引值
}

// etcd/raft/storage.go
type MemoryStorage struct {
    sync.Mutex
    hardState pb.HardState
    snapshot  pb.Snapshot
    // ents[i] has raft log position i+snapshot.Metadata.Index
    ents []pb.Entry       // 需要注意的是，ents[0] 是一条假数据
}

// etcd/raft/log.go
type raftLog struct {
    storage Storage
    unstable unstable
    committed uint64      // 已提交的 Entry 日志的最大索引值
    applied uint64        // 已应用的 Entry 日志的最大索引值
    logger Logger
    maxNextEntsSize uint64
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

func (u *unstable) truncateAndAppend(ents []pb.Entry) {
    after := ents[0].Index // 获取第一条待追加的 Entry 记录的索引值
    switch {
    case after == u.offset+uint64(len(u.entries)):
        // 如果待追加的记录与 entries 中的记录正好连续，则可以直接向 entries 中追加
        u.entries = append(u.entries, ents...)
    case after <= u.offset:
        // 用待追加的 Entry 记录替换当前的 entries 字段，并更新 offset
        u.logger.Infof("replace the unstable entries from index %d", after)
        u.offset = after
        u.entries = ents
    default:
        // after 在 offset～last 之间，将 offset~after 之间的记录保留，抛弃 after 之后的记录
        u.logger.Infof("truncate the unstable entries before index %d", after)
        u.entries = append([]pb.Entry{}, u.slice(u.offset, after)...)
        u.entries = append(u.entries, ents...)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// etcd/raft/node.go（代码已删减）
func (n *node) run() {
    // 省略初始化变量
    for {
        if advancec != nil {
            readyc = nil
        } else if n.rn.HasReady() {        // 检查是否有需要处理的数据
            rd = n.rn.readyWithoutAccept() // 创建 Ready 数据结构给上层处理
            readyc = n.readyc
        }
        if lead != r.lead {                // 检测当前的 Leader 节点是否发生变化
            if r.hasLeader() {             // 如果当前节点无法确定集群中的 Leader
                propc = n.propc            // 清空 propc，此次循环不处理 MsgProp 消息
            } else {
                propc = nil
            }
            lead = r.lead
        }
        select {
        case pm := <-propc:                // 处理客户端的请求
        case m := <-n.recvc:               // 处理节点间的通信
        case readyc <- rd:
            n.rn.acceptReady(rd)
            advancec = n.advancec
        // 省略监听其它类型的 channel
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// etcd/raft/node.go
type Ready struct {
    *SoftState
    pb.HardState                // 节点状态
    ReadStates []ReadState      // 当前节点等待处理的只读请求
    Entries []pb.Entry          // unstable 待持久化到 WAL 的日志
    Snapshot pb.Snapshot        // 待持久化的快照数据
    CommittedEntries []pb.Entry // 己提交、待应用的 Entry 日志
    Messages []pb.Message       // 当前节点等待发送到集群其他节点的消息
    MustSync bool               // 标记 HardState 和 Entries 同步还是异步写入磁盘
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// etcd/wal/walpb/record.pd.go（代码已删减）
type Record struct {
    Type int64  // 该 Record 实例类型
    Crc uint32  // 该 Record 实例的校验码
    Data []byte // 日志数据
}

// etcd/wal/wal.go
const (          // Record 实例类型
    metadataType int64 = iota + 1 // 该类型 Record 的 Data 字段中保存的是元数据
    entryType    // 该类型 Record 的 Data 字段中保存的是 Entry 日志
    stateType    // 该类型 Record 的 Data 字段中保存了当前集群的状态信息（raftpb.HardState）
    crcType      // 该类型 Record 主要用于数据校验
    snapshotType // 该类型 Record 保存了快照数据的相关信息（walpb.Snapshot）
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

// etcd/wal/wal.go（代码已删减）
func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) {
    if Exist(dirpath) { return nil, os.ErrExist }

    // 创建临时文件
    tmpdirpath := filepath.Clean(dirpath) + ".tmp"
    if fileutil.Exist(tmpdirpath) {
        if err := os.RemoveAll(tmpdirpath); err != nil {
            return nil, err
        }
    }
    if err := fileutil.CreateDirAll(tmpdirpath); err != nil {……}

    p := filepath.Join(tmpdirpath, walName(0, 0))
    // 对文件上互斥锁
    f, err := fileutil.LockFile(p, os.O_WRONLY|os.O_CREATE, fileutil.PrivateFileMode)
    // 定位到文件末尾
    if _, err = f.Seek(0, io.SeekEnd); err != nil {……}
    // 预分配文件空间，大小为 64MB
    if err = fileutil.Preallocate(f.File, SegmentSizeBytes, true); err != nil {……}

    w := &WAL{
        lg:       lg,
        dir:      dirpath,
        metadata: metadata,
    }
    // 对 WAL 文件创建一个 encoder
    w.encoder, err = newFileEncoder(f.File, 0)
    // 将 FD 添加到 locks 数组中
    w.locks = append(w.locks, f)
    if err = w.saveCrc(0); err != nil {……}
    // 将 metadataType 类型的 Record 记录在 WAL 文件的头部
    if err = w.encoder.encode(&walpb.Record{Type: metadataType, Data: metadata}); err != nil {……}
    // 随后写入一条空的 snapshotType Record
    if err = w.SaveSnapshot(walpb.Snapshot{}); err != nil {……}

    if w, err = w.renameWAL(tmpdirpath); err != nil {……} // 初始化完成之后重命名文件

    var perr error
    if perr = fileutil.Fsync(pdir); perr != nil {……} // 同步上述文件操作

    if perr = pdir.Close(); perr != nil {……} // 关闭目录
    return w, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// etcd/wal/file_pipeline.go
func (fp *filePipeline) run() {
    defer close(fp.errc)
    for {
        f, err := fp.alloc() // 创建临时文件
        if err != nil { // 如果创建临时文件失败，则将异常传递到 errc channel 中
            fp.errc <- err
            return
        }
        select { // 将上面创建的临时文件句柄传递到 filec 中
        case fp.filec <- f:
            case <-fp.donec: // 调用 filePipeline.Close() 方法时会关闭该通道，删除最后一次创建的临时文件
            os.Remove(f.Name())
            f.Close()
            return
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// etcd/wal/wal.go（代码已删减）
func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error {
    for i := range ents { // 批量写入 Entry 日志
        if err := w.saveEntry(&ents[i]); err != nil {
            return err
        }
    }
    
    if err := w.saveState(&st); err != nil { // 写入 Entry 后再写入一条 stateType 类型 Record
        return err
    }
    curOff, err := w.tail().Seek(0, io.SeekCurrent)
    if err != nil {
        return err
    }
    if curOff < SegmentSizeBytes {
        if mustSync {
            return w.sync()
        }
        return nil
    }

    return w.cut()
}

func (w *WAL) saveEntry(e *raftpb.Entry) error {……}
func (w *WAL) saveState(s *raftpb.HardState) error {……}
func (w *WAL) SaveSnapshot(e walpb.Snapshot) error {……}
func (w *WAL) saveCrc(prevCrc uint32) error {……}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

// etcd/wal/encoder.go（代码已删减）
func (e *encoder) encode(rec *walpb.Record) error {
    e.mu.Lock()
    defer e.mu.Unlock()

    e.crc.Write(rec.Data)
    rec.Crc = e.crc.Sum32()
    var (
        data []byte
        err  error
        n    int
    )

    if rec.Size() > len(e.buf) { // 如果超过预分配的 buf，就使用动态分配
        data, err = rec.Marshal()
    } else {                     // 使用预分配的 buf
        n, err = rec.MarshalTo(e.buf)
        data = e.buf[:n]
    }

    lenField, padBytes := encodeFrameSize(len(data)) // 计算数据填充长度
    // 写入 Record 编码后的长度
    if err = writeUint64(e.bw, lenField, e.uint64buf); err != nil {
        return err
    }

    if padBytes != 0 {           // 写入填充数据
        data = append(data, make([]byte, padBytes)...)
    }
    n, err = e.bw.Write(data)    // 写入 Record 数据
    walWriteBytes.Add(float64(n))
    return err
}

1
2
3
4
5
6
7
8

// etcd/etcdserver/server.go（代码已删减）
const DefaultSnapshotCount = 100000

func (s *EtcdServer) triggerSnapshot(ep *etcdProgress) {
    if ep.appliedi-ep.snapi <= s.Cfg.SnapshotCount { return }
    s.snapshot(ep.appliedi, ep.confState) // 保存快照
    ep.snapi = ep.appliedi
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// etcd/etcdserver/storage.go
func (st *storage) SaveSnap(snap raftpb.Snapshot) error {
    walsnap := walpb.Snapshot{
        Index: snap.Metadata.Index,
        Term:  snap.Metadata.Term,
    }
    err := st.Snapshotter.SaveSnap(snap)
    if err != nil {
        return err
    }
    // 创建快照后会在 WAL 中写入一条 Snapshot Record
    return st.WAL.SaveSnapshot(walsnap)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// etcd/raft/raft.go
func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
    pr := r.prs.Progress[to]
    m := pb.Message{}
    m.To = to

    term, errt := r.raftLog.term(pr.Next - 1)
    ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
    if len(ents) == 0 && !sendIfEmpty {
        return false
    }
    if errt != nil || erre != nil { // send snapshot if we failed to get term or entries
        m.Type = pb.MsgSnap
        snapshot, err := r.raftLog.snapshot()
        m.Snapshot = snapshot
        sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term
        pr.BecomeSnapshot(sindex)
    } else {……}
    r.send(m) // 发送 MsgSnap
    return true
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

// etcd/etcdserver/snap/db.go（代码已删减）
func (s *Snapshotter) SaveDBFrom(r io.Reader, id uint64) (int64, error) {
    start := time.Now()

    f, err := ioutil.TempFile(s.dir, "tmp")
    var n int64
    n, err = io.Copy(f, r)        // 将快照数据写入临时文件
    if err == nil {
        fsyncStart := time.Now()
        err = fileutil.Fsync(f)   // 将临时文件的改动刷新到磁盘
        snapDBFsyncSec.Observe(time.Since(fsyncStart).Seconds())
    }
    
    f.Close()                     // 关闭临时文件
    fn := s.dbFilePath(id)        // 检查是否存在指定的".snap.db"文件，如果存在则将其删除
    if fileutil.Exist(fn) {
        os.Remove(f.Name())
        return n, nil
    }
    err = os.Rename(f.Name(), fn) // 重命名临时文件

    snapDBSaveSec.Observe(time.Since(start).Seconds())
    return n, nil
}

1
2
3
4

rev={1 0}, key=key1, value="valuel" 
rev={1 1}, key=key2, value="value2" 
rev={2 0}, key=key1, value="updatel"
rev={2 1}, key=key2, value="update2"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// etcd/mvcc/backend/backend.go
type keyIndex struct {
    key []byte        // 客户端提供的原始 Key 值
    modified revision // 该 Key 值最后一次修改时对应的 revision 信息
    generations []generation
}

type revision struct {
    main int64
    sub int64
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

compact(2)
generations:
    {6.0, 7.0}
    {4.0, 5.0(t)}
    {2.0, 3.0(t)}

compact(4)
generations:
    {6.0, 7.0}
    {4.0, 5.0(t)}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// etcd/mvcc/key_index.go（代码已删减）
func (ki *keyIndex) get(lg *zap.Logger, atRev int64) (modified, created revision, ver int64, err error) {
    g := ki.findGeneration(atRev)
    if g.isEmpty() {
        return revision{}, revision{}, 0, ErrRevisionNotFound
    }

    n := g.walk(func(rev revision) bool { return rev.main > atRev })
    if n != -1 {
        return g.revs[n], g.created, g.ver - int64(len(g.revs)-n-1), nil
    }

    return revision{}, revision{}, 0, ErrRevisionNotFound
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// etcd/mvcc/backend/backend.go
type Backend interface {
    ReadTx() ReadTx           // 只读事务，已经被 ConcurrentReadTx() 取代，
    BatchTx() BatchTx         // 读写事务（批量事务）
    ConcurrentReadTx() ReadTx // 非阻塞的只读事务，代替了 ReadTx()，详情见 PR#10523
    Snapshot() Snapshot       // 创建快照
    Hash(ignores map[IgnoreKey]struct{}) (uint32, error)
    Size() int64
    SizeInUse() int64
    OpenReadTxN() int64       // 返回当前正在进行的只读事务数量
    Defrag() error            // 碎片整理
    ForceCommit()             // 提交批量读写事务
    Close() error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// etcd/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
}

// etcd/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()        // 提交当前的读写事务，并立即打开一个新的读写事务
    CommitAndStop() // 提交当前的读写事务，但是不会打开新的读写事务
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// etcd/mvcc/backend/backend.go
var defaultBatchInterval = 100 * time.Millisecond

func (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)
    }
}

1
2
3
4
5
6
7
8
9

// etcd/mvcc/backend/read_tx.go
type readTx struct {
    mu  sync.RWMutex
    buf txReadBuffer // 缓存键值对集合
    txMu    sync.RWMutex
    tx      *bolt.Tx
    buckets map[string]*bolt.Bucket
    txWg *sync.WaitGroup
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// etcd/mvcc/backend/batch_tx.go
type batchTx struct {
    sync.Mutex
    tx      *bolt.Tx
    backend *backend

    pending int // 当前事务中执行的修改操作个数，读写事务提交时该值会被重置为 0
}

type batchTxBuffered struct {
    batchTx
    buf txWriteBuffer
}

1
2
3
4
5
6
7
8
9

// etcd/mvcc/backend/batch_tx.go
var defaultBatchLimit = 10000

func (t *batchTx) Unlock() {
    if t.pending >= t.backend.batchLimit {
        t.commit(false)
    }
    t.Mutex.Unlock()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// etcd/etcdserver/v3_server.go（代码已删减）
func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
    if !r.Serializable { // 执行 linearizableReadNotify()，并等待通知
        err = s.linearizableReadNotify(ctx)
        trace.Step("agreement among raft nodes before linearized reading")
    }
    chk := func(ai *auth.AuthInfo) error {
        return s.authStore.IsRangePermitted(ai, r.Key, r.RangeEnd)
    }

    // 此时 AppliedIndex >= ReadIndex，可以读取数据
    get := func() { resp, err = s.applyV3Base.Range(ctx, nil, r) }
    if serr := s.doSerialize(ctx, chk, get); serr != nil {
        err = serr
        return nil, err
    }
    return resp, err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// etcd/etcdserver/v3_server.go（代码已删减）
func (s *EtcdServer) linearizableReadNotify(ctx context.Context) error {
    s.readMu.RLock()
    nc := s.readNotifier
    s.readMu.RUnlock()

    select {
    case s.readwaitc <- struct{}{}: // 等待读请求处理协程可以处理此次读请求
    default:
    }

    select {
    case <-nc.c: // 接收可以安全执行读请求的信号（此时 AppliedIndex >= ReadIndex）
        return nc.err
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// etcd/etcdserver/v3_server.go（代码已删减）
func (s *EtcdServer) linearizableReadLoop() {
    for {
        select {
        case <-s.readwaitc: // 新的读请求到来时会从 readwaitc 接收到信号
        }
        // 调用 raft 模块执行 readIndex()
        cctx, cancel := context.WithTimeout(context.Background(), s.Cfg.ReqTimeout())
        if err := s.r.ReadIndex(cctx, ctxToSend); err != nil {
            // ……
        }
        cancel()
        // ……
    }
}

1
2
3
4

// etcd/raft/node.go（代码已删减）
func (n *node) ReadIndex(ctx context.Context, rctx []byte) error {
    return n.step(ctx, pb.Message{Type: pb.MsgReadIndex, Entries: []pb.Entry{{Data: rctx}}})
}

1
2
3
4
5
6
7
8
9

// etcd/Raft/raft.go（代码已删减）
func stepFollower(r *raft, m pb.Message) error {
    switch m.Type {
    case pb.MsgReadIndex:
        m.To = r.lead // 将消息的目标地址修改为 Leader，把读请求转发给 Leader
        r.send(m)
    }
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

// etcd/raft/raft.go（代码已删减）
func stepLeader(r *raft, m pb.Message) error {
    switch m.Type {
    case pb.MsgReadIndex:  // 处理 ReadIndex
        switch r.readOnly.option {
        case ReadOnlySafe: // ReadOnlySafe 执行 ReadIndex 算法的逻辑
            r.readOnly.addRequest(r.raftLog.committed, m) // 将请求添加到队列中
            // The local node automatically acks the request.
            r.readOnly.recvAck(r.id, m.Entries[0].Data)
            r.bcastHeartbeatWithCtx(m.Entries[0].Data) // Leader 节点向其他节点发起广播
        case ReadOnlyLeaseBased: // 另一种实现线性一致性的机制
            // ……
        }
    case pb.MsgHeartbeatResp: // 心跳消息响应
        // 清理 readOnly 的 map 和 queue 并将此时读状态添加到 readStates 队列中
        // run 方法会将 readStates 包装成 Ready 数据结构交给应用层处理
        rss := r.readOnly.advance(m)
        for _, rs := range rss {
            req := rs.req
            if req.From == None || req.From == r.id { // from local member
                r.readStates = append(r.readStates, ReadState{Index: rs.index, RequestCtx: req.Entries[0].Data})
            } else {
                r.send(pb.Message{To: req.From, Type: pb.MsgReadIndexResp, Index: rs.index, Entries: req.Entries})
            }
        }
    }
    return nil
    }
}

// etcd/raft/node.go（代码已删减）
func (n *node) run(r *raft) {
    for {
        if advancec != nil {
            readyc = nil
        } else if n.rn.HasReady() {
            rd = n.rn.readyWithoutAccept() // 创建新的 Ready 结构体
            readyc = n.readyc
        }
        select {
            case readyc <- rd:
            // 将创建的 Ready 写入 node.readyc 通道中，等待应用层读取
            n.rn.acceptReady(rd)
            advancec = n.advancec
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// etcd/etcdserver/raft.go（代码已删减）
func (r *raftNode) start(rh *raftReadyHandler) {
    go func() {
        for {
        case rd := <-r.Ready():
            // 处理已完成 readIndex 的读请求
            if len(rd.ReadStates) != 0 {
                select {
                // 每次只将最后一个 readState 发送给 r.readStateC
                case r.readStateC <- rd.ReadStates[len(rd.ReadStates)-1]:
                }
            } 
        }
    }()
}

// etcd/etcdserver/v3_server.go（代码已删减）
func (s *EtcdServer) linearizableReadLoop() {
    for {
        var (
            timeout bool
            done    bool
        )
        for !timeout && !done {
            select {
            case rs = <-s.r.readStateC: // readIndex() 请求完成
                done = bytes.Equal(rs.RequestCtx, ctxToSend)
                // ……
        }
        // readIndex() 请求执行完成，等待 AppliedIndex >= ReadIndex
        if ai := s.getAppliedIndex(); ai < rs.Index {
            select {
            case <-s.applyWait.Wait(rs.Index):
            case <-s.stopping:
                return
            }
        }
        // unblock all l-reads requested at indices before rs.Index
        // 发送可以读取状态机的信号
        nr.notify(nil)
    }
}