从零构建分布式KV存储引擎 MiniKV：LSM-Tree、Raft 与并发 B+ 树的工程实践

3.2 SSTable：磁盘上的有序不可变文件

SSTable 是 LSM-Tree 的核心磁盘数据结构。它的设计理念是：一旦写入，永不修改，所有更新和删除都通过新文件来实现。文件结构从上到下依次为：Data Block、Filter Block、Index Block、Footer。Data Block 内部不编码删除操作，删除通过 Tombstone 标记实现。

package lsmtree

import (
 "bytes"
 "encoding/binary"
 "os"
 "sort"
)

const SSTableBlockSize = 4 * 1024

type ByteEntry struct {
 Key       []byte
 Value     []byte
 Tombstone bool
}

type IndexEntry struct {
 MaxKey []byte
 Offset uint32
 Size   uint32
}

type IndexBlock struct {
 Entries []IndexEntry
}

type SSTable struct {
 file     *os.File
 index    *IndexBlock
 bloom    *BloomFilter
 smallest []byte
 largest  []byte
 entries  []ByteEntry
}

func BuildSSTable(entries []ByteEntry, path string) (*SSTable, error) {
 sort.Slice(entries, func(i, j int) bool {
  return bytes.Compare(entries[i].Key, entries[j].Key) < 0 xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed>= SSTableBlockSize {
   data := encodeDataBlock(currentEntries)
   offset, _ := file.Seek(0, io.SeekCurrent)
   indexEntries = append(indexEntries, IndexEntry{
    MaxKey: currentEntries[len(currentEntries)-1].Key,
    Offset: uint32(offset),
    Size:   uint32(len(data)),
   })
   file.Write(data)
   currentEntries = nil
   currentSize = 0
  }
 }
 if len(currentEntries) > 0 {
  data := encodeDataBlock(currentEntries)
  offset, _ := file.Seek(0, io.SeekCurrent)
  indexEntries = append(indexEntries, IndexEntry{
   MaxKey: currentEntries[len(currentEntries)-1].Key,
   Offset: uint32(offset),
   Size:   uint32(len(data)),
  })
  file.Write(data)
 }

 indexData := encodeIndexBlock(indexEntries)
 indexOffset, _ := file.Seek(0, io.SeekCurrent)
 file.Write(indexData)

 bloomData := ss.bloom.Encode()
 bloomOffset, _ := file.Seek(0, io.SeekCurrent)
 file.Write(bloomData)

 footer := make([]byte, 20)
 binary.LittleEndian.PutUint64(footer[0:8], uint64(bloomOffset))
 binary.LittleEndian.PutUint64(footer[8:16], uint64(indexOffset))
 binary.LittleEndian.PutUint32(footer[16:20], uint32(len(entries)))
 file.Write(footer)

 return ss, nil
}

func encodeDataBlock(entries []ByteEntry) []byte {
 var buf bytes.Buffer
 for _, e := range entries {
  var header [9]byte
  binary.LittleEndian.PutUint32(header[0:4], uint32(len(e.Key)))
  binary.LittleEndian.PutUint32(header[4:8], uint32(len(e.Value)))
  if e.Tombstone {
   header[8] = 1
  }
  buf.Write(header[:])
  buf.Write(e.Key)
  buf.Write(e.Value)
 }
 return buf.Bytes()
}

func encodeIndexBlock(entries []IndexEntry) []byte {
 var buf bytes.Buffer
 for _, e := range entries {
  var header [8]byte
  binary.LittleEndian.PutUint32(header[0:4], uint32(len(e.MaxKey)))
  binary.LittleEndian.PutUint32(header[4:8], e.Size)
  buf.Write(header[:])
  buf.Write(e.MaxKey)
 }
 footer := make([]byte, 4)
 binary.LittleEndian.PutUint32(footer, uint32(len(entries)))
 buf.Write(footer)
 return buf.Bytes()
}

四、Compaction：空间回收与读放大权衡

LSM-Tree 最大的挑战来自于读放大和空间放大。Compaction 是将上层数据逐层向下合并的过程。Leveled Compaction（RocksDB 和 LevelDB 的选择）的核心思想是：L0 允许文件 key 范围重叠，L1 及以下每层内文件 key 范围互不重叠。每层大小上限是上层的 10 倍。

package lsmtree

type CompactionManager struct {
 levels     [][]*SSTable
 thresholds []int64
 strategy   string
}

func NewCompactionManager() *CompactionManager {
 return &CompactionManager{
  levels:     make([][]*SSTable, 8),
  strategy:   "leveled",
  thresholds: []int64{4, 10, 100, 1000, 10000, 100000, 1000000, 10000000},
 }
}

func (cm *CompactionManager) PickCompactionLevel() int {
 maxScore := 0.0
 pickLevel := -1
 for i := 0; i < len xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed> maxScore {
   maxScore = score
   pickLevel = i
  }
 }
 return pickLevel
}

func (t *SSTable) Size() int64 {
 info, _ := t.file.Stat()
 if info != nil {
  return info.Size()
 }
 return int64(len(t.entries) * 100)
}

