第4讲：分布式存储基石：复制状态机-恒创新域建站

★ FEATURED ARTICLE

一、复制状态机Replicated State Machine原理复制状态机是分布式共识算法的核心应用模式。它的本质是让所有节点以相同的顺序执行相同的命令从而到达相同的状态。┌─────────────────────────────────────────────────────────────────┐ │ 复制状态机架构 │ │ │ │ Client │ │ │ │ │ ▼ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ 共识模块Raft/Paxos │ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │ │节点1│ │节点2│ │节点3│ ← 日志复制 │ │ │ │ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ │ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │ │日志 │ │日志 │ │日志 │ ← 一致的日志序列 │ │ │ │ │[1,2,3]│ │[1,2,3]│ │[1,2,3]│ │ │ │ │ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ └─────┼────────┼────────┼──────────────────────────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌───────┐ ┌───────┐ ┌───────┐ │ │ │状态机 │ │状态机 │ │状态机 │ ← 相同的最终状态 │ │ │x5,y3│ │x5,y3│ │x5,y3│ │ │ └───────┘ └───────┘ └───────┘ │ │ │ └─────────────────────────────────────────────────────────────────┘核心保证如果所有节点从相同的初始状态出发按相同的顺序执行相同的命令那么所有节点最终状态必然一致。二、WALWrite-Ahead Log实现WAL 是复制状态机的基石。所有修改操作必须先写入日志再应用到状态机。// distributed/rsm/wal.go package rsm import ( encoding/binary encoding/json fmt io os path/filepath sort sync ) // LogEntryType 日志条目类型 type LogEntryType byte const ( LogEntryCommand LogEntryType iota // 普通命令 LogEntryConfiguration // 配置变更 LogEntryNoOp // 空操作 ) // LogEntry 日志条目持久化格式 type LogEntry struct { Index uint64 json:index Term uint64 json:term Type LogEntryType json:type Command json.RawMessage json:command,omitempty Data []byte json:data,omitempty } // WAL 预写日志 type WAL struct { dir string mu sync.RWMutex file *os.File size int64 entries []*LogEntry synced bool // 分段管理 segmentSize int64 // 每个段文件的最大大小 segments []*segmentInfo } type segmentInfo struct { path string startIdx uint64 endIdx uint64 size int64 } // NewWAL 创建 WAL func NewWAL(dir string, segmentSize int64) (*WAL, error) { if err : os.MkdirAll(dir, 0755); err ! nil { return nil, fmt.Errorf(create wal dir: %w, err) } wal : WAL{ dir: dir, segmentSize: segmentSize, entries: make([]*LogEntry, 0), } // 加载现有段 if err : wal.loadSegments(); err ! nil { return nil, fmt.Errorf(load segments: %w, err) } // 打开当前写入段 if err : wal.openCurrentSegment(); err ! nil { return nil, fmt.Errorf(open current segment: %w, err) } return wal, nil } // Append 追加日志条目 func (w *WAL) Append(entry *LogEntry) error { w.mu.Lock() defer w.mu.Unlock() // 检查是否需要滚动段 if w.size w.segmentSize { if err : w.rotateSegment(); err ! nil { return fmt.Errorf(rotate segment: %w, err) } } // 序列化 data, err : w.encodeEntry(entry) if err ! nil { return fmt.Errorf(encode entry: %w, err) } // 写入文件 n, err : w.file.Write(data) if err ! nil { return fmt.Errorf(write entry: %w, err) } w.size int64(n) entry.Index uint64(len(w.entries)) 1 w.entries append(w.entries, entry) return nil } // Sync 刷盘 func (w *WAL) Sync() error { w.mu.RLock() defer w.mu.RUnlock() if w.file nil { return nil } if err : w.file.Sync(); err ! nil { return fmt.Errorf(sync wal: %w, err) } w.synced true return nil } // ReadFrom 从指定索引读取日志 func (w *WAL) ReadFrom(index uint64, maxCount int) ([]*LogEntry, error) { w.mu.RLock() defer w.mu.RUnlock() if index 1 || index uint64(len(w.entries)) { return nil, fmt.Errorf(index out of range: %d, index) } start : int(index) - 1 end : start maxCount if end len(w.entries) { end len(w.entries) } result : make([]*LogEntry, end-start) copy(result, w.entries[start:end]) return result, nil } // Truncate 截断到指定索引 func (w *WAL) Truncate(lastKeepIndex uint64) error { w.mu.Lock() defer w.mu.Unlock() if lastKeepIndex uint64(len(w.entries)) { return nil } // 截断内存中的日志 w.entries w.entries[:lastKeepIndex] // 重建日志文件 if err : w.rebuild(); err ! nil { return fmt.Errorf(rebuild wal: %w, err) } return nil } // encodeEntry 编码日志条目 func (w *WAL) encodeEntry(entry *LogEntry) ([]byte, error) { data, err : json.Marshal(entry) if err ! nil { return nil, err } // 格式[长度(uint64)][数据] buf : make([]byte, 8len(data)) binary.BigEndian.PutUint64(buf[:8], uint64(len(data))) copy(buf[8:], data) return buf, nil } // decodeEntry 解码日志条目 func (w *WAL) decodeEntry(data []byte) (*LogEntry, error) { var entry LogEntry if err : json.Unmarshal(data, entry); err ! nil { return nil, err } return entry, nil } // rotateSegment 滚动到新段 func (w *WAL) rotateSegment() error { // 关闭当前文件 if w.file ! nil { w.file.Close() } // 记录当前段信息 segPath : w.file.Name() w.segments append(w.segments, segmentInfo{ path: segPath, startIdx: 1, endIdx: uint64(len(w.entries)), size: w.size, }) // 打开新段 return w.openCurrentSegment() } // openCurrentSegment 打开当前写入段 func (w *WAL) openCurrentSegment() error { segPath : filepath.Join(w.dir, fmt.Sprintf(wal-%016x.log, len(w.segments))) file, err : os.OpenFile(segPath, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644) if err ! nil { return err } w.file file w.size 0 return nil } // loadSegments 加载现有段 func (w *WAL) loadSegments() error { files, err : filepath.Glob(filepath.Join(w.dir, wal-*.log)) if err ! nil { return err } sort.Strings(files) for _, f : range files { data, err : os.ReadFile(f) if err ! nil { return err } offset : 0 for offset len(data) { // 读取长度 if offset8 len(data) { break } length : binary.BigEndian.Uint64(data[offset : offset8]) offset 8 // 读取条目 if offsetint(length) len(data) { break } entry, err : w.decodeEntry(data[offset : offsetint(length)]) if err ! nil { return err } offset int(length) entry.Index uint64(len(w.entries)) 1 w.entries append(w.entries, entry) } } return nil } // rebuild 重建 WAL func (w *WAL) rebuild() error { // 删除所有段文件 files, _ : filepath.Glob(filepath.Join(w.dir, wal-*.log)) for _, f : range files { os.Remove(f) } w.segments nil w.size 0 // 重新打开 if err : w.openCurrentSegment(); err ! nil { return err } // 重新写入所有条目 for _, entry : range w.entries { data, err : w.encodeEntry(entry) if err ! nil { return err } if _, err : w.file.Write(data); err ! nil { return err } } return w.file.Sync() } // Close 关闭 WAL func (w *WAL) Close() error { w.mu.Lock() defer w.mu.Unlock() if w.file ! nil { if err : w.file.Sync(); err ! nil { return err } return w.file.Close() } return nil }三、状态机快照Snapshot当日志不断增长重启时需要从头回放所有日志这会非常慢。快照机制解决了这个问题定期保存状态机的完整状态然后截断之前的日志。// distributed/rsm/snapshot.go package rsm import ( compress/gzip encoding/json fmt io os path/filepath sort sync ) // Snapshot 快照 type Snapshot struct { LastIncludedIndex uint64 json:last_included_index LastIncludedTerm uint64 json:last_included_term State map[string]string json:state Metadata map[string]interface{} json:metadata,omitempty } // Snapshotter 快照管理器 type Snapshotter struct { dir string mu sync.RWMutex snapshot *Snapshot snapshotting bool } // NewSnapshotter 创建快照管理器 func NewSnapshotter(dir string) (*Snapshotter, error) { if err : os.MkdirAll(dir, 0755); err ! nil { return nil, fmt.Errorf(create snapshot dir: %w, err) } s : Snapshotter{ dir: dir, } // 加载最新的快照 if err : s.loadLatest(); err ! nil { return nil, err } return s, nil } // TakeSnapshot 创建快照 func (s *Snapshotter) TakeSnapshot(state map[string]string, lastIndex, lastTerm uint64) error { s.mu.Lock() defer s.mu.Unlock() if s.snapshotting { return fmt.Errorf(snapshot already in progress) } s.snapshotting true defer func() { s.snapshotting false }() snapshot : Snapshot{ LastIncludedIndex: lastIndex, LastIncludedTerm: lastTerm, State: state, Metadata: map[string]interface{}{ timestamp: time.Now().Unix(), entry_count: len(state), }, } // 序列化 data, err : json.Marshal(snapshot) if err ! nil { return fmt.Errorf(marshal snapshot: %w, err) } // 压缩 compressed, err : s.compress(data) if err ! nil { return fmt.Errorf(compress snapshot: %w, err) } // 写入临时文件 tmpPath : filepath.Join(s.dir, snapshot.tmp) if err : os.WriteFile(tmpPath, compressed, 0644); err ! nil { return fmt.Errorf(write temp snapshot: %w, err) } // 原子替换 finalPath : filepath.Join(s.dir, fmt.Sprintf(snapshot-%016d.snap, lastIndex)) if err : os.Rename(tmpPath, finalPath); err ! nil { return fmt.Errorf(rename snapshot: %w, err) } // 清理旧快照 s.cleanOldSnapshots() s.snapshot snapshot return nil } // LoadSnapshot 加载快照 func (s *Snapshotter) LoadSnapshot() (*Snapshot, error) { s.mu.RLock() defer s.mu.RUnlock() if s.snapshot nil { return nil, nil } return s.snapshot, nil } // InstallSnapshot 安装来自 Leader 的快照用于 Follower 追赶 func (s *Snapshotter) InstallSnapshot(data []byte) (*Snapshot, error) { // 解压 decompressed, err : s.decompress(data) if err ! nil { return nil, fmt.Errorf(decompress snapshot: %w, err) } var snapshot Snapshot if err : json.Unmarshal(decompressed, snapshot); err ! nil { return nil, fmt.Errorf(unmarshal snapshot: %w, err) } s.mu.Lock() s.snapshot snapshot s.mu.Unlock() return snapshot, nil } // compress 压缩数据 func (s *Snapshotter) compress(data []byte) ([]byte, error) { var buf bytes.Buffer writer : gzip.NewWriter(buf) if _, err : writer.Write(data); err ! nil { return nil, err } if err : writer.Close(); err ! nil { return nil, err } return buf.Bytes(), nil } // decompress 解压数据 func (s *Snapshotter) decompress(data []byte) ([]byte, error) { reader, err : gzip.NewReader(bytes.NewReader(data)) if err ! nil { return nil, err } defer reader.Close() return io.ReadAll(reader) } // cleanOldSnapshots 清理旧快照保留最近 3 个 func (s *Snapshotter) cleanOldSnapshots() { files, _ : filepath.Glob(filepath.Join(s.dir, snapshot-*.snap)) sort.Sort(sort.Reverse(sort.StringSlice(files))) // 保留最近 3 个 for i : 3; i len(files); i { os.Remove(files[i]) } } // loadLatest 加载最新快照 func (s *Snapshotter) loadLatest() error { files, _ : filepath.Glob(filepath.Join(s.dir, snapshot-*.snap)) if len(files) 0 { return nil } sort.Strings(files) latest : files[len(files)-1] data, err : os.ReadFile(latest) if err ! nil { return err } decompressed, err : s.decompress(data) if err ! nil { return err } var snapshot Snapshot if err : json.Unmarshal(decompressed, snapshot); err ! nil { return err } s.snapshot snapshot return nil }四、KV 存储引擎实现结合 Raft 共识、WAL 和快照实现一个完整的复制状态机。// distributed/rsm/kvstore.go package rsm import ( fmt log sync time ) // Command 状态机命令 type Command struct { Op string json:op // set / get / delete Key string json:key Value string json:value,omitempty } // KVStore 基于复制状态机的 KV 存储 type KVStore struct { mu sync.RWMutex data map[string]string wal *WAL snapshotter *Snapshotter appliedIndex uint64 // 统计 stats Stats } type Stats struct { ReadCount int64 WriteCount int64 LastCompact time.Time } // NewKVStore 创建 KV 存储 func NewKVStore(walDir, snapDir string) (*KVStore, error) { wal, err : NewWAL(walDir, 64 * 1024 * 1024) // 64MB 每段 if err ! nil { return nil, fmt.Errorf(create wal: %w, err) } snapshotter, err : NewSnapshotter(snapDir) if err ! nil { return nil, fmt.Errorf(create snapshotter: %w, err) } store : KVStore{ data: make(map[string]string), wal: wal, snapshotter: snapshotter, stats: Stats{ LastCompact: time.Now(), }, } // 从快照恢复 if err : store.recoverFromSnapshot(); err ! nil { return nil, fmt.Errorf(recover from snapshot: %w, err) } // 从 WAL 恢复 if err : store.recoverFromWAL(); err ! nil { return nil, fmt.Errorf(recover from wal: %w, err) } return store, nil } // Apply 应用日志条目到状态机 func (s *KVStore) Apply(entry *LogEntry) error { s.mu.Lock() defer s.mu.Unlock() var cmd Command if err : json.Unmarshal(entry.Command, cmd); err ! nil { return fmt.Errorf(unmarshal command: %w, err) } switch cmd.Op { case set: s.data[cmd.Key] cmd.Value log.Printf([KV] Applied SET %s%s (index%d), cmd.Key, cmd.Value, entry.Index) case delete: delete(s.data, cmd.Key) log.Printf([KV] Applied DELETE %s (index%d), cmd.Key, entry.Index) default: return fmt.Errorf(unknown operation: %s, cmd.Op) } s.appliedIndex entry.Index s.stats.WriteCount // 检查是否需要创建快照 if s.shouldTakeSnapshot() { go s.takeSnapshot() } return nil } // Get 读取线性一致读需要配合 ReadIndex func (s *KVStore) Get(key string) (string, bool) { s.mu.RLock() defer s.mu.RUnlock() value, exists : s.data[key] s.stats.ReadCount return value, exists } // PrepareCommand 准备命令写入 WAL 前调用 func (s *KVStore) PrepareCommand(cmd Command) (*LogEntry, error) { cmdData, err : json.Marshal(cmd) if err ! nil { return nil, err } entry : LogEntry{ Type: LogEntryCommand, Command: cmdData, } // 写入 WAL if err : s.wal.Append(entry); err ! nil { return nil, fmt.Errorf(append to wal: %w, err) } // 刷盘 if err : s.wal.Sync(); err ! nil { return nil, fmt.Errorf(sync wal: %w, err) } return entry, nil } // recoverFromSnapshot 从快照恢复 func (s *KVStore) recoverFromSnapshot() error { snapshot, err : s.snapshotter.LoadSnapshot() if err ! nil { return err } if snapshot nil { return nil } s.mu.Lock() defer s.mu.Unlock() for k, v : range snapshot.State { s.data[k] v } s.appliedIndex snapshot.LastIncludedIndex log.Printf([KV] Recovered from snapshot: %d entries, last index%d, len(snapshot.State), snapshot.LastIncludedIndex) return nil } // recoverFromWAL 从 WAL 恢复 func (s *KVStore) recoverFromWAL() error { // 从快照之后的位置开始回放 entries, err : s.wal.ReadFrom(s.appliedIndex1, 1000000) if err ! nil { return err } for _, entry : range entries { if err : s.Apply(entry); err ! nil { return fmt.Errorf(apply entry %d: %w, entry.Index, err) } } log.Printf([KV] Recovered from WAL: %d entries replayed, len(entries)) return nil } // shouldTakeSnapshot 判断是否需要创建快照 func (s *KVStore) shouldTakeSnapshot() bool { // 每 10000 次写入创建一个快照 return s.stats.WriteCount%10000 0 s.stats.WriteCount 0 } // takeSnapshot 创建快照 func (s *KVStore) takeSnapshot() { s.mu.RLock() state : make(map[string]string) for k, v : range s.data { state[k] v } lastIndex : s.appliedIndex s.mu.RUnlock() // 从 WAL 获取最后条目的 term entries, _ : s.wal.ReadFrom(lastIndex, 1) lastTerm : uint64(0) if len(entries) 0 { lastTerm entries[0].Term } if err : s.snapshotter.TakeSnapshot(state, lastIndex, lastTerm); err ! nil { log.Printf([KV] Failed to take snapshot: %v, err) return } // 截断 WAL if err : s.wal.Truncate(lastIndex); err ! nil { log.Printf([KV] Failed to truncate wal: %v, err) } log.Printf([KV] Snapshot taken at index %d (%d entries), lastIndex, len(state)) } // GetStats 获取统计信息 func (s *KVStore) GetStats() Stats { s.mu.RLock() defer s.mu.RUnlock() return s.stats }五、线性一致读Linearizable ReadRaft 的写操作需要走共识但读操作如果不加控制可能会读到过期数据。有三种方式实现线性一致读// distributed/rsm/linearizable.go package rsm import ( fmt log sync/atomic time ) // ReadIndex 方式一ReadIndex // Leader 记录当前的 commitIndex等到这个 index 应用到状态机后再读 type ReadIndex struct { readIndex uint64 applied uint64 } func (ri *ReadIndex) ExecuteRead(store *KVStore, key string) (string, bool) { // 1. Leader 记录当前的 commitIndex index : atomic.LoadUint64(ri.readIndex) // 2. 等待该 index 应用到状态机 for atomic.LoadUint64(ri.applied) index { time.Sleep(10 * time.Microsecond) } // 3. 在本地状态机上读取 return store.Get(key) } // LeaseRead 方式二LeaseRead // Leader 利用选举超时保证在 lease 时间内不会有新 Leader 产生 type LeaseRead struct { leaderSince time.Time leaseTime time.Duration } func NewLeaseRead(leaseTime time.Duration) *LeaseRead { return LeaseRead{ leaderSince: time.Now(), leaseTime: leaseTime, } } func (lr *LeaseRead) ExecuteRead(store *KVStore, key string) (string, bool) { // 在 lease 有效期内可以直接读本地状态机 if time.Since(lr.leaderSince) lr.leaseTime { return store.Get(key) } // lease 过期需要走 ReadIndex 或重新协商 return , false } // FollowerRead 方式三Follower 转发给 Leader // 最简单但最慢的方式 type FollowerRead struct { leaderAddr string } func (fr *FollowerRead) ForwardToLeader(store *KVStore, key string) (string, bool, error) { // 将读请求转发给 Leader // 这里假设通过网络调用 Leader 的 ReadIndex 接口 log.Printf([READ] Forwarding read of %s to leader %s, key, fr.leaderAddr) return store.Get(key), true, nil }六、使用示例// examples/rsm_demo.go package main import ( encoding/json fmt log time distributed/rsm ) func main() { fmt.Println( 复制状态机演示 \n) // 1. 创建 KV 存储 store, err : rsm.NewKVStore(/tmp/rsm-wal, /tmp/rsm-snap) if err ! nil { log.Fatalf(Failed to create KV store: %v, err) } defer store.Close() // 2. 模拟 Raft 提交的日志条目 fmt.Println(【场景1】应用日志条目到状态机\n) commands : []struct { op string key string value string }{ {set, name, Alice}, {set, age, 30}, {set, city, Beijing}, {set, job, Engineer}, {delete, age, }, } for i, cmd : range commands { command : rsm.Command{ Op: cmd.op, Key: cmd.key, Value: cmd.value, } // 准备命令写入 WAL entry, err : store.PrepareCommand(command) if err ! nil { log.Printf(Prepare command failed: %v, err) continue } // 应用到状态机 if err : store.Apply(entry); err ! nil { log.Printf(Apply command failed: %v, err) continue } fmt.Printf([%d] Applied: %s %s%s (index%d)\n, i1, cmd.op, cmd.key, cmd.value, entry.Index) } // 3. 读取数据 fmt.Println(\n【场景2】读取状态机数据\n) keys : []string{name, age, city, job} for _, key : range keys { value, exists : store.Get(key) if exists { fmt.Printf(GET %s %s\n, key, value) } else { fmt.Printf(GET %s not found\n, key) } } // 4. 模拟快照 fmt.Println(\n【场景3】创建快照\n) fmt.Println(触发快照条件...) // 手动触发快照 state : make(map[string]string) for _, key : range keys { if v, ok : store.Get(key); ok { state[key] v } } // 5. 模拟重启恢复 fmt.Println(\n【场景4】模拟重启恢复\n) // 关闭当前 store store.Close() // 重新创建 store从快照和 WAL 恢复 store2, err : rsm.NewKVStore(/tmp/rsm-wal, /tmp/rsm-snap) if err ! nil { log.Fatalf(Failed to recreate KV store: %v, err) } defer store2.Close() // 验证数据 fmt.Println(恢复后的数据:) for _, key : range keys { value, exists : store2.Get(key) if exists { fmt.Printf( %s %s ✓\n, key, value) } else { fmt.Printf( %s not found ✗\n, key) } } // 6. 统计信息 stats : store2.GetStats() fmt.Printf(\n统计信息:\n) fmt.Printf( 读取次数: %d\n, stats.ReadCount) fmt.Printf( 写入次数: %d\n, stats.WriteCount) fmt.Println(\n演示完成) }七、生产坑位#坑位现象解决方案1快照过大全量快照导致磁盘 IO 暴涨、网络带宽打满增量快照分片快照限速2WAL 无限增长磁盘空间耗尽定期快照后截断 WAL设置 WAL 保留策略3读请求读到过期数据Follower 返回旧数据ReadIndex / LeaseRead强制读 Leader4快照期间的写阻塞创建快照时长时间锁住状态机COWCopy-on-Write快照异步快照5Follower 追赶太慢Follower 落后太多重放日志耗时过长直接安装快照而不是重放日志6状态机非确定性操作不同节点执行结果不同如 time.Now()所有非确定性操作必须在 Leader 侧完成将结果作为命令参数传入八、本讲总结┌──────────────────────────────────────────────────────────────────────────┐ │ 第4讲核心收获 │ │ │ │ ✅ 理解了复制状态机的核心原理 │ │ - 相同初始状态相同命令序列相同最终状态 │ │ - 共识模块保证命令序列一致 │ │ │ │ ✅ 实现了 WAL预写日志 │ │ - 分段管理防止单个文件过大 │ │ - 编码/解码JSON 长度前缀 │ │ - 刷盘保证持久性 │ │ │ │ ✅ 实现了快照机制 │ │ - 定期保存状态机快照 │ │ - 压缩存储节省空间 │ │ - 快照后截断 WAL │ │ │ │ ✅ 实现了 KV 存储引擎 │ │ - 从快照和 WAL 恢复 │ │ - 线性一致读ReadIndex / LeaseRead │ │ - 统计信息跟踪 │ │ │ │ ✅ 覆盖了 6 个生产坑位与解决方案 │ │ │ └──────────────────────────────────────────────────────────────────────────┘九、下一讲预告第5讲《分布式事务从 2PC 到 TCC》你将学到XA 协议与两阶段提交的缺陷TCCTry-Confirm-Cancel模式Saga 模式编排式与协同式Seata 实战分布式事务的最终一致性保障开发之余的小工具推荐在调试复制状态机的快照和 WAL 恢复时zz365.top 的 JSON 格式化工具可以将快照文件和 WAL 日志导出为结构化数据便于比对不同节点的状态差异。其文本差异比较工具则可以快速定位快照与 WAL 之间的数据断层。所有处理均在浏览器本地完成不会泄露你的业务数据。

阅读完成 · 觉得有帮助？

咨询建站