一、复制状态机（Replicated State Machine）原理

复制状态机是分布式共识算法的核心应用模式。它的本质是：让所有节点以相同的顺序执行相同的命令，从而到达相同的状态。

┌─────────────────────────────────────────────────────────────────┐ │ 复制状态机架构 │ │ │ │ Client │ │ │ │ │ ▼ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ 共识模块（Raft/Paxos） │ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │ │节点1│ │节点2│ │节点3│ ← 日志复制 │ │ │ │ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ │ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │ │日志 │ │日志 │ │日志 │ ← 一致的日志序列 │ │ │ │ │[1,2,3]│ │[1,2,3]│ │[1,2,3]│ │ │ │ │ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ └─────┼────────┼────────┼──────────────────────────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌───────┐ ┌───────┐ ┌───────┐ │ │ │状态机 │ │状态机 │ │状态机 │ ← 相同的最终状态 │ │ │x=5,y=3│ │x=5,y=3│ │x=5,y=3│ │ │ └───────┘ └───────┘ └───────┘ │ │ │ └─────────────────────────────────────────────────────────────────┘

核心保证：如果所有节点从相同的初始状态出发，按相同的顺序执行相同的命令，那么所有节点最终状态必然一致。

二、WAL（Write-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, 8+len(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 offset+8 > len(data) { break } length := binary.BigEndian.Uint64(data[offset : offset+8]) offset += 8 // 读取条目 if offset+int(length) > len(data) { break } entry, err := w.decodeEntry(data[offset : offset+int(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.appliedIndex+1, 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 Read）

Raft 的写操作需要走共识，但读操作如果不加控制，可能会读到过期数据。有三种方式实现线性一致读：

// 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", i+1, 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演示完成！") }

七、生产坑位

八、本讲总结

┌──────────────────────────────────────────────────────────────────────────┐ │ 第4讲 核心收获 │ │ │ │ ✅ 理解了复制状态机的核心原理： │ │ - 相同初始状态 + 相同命令序列 = 相同最终状态 │ │ - 共识模块保证命令序列一致 │ │ │ │ ✅ 实现了 WAL（预写日志）： │ │ - 分段管理（防止单个文件过大） │ │ - 编码/解码（JSON + 长度前缀） │ │ - 刷盘保证持久性 │ │ │ │ ✅ 实现了快照机制： │ │ - 定期保存状态机快照 │ │ - 压缩存储节省空间 │ │ - 快照后截断 WAL │ │ │ │ ✅ 实现了 KV 存储引擎： │ │ - 从快照和 WAL 恢复 │ │ - 线性一致读（ReadIndex / LeaseRead） │ │ - 统计信息跟踪 │ │ │ │ ✅ 覆盖了 6 个生产坑位与解决方案 │ │ │ └──────────────────────────────────────────────────────────────────────────┘

九、下一讲预告

第5讲：《分布式事务：从 2PC 到 TCC》

你将学到：

• XA 协议与两阶段提交的缺陷
• TCC（Try-Confirm-Cancel）模式
• Saga 模式（编排式与协同式）
• Seata 实战
• 分布式事务的最终一致性保障

🧰 开发之余的小工具推荐

在调试复制状态机的快照和 WAL 恢复时，zz365.top 的 JSON 格式化工具可以将快照文件和 WAL 日志导出为结构化数据，便于比对不同节点的状态差异。其文本差异比较工具则可以快速定位快照与 WAL 之间的数据断层。所有处理均在浏览器本地完成，不会泄露你的业务数据。