func (cm *CompactionManager) RunCompaction(level int) error {
 if len(cm.levels[level]) == 0 {
  return nil
 }
 if level == 0 {
  tables := cm.levels[0]
  merged := mergeTables(tables)
  sort.Slice(merged, func(i, j int) bool {
   return bytes.Compare(merged[i].Key, merged[j].Key) < 0 xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed xss=removed> len(existing.Value) {
    entryMap[key] = e
   }
  }
 }
 for _, e := range entryMap {
  if !e.Tombstone {
   result = append(result, e)
  }
 }
 return result
}

五、Raft 共识：让多副本数据强一致

单节点 KV 存储存在单点故障风险。Raft 提供了一种比 Paxos 更易懂的共识算法：将一致性问题分解为 Leader 选举、日志复制和安全性三个子问题，通过状态机方式让多个节点对外表现为一个整体。

5.1 Raft 核心数据结构

package raft

import (
 "sync"
 "sync/atomic"
 "time"
)

type Role int

const (
 Follower Role = iota
 Candidate
 Leader
)

type Raft struct {
 mu              sync.Mutex
 id              uint64
 peers           []*raftPeer
 role            Role
 currentTerm     int64
 votedFor        int64
 log             []LogEntry
 commitIndex     int64
 lastApplied     int64
 nextIndex       map[uint64]int64
 matchIndex      map[uint64]int64
 applyCh         chan ApplyMsg
 electionTimer   *time.Timer
}

type LogEntry struct {
 Term    int64
 Index   int64
 Command interface{}
}

type ApplyMsg struct {
 Index   int64
 Command interface{}
}

5.2 Leader 选举

func (rf *Raft) StartElection() {
 rf.mu.Lock()
 rf.role = Candidate
 rf.currentTerm++
 rf.votedFor = rf.id
 currentTerm := rf.currentTerm
 lastLogIndex := int64(len(rf.log) - 1)
 lastLogTerm := rf.log[lastLogIndex].Term
 rf.resetElectionTimer()
 rf.mu.Unlock()

 votes := int32(1)

 for _, peer := range rf.peers {
  if peer.id == rf.id {
   continue
  }
  go func(p *raftPeer) {
   args := &RequestVoteArgs{
    Term:         currentTerm,
    CandidateId:  rf.id,
    LastLogIndex: lastLogIndex,
    LastLogTerm:  lastLogTerm,
   }
   reply := &RequestVoteReply{}
   if p.call("Raft.RequestVote", args, reply) {
    rf.mu.Lock()
    defer rf.mu.Unlock()
    if reply.Term > rf.currentTerm {
     rf.currentTerm = reply.Term
     rf.role = Follower
     rf.votedFor = -1
     return
    }
    if reply.VoteGranted && rf.role == Candidate && rf.currentTerm == currentTerm {
     atomic.AddInt32(&votes, 1)
     if int(atomic.LoadInt32(&votes)) > (len(rf.peers)+1)/2 {
      rf.becomeLeader()
     }
    }
   }
  }(peer)
 }
}

func (rf *Raft) becomeLeader() {
 rf.role = Leader
 lastLogIndex := int64(len(rf.log) - 1)
 for _, peer := range rf.peers {
  rf.nextIndex[peer.id] = lastLogIndex + 1
  rf.matchIndex[peer.id] = -1
 }
 rf.resetElectionTimer()
}

5.3 日志复制

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
 rf.mu.Lock()
 defer rf.mu.Unlock()

 reply.Term = rf.currentTerm
 reply.Success = false

 if args.Term < rf> rf.currentTerm {
  rf.currentTerm = args.Term
  rf.role = Follower
  rf.votedFor = -1
 }

 rf.resetElectionTimer()

 if args.PrevLogIndex >= 0 {
  if args.PrevLogIndex >= int64(len(rf.log)) {
   return
  }
  if rf.log[args.PrevLogIndex].Term != args.PrevLogTerm {
   return
  }
 }

 for i, entry := range args.Entries {
  idx := args.PrevLogIndex + 1 + int64(i)
  if idx < int64 xss=removed xss=removed xss=removed xss=removed> rf.commitIndex {
  lastNew := args.PrevLogIndex + int64(len(args.Entries))
  if args.LeaderCommit < lastNew xss=removed xss=removed xss=removed>

package bptree

import (
 "bytes"
 "sort"
 "sync"
 "sync/atomic"
)

const BTreeOrder = 128

type Node struct {
 IsLeaf   bool
 Keys     [][]byte
 Values   [][]byte
 Children []*Node
 Next     *Node
 version  uint64
 mu       sync.RWMutex
}

type BPlusTree struct {
 root *Node
 mu   sync.RWMutex
}

func (t *BPlusTree) Get(key []byte) ([]byte, bool) {
 leaf := t.findLeaf(key)
 idx := sort.Search(len(leaf.Keys), func(i int) bool {
  return bytes.Compare(leaf.Keys[i], key) >= 0
 })
 if idx < len xss=removed xss=removed>= 0
  })
  curr = curr.Children[idx]
 }
 return curr
}

func (t *BPlusTree) Insert(key, value []byte) {
 leaf := t.findLeaf(key)
 leaf.mu.Lock()
 defer leaf.mu.Unlock()

 if len(leaf.Keys) < BTreeOrder xss=removed xss=removed>= 0
 })
 leaf.Keys = append(leaf.Keys[:idx], append([][]byte{key}, leaf.Keys[idx:]...)...)
 leaf.Values = append(leaf.Values[:idx], append([][]byte{value}, leaf.Values[idx:]...)...)
 atomic.AddUint64(&leaf.version, 1)
}

func (n *Node) optimisticRead() uint64 {
 return atomic.LoadUint64(&n.version)
}