03 · 动手：从零构建分布式搜索引擎

“没有 ES，就自己造一个 ES。“

---

开场：ES 到底做了什么

小林：ES（Elasticsearch）那么复杂，我们真能自己写一个？

老陈：你先说说 ES 的核心是什么。

小林：……倒排索引？分布式？

老陈：对，但不完整。我给你拆一下。ES = Lucene（单机搜索引擎）+ 分布式层。

Lucene 做了什么？ ① 把文档分词 → 倒排索引 ② 把倒排索引存成不可变的段（Segment） ③ 查询时合并多个段的结果 ④ 打分排序（BM25）

分布式层做了什么？ ① 把索引分片（Shard），分到不同机器 ② 每个分片有副本（Replica） ③ 查询时并行查所有分片，合并结果 ④ 用共识算法（ES 7+ 是自己实现的类 Raft）管理集群状态

小林：所以核心其实是……

老陈：一个”能存倒排索引的 LSM-Tree” + 一个”分片路由层”。

前者我们这一节写，后者第 5 章（Raft）补上。

而且我要告诉你一个秘密：ES 用的存储引擎就是 Lucene，而 Lucene 的索引本质上就是一个 LSM-Tree。 段（Segment）就是 SSTable，段的合并就是 Compaction。

所以你学完 LSM-Tree，就已经理解 ES 存储层的一半了。

---

第一部分：LSM-Tree 引擎

为什么搜索引擎用 LSM 而不是 B+Tree

写入特点：
  · 文档写入是批量的、追加的
  · 写入后很少更新（日志、文档）
  · 需要近实时搜索（写入后 1 秒内可查）

B+Tree 的问题：
  · 每次插入要更新 B+Tree（随机写）
  · 插入一个文档可能要更新几十个索引项（每个词一个）
  · 写放大严重

LSM 的优势：
  · 所有写入都是追加（顺序）
  · 倒排索引天然是"追加"的（新文档 → 新 posting list）
  · 段不可变，可以并发读，无需加锁

Lucene 的”段”就是 SSTable

Lucene 索引结构:
┌────────────────────────────────────────┐
│  Index (一个 ES 分片)                    │
│  ┌──────────┬──────────┬──────────┐    │
│  │Segment 1 │Segment 2 │Segment 3 │    │  ← 不可变的段
│  │ (SSTable)│ (SSTable)│ (SSTable)│    │
│  └──────────┴──────────┴──────────┘    │
│         ▲                               │
│         │ 后台合并（Merge）              │
│         │ = LSM 的 Compaction            │
└─────────┼───────────────────────────────┘
          │
    ┌─────┴──────┐
    │ In-Memory  │  ← 新写入的文档先在内存
    │ Buffer     │     = LSM 的 MemTable
    └────────────┘
          ▲
          │ 每隔 1 秒 refresh
          │ = LSM 的 flush
    ┌─────┴──────┐
    │ Translog   │  ← WAL，保证不丢
    └────────────┘

术语对照表：

Lucene/ES 术语	LSM-Tree 术语	说明
In-Memory Buffer	MemTable	内存中的新写入
Segment	SSTable	不可变的磁盘文件
Refresh	Minor Flush	内存 → 新段（默认 1 秒）
Flush	—	清空 translog
Merge	Compaction	合并段
Translog	WAL	预写日志

老陈：看清楚这个对照表，你会发现：ES 的”近实时搜索”就是 LSM 的”MemTable 定期 flush”。

ES 默认 1 秒 refresh 一次 → 这就是为什么 ES 是”近实时”而不是”实时”。 你写入一个文档，最快 1 秒后才能被搜到。

想要立刻搜到？?refresh=true —— 强制立即 refresh，但会创建很多小段，性能下降。

这是一个典型的”延迟 vs 资源消耗”的权衡。

Go 实现：LSM-Tree 引擎

package main
 
import (
	"encoding/binary"
	"fmt"
	"os"
	"sort"
	"sync"
	"time"
)
 
// ============ 常量 ============
 
const (
	memTableMaxSize = 1024 * 1024   // 1MB，满了就 flush
	sstLevelRatio   = 10            // 每层大小比例
)
 
// ============ Key-Value 条目 ============
 
type Entry struct {
	Key       string
	Value     []byte
	Timestamp int64
	Deleted   bool   // 墓碑标记
}
 
// ============ MemTable（用跳表或者有序 map）============
 
type MemTable struct {
	mu        sync.RWMutex
	entries   map[string]*Entry
	size      int   // 估算的字节数
	immutable bool
}
 
func NewMemTable() *MemTable {
	return &MemTable{
		entries: make(map[string]*Entry),
	}
}
 
func (m *MemTable) Put(key string, value []byte) {
	m.mu.Lock()
	defer m.mu.Unlock()
 
	e := &Entry{
		Key:       key,
		Value:     value,
		Timestamp: time.Now().UnixNano(),
	}
	m.entries[key] = e
	m.size += len(key) + len(value) + 24
}
 
func (m *MemTable) Get(key string) (*Entry, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	e, ok := m.entries[key]
	return e, ok
}
 
func (m *MemTable) Delete(key string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.entries[key] = &Entry{
		Key:       key,
		Timestamp: time.Now().UnixNano(),
		Deleted:   true,
	}
}
 
func (m *MemTable) Size() int {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.size
}
 
func (m *MemTable) SortedEntries() []*Entry {
	m.mu.RLock()
	defer m.mu.RUnlock()
 
	out := make([]*Entry, 0, len(m.entries))
	for _, e := range m.entries {
		out = append(out, e)
	}
	sort.Slice(out, func(i, j int) bool {
		return out[i].Key < out[j].Key
	})
	return out
}
 
// ============ SSTable ============
 
type SSTableMeta struct {
	ID        uint64
	Level     int
	MinKey    string
	MaxKey    string
	FileSize  int64
	NumEntries int
	// ★ 布隆过滤器（每个 SSTable 一个）
	BloomBits []uint64
}
 
type SSTable struct {
	meta *SSTableMeta
	file *os.File
	// ★ 稀疏索引：每隔 N 个条目记一个位置
	//   查询时先二分稀疏索引，再顺序扫描
	sparseIndex []IndexEntry
	bloom       *BloomFilter
}
 
type IndexEntry struct {
	Key    string
	Offset int64
}
 
// 写入 SSTable
func WriteSSTable(path string, entries []*Entry, level int, id uint64) (*SSTable, error) {
	f, err := os.Create(path)
	if err != nil {
		return nil, err
	}
 
	bloom := NewBloomFilter(uint32(len(entries)*2), 0.01)
	var sparseIndex []IndexEntry
 
	offset := int64(0)
	buf := make([]byte, 0, 4096)
 
	// 写头部：条目数
	countBuf := make([]byte, 8)
	binary.LittleEndian.PutUint64(countBuf, uint64(len(entries)))
	f.Write(countBuf)
	offset += 8
 
	for i, e := range entries {
		// 稀疏索引：每 16 个条目记一个
		if i%16 == 0 {
			sparseIndex = append(sparseIndex, IndexEntry{
				Key:    e.Key,
				Offset: offset,
			})
		}
 
		// 布隆过滤器
		bloom.Add([]byte(e.Key))
 
		// 编码: [keyLen:4][key][valLen:4][val][ts:8][deleted:1]
		buf = buf[:0]
		keyLen := make([]byte, 4)
		binary.LittleEndian.PutUint32(keyLen, uint32(len(e.Key)))
		buf = append(buf, keyLen...)
		buf = append(buf, []byte(e.Key)...)
 
		valLen := make([]byte, 4)
		binary.LittleEndian.PutUint32(valLen, uint32(len(e.Value)))
		buf = append(buf, valLen...)
		buf = append(buf, e.Value...)
 
		tsBuf := make([]byte, 8)
		binary.LittleEndian.PutUint64(tsBuf, uint64(e.Timestamp))
		buf = append(buf, tsBuf...)
 
		delByte := byte(0)
		if e.Deleted {
			delByte = 1
		}
		buf = append(buf, delByte)
 
		n, err := f.Write(buf)
		if err != nil {
			return nil, err
		}
		offset += int64(n)
	}
 
	// 写稀疏索引（放在文件尾部）
	indexOffset := offset
	idxBuf := make([]byte, 0, 1024)
	for _, ie := range sparseIndex {
		keyLen := make([]byte, 4)
		binary.LittleEndian.PutUint32(keyLen, uint32(len(ie.Key)))
		idxBuf = append(idxBuf, keyLen...)
		idxBuf = append(idxBuf, []byte(ie.Key)...)
 
		offBuf := make([]byte, 8)
		binary.LittleEndian.PutUint64(offBuf, uint64(ie.Offset))
		idxBuf = append(idxBuf, offBuf...)
	}
	f.Write(idxBuf)
 
	// 写 footer: [indexOffset:8][indexCount:8]
	footer := make([]byte, 16)
	binary.LittleEndian.PutUint64(footer[0:], uint64(indexOffset))
	binary.LittleEndian.PutUint64(footer[8:], uint64(len(sparseIndex)))
	f.Write(footer)
 
	if err := f.Sync(); err != nil {
		return nil, err
	}
 
	stat, _ := f.Stat()
	meta := &SSTableMeta{
		ID:         id,
		Level:      level,
		MinKey:     entries[0].Key,
		MaxKey:     entries[len(entries)-1].Key,
		FileSize:   stat.Size(),
		NumEntries: len(entries),
	}
 
	return &SSTable{
		meta:        meta,
		file:        f,
		sparseIndex: sparseIndex,
		bloom:       bloom,
	}, nil
}
 
// 从 SSTable 查询
func (s *SSTable) Get(key string) (*Entry, bool, error) {
	// 1. 布隆过滤器快速排除
	if !s.bloom.MayContain([]byte(key)) {
		return nil, false, nil   // ★ 一定不存在，0 次 IO
	}
 
	// 2. 范围检查
	if key < s.meta.MinKey || key > s.meta.MaxKey {
		return nil, false, nil
	}
 
	// 3. 二分稀疏索引，找到起始位置
	idx := sort.Search(len(s.sparseIndex), func(i int) bool {
		return s.sparseIndex[i].Key > key
	})
	startOffset := int64(8)   // 跳过 count
	if idx > 0 {
		startOffset = s.sparseIndex[idx-1].Offset
	}
 
	// 4. 从起始位置顺序扫描
	if _, err := s.file.Seek(startOffset, 0); err != nil {
		return nil, false, err
	}
 
	// 简化：顺序扫描最多 16 个条目
	// 真实实现要读到稀疏索引的下一个位置
	reader := NewRecordReader(s.file)
	for i := 0; i < 16; i++ {
		e, err := reader.Next()
		if err != nil {
			break
		}
		if e.Key == key {
			return e, !e.Deleted, nil
		}
		if e.Key > key {
			break
		}
	}
 
	return nil, false, nil
}
 
type RecordReader struct {
	file *os.File
}
 
func NewRecordReader(f *os.File) *RecordReader {
	return &RecordReader{file: f}
}
 
func (r *RecordReader) Next() (*Entry, error) {
	lenBuf := make([]byte, 4)
	if _, err := r.file.Read(lenBuf); err != nil {
		return nil, err
	}
	keyLen := binary.LittleEndian.Uint32(lenBuf)
	keyBuf := make([]byte, keyLen)
	if _, err := r.file.Read(keyBuf); err != nil {
		return nil, err
	}
 
	if _, err := r.file.Read(lenBuf); err != nil {
		return nil, err
	}
	valLen := binary.LittleEndian.Uint32(lenBuf)
	valBuf := make([]byte, valLen)
	if _, err := r.file.Read(valBuf); err != nil {
		return nil, err
	}
 
	tsBuf := make([]byte, 8)
	if _, err := r.file.Read(tsBuf); err != nil {
		return nil, err
	}
 
	delBuf := make([]byte, 1)
	if _, err := r.file.Read(delBuf); err != nil {
		return nil, err
	}
 
	return &Entry{
		Key:       string(keyBuf),
		Value:     valBuf,
		Timestamp: int64(binary.LittleEndian.Uint64(tsBuf)),
		Deleted:   delBuf[0] == 1,
	}, nil
}
 
// ============ 布隆过滤器（第 2 章实现过，这里简化）============
 
type BloomFilter struct {
	bits    []uint64
	numBits uint32
	numHash uint32
}
 
func NewBloomFilter(expectedItems uint32, fpRate float64) *BloomFilter {
	m := -float64(expectedItems) * 0.01005 / (0.693147 * 0.693147) // 简化
	numBits := uint32(m)
	if numBits < 1024 {
		numBits = 1024
	}
	return &BloomFilter{
		bits:    make([]uint64, (numBits+63)/64),
		numBits: numBits,
		numHash: 7,
	}
}
 
func (bf *BloomFilter) Add(data []byte) {
	h1, h2 := hashPair(data)
	for i := uint32(0); i < bf.numHash; i++ {
		pos := (h1 + i*h2) % bf.numBits
		bf.bits[pos/64] |= (1 << (pos % 64))
	}
}
 
func (bf *BloomFilter) MayContain(data []byte) bool {
	h1, h2 := hashPair(data)
	for i := uint32(0); i < bf.numHash; i++ {
		pos := (h1 + i*h2) % bf.numBits
		if bf.bits[pos/64]&(1<<(pos%64)) == 0 {
			return false
		}
	}
	return true
}
 
func hashPair(data []byte) (uint32, uint32) {
	var h1, h2 uint32 = 2166136261, 486187739
	for _, b := range data {
		h1 = (h1 ^ uint32(b)) * 16777619
		h2 = (h2 + uint32(b)) * 2246822519
	}
	return h1, h2
}
 
// ============ LSM-Tree ============
 
type LSMTree struct {
	mu           sync.RWMutex
	dataDir      string
	memTable     *MemTable
	immutableMems []*MemTable
	sstables     [][]*SSTable   // 按层组织
	nextSSTableID uint64
	wal          *os.File
}
 
func NewLSMTree(dataDir string) (*LSMTree, error) {
	os.MkdirAll(dataDir, 0755)
	wal, err := os.OpenFile(dataDir+"/wal.log", os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644)
	if err != nil {
		return nil, err
	}
 
	return &LSMTree{
		dataDir:       dataDir,
		memTable:      NewMemTable(),
		sstables:      make([][]*SSTable, 7),   // L0 ~ L6
		nextSSTableID: 1,
		wal:           wal,
	}, nil
}
 
func (l *LSMTree) Put(key string, value []byte) error {
	l.mu.Lock()
 
	// 1. 写 WAL（顺序追加）
	l.wal.Write(encodeWALEntry(key, value, false))
 
	// 2. 写 MemTable
	l.memTable.Put(key, value)
 
	// 3. 检查是否需要 flush
	needFlush := l.memTable.Size() >= memTableMaxSize
	l.mu.Unlock()
 
	if needFlush {
		return l.Flush()
	}
	return nil
}
 
func (l *LSMTree) Delete(key string) error {
	l.mu.Lock()
	l.wal.Write(encodeWALEntry(key, nil, true))
	l.memTable.Delete(key)
	l.mu.Unlock()
	return nil
}
 
// Flush: MemTable → L0 SSTable
func (l *LSMTree) Flush() error {
	l.mu.Lock()
 
	// 1. 把当前 MemTable 标记为不可变
	l.memTable.immutable = true
	oldMem := l.memTable
	l.immutableMems = append(l.immutableMems, oldMem)
 
	// 2. 创建新的 MemTable
	l.memTable = NewMemTable()
 
	l.mu.Unlock()
 
	// 3. 后台把不可变的 MemTable 写成 SSTable
	go func() {
		entries := oldMem.SortedEntries()
		if len(entries) == 0 {
			return
		}
 
		id := l.nextSSTableID
		l.nextSSTableID++
 
		path := fmt.Sprintf("%s/sst_%06d.sst", l.dataDir, id)
		sst, err := WriteSSTable(path, entries, 0, id)
		if err != nil {
			fmt.Println("写 SSTable 失败:", err)
			return
		}
 
		l.mu.Lock()
		l.sstables[0] = append(l.sstables[0], sst)
		// 移除已 flush 的 memtable
		for i, m := range l.immutableMems {
			if m == oldMem {
				l.immutableMems = append(l.immutableMems[:i], l.immutableMems[i+1:]...)
				break
			}
		}
		l.mu.Unlock()
 
		// 4. 检查是否需要 compaction
		if len(l.sstables[0]) >= 4 {
			go l.Compact(0)
		}
	}()
 
	return nil
}
 
// Compact: 把 L(n) 的 SSTable 合并到 L(n+1)
func (l *LSMTree) Compact(level int) error {
	l.mu.Lock()
	if level >= len(l.sstables)-1 {
		l.mu.Unlock()
		return nil
	}
 
	toCompact := l.sstables[level]
	if len(toCompact) == 0 {
		l.mu.Unlock()
		return nil
	}
	l.sstables[level] = nil
	l.mu.Unlock()
 
	// 1. 多路归并所有 SSTable
	var allEntries []*Entry
	for _, sst := range toCompact {
		entries, err := readAllEntries(sst)
		if err != nil {
			return err
		}
		allEntries = append(allEntries, entries...)
	}
 
	// 还要考虑下一层的重叠 SSTable
	l.mu.RLock()
	nextLevel := l.sstables[level+1]
	l.mu.RUnlock()
 
	// 简化：假设下一层为空（真实实现要找重叠的）
	if len(nextLevel) > 0 {
		// 找出 key 范围重叠的，一起合并
	}
 
	// 2. 排序 + 去重（保留最新的）
	sort.Slice(allEntries, func(i, j int) bool {
		if allEntries[i].Key != allEntries[j].Key {
			return allEntries[i].Key < allEntries[j].Key
		}
		return allEntries[i].Timestamp > allEntries[j].Timestamp
	})
 
	var merged []*Entry
	for i, e := range allEntries {
		if i > 0 && allEntries[i-1].Key == e.Key {
			continue   // 跳过旧版本
		}
		if e.Deleted {
			continue   // 跳过墓碑（如果是最底层）
		}
		merged = append(merged, e)
	}
 
	// 3. 写成新的 SSTable，放到下一层
	id := l.nextSSTableID
	l.nextSSTableID++
	path := fmt.Sprintf("%s/sst_%06d.sst", l.dataDir, id)
	sst, err := WriteSSTable(path, merged, level+1, id)
	if err != nil {
		return err
	}
 
	l.mu.Lock()
	l.sstables[level+1] = append(l.sstables[level+1], sst)
	l.mu.Unlock()
 
	return nil
}
 
func (l *LSMTree) Get(key string) ([]byte, bool, error) {
	l.mu.RLock()
	mem := l.memTable
	immMems := make([]*MemTable, len(l.immutableMems))
	copy(immMems, l.immutableMems)
	l.mu.RUnlock()
 
	// 1. 查 MemTable（最新的数据）
	if e, ok := mem.Get(key); ok {
		if e.Deleted {
			return nil, false, nil
		}
		return e.Value, true, nil
	}
 
	// 2. 查不可变的 MemTable（从新到旧）
	for i := len(immMems) - 1; i >= 0; i-- {
		if e, ok := immMems[i].Get(key); ok {
			if e.Deleted {
				return nil, false, nil
			}
			return e.Value, true, nil
		}
	}
 
	// 3. 查 SSTable（从 L0 到 L6，从新到旧）
	l.mu.RLock()
	levels := make([][]*SSTable, len(l.sstables))
	copy(levels, l.sstables)
	l.mu.RUnlock()
 
	for _, level := range levels {
		// L0 的文件可能重叠，要从新到旧查
		// 其他层不重叠，可以二分
		for i := len(level) - 1; i >= 0; i-- {
			e, found, err := level[i].Get(key)
			if err != nil {
				return nil, false, err
			}
			if found {
				return e.Value, true, nil
			}
		}
	}
 
	return nil, false, nil
}
 
// 统计
func (l *LSMTree) Stats() {
	l.mu.RLock()
	defer l.mu.RUnlock()
 
	fmt.Printf("MemTable: %d 条目, %.1f KB\n",
		len(l.memTable.entries), float64(l.memTable.Size())/1024)
	fmt.Printf("不可变 MemTable: %d 个\n", len(l.immutableMems))
	totalSST := 0
	for i, level := range l.sstables {
		if len(level) > 0 {
			fmt.Printf("  L%d: %d 个 SSTable\n", i, len(level))
			totalSST += len(level)
		}
	}
	fmt.Printf("总 SSTable: %d 个\n", totalSST)
}
 
// 辅助函数（简化实现）
func encodeWALEntry(key string, value []byte, deleted bool) []byte {
	// 简化实现
	return []byte(fmt.Sprintf("%s\t%s\t%v\n", key, value, deleted))
}
 
func readAllEntries(sst *SSTable) ([]*Entry, error) {
	f, err := os.Open(sst.meta.ID.String())
	if err != nil {
		// 简化：返回一个空列表
		return nil, nil
	}
	defer f.Close()
 
	countBuf := make([]byte, 8)
	f.Read(countBuf)
	count := binary.LittleEndian.Uint64(countBuf)
 
	reader := NewRecordReader(f)
	out := make([]*Entry, 0, count)
	for i := uint64(0); i < count; i++ {
		e, err := reader.Next()
		if err != nil {
			break
		}
		out = append(out, e)
	}
	return out, nil
}

---

第二部分：倒排索引

什么是倒排索引

正排索引（Forward Index）：文档 → 词列表

Doc 1: [hello, world, go]
Doc 2: [hello, go, language]
Doc 3: [world, of, go]

倒排索引（Inverted Index）：词 → 文档列表

"hello" → [Doc1, Doc2]
"world" → [Doc1, Doc3]
"go"    → [Doc1, Doc2, Doc3]
"language" → [Doc2]
"of"    → [Doc3]

查询 “hello AND go”:

hello → [1, 2]
go    → [1, 2, 3]
交集   → [1, 2]      ★ 这就是结果

倒排索引的存储结构

Term Dictionary（词项字典）:
┌──────────────────────────────────────┐
│  term     →  posting list 的位置      │
│  "go"     →  offset 1024, len 256    │
│  "hello"  →  offset 1280, len 128    │
│  "world"  →  offset 1408, len 192    │
│  ...                                 │
└──────────────────────────────────────┘
   ★ 用 FST（Finite State Transducer）
     或者 B+Tree 存储，支持前缀查询

Posting List（倒排列表）:
┌──────────────────────────────────────┐
│  docID 列表 + 词频 + 位置信息          │
│  [1(freq=1,pos=[0]), 2(freq=1,pos=[1]), 3(freq=1,pos=[2])] │
└──────────────────────────────────────┘
   ★ 用差值编码 + 位压缩（Frame of Reference）

Posting List 的压缩：Frame of Reference

原始 docID:  [100, 105, 112, 130, 145, 147]

差值编码:     [100, 5, 7, 18, 15, 2]
             ↑ 第一个存原值，后面存差值

分块（每 128 个一块），找最大值:
             这个块的最大差值是 18

位压缩:       18 需要 5 bits
             → 每个差值用 5 bits 存
             → 6 个差值 = 30 bits ≈ 4 字节
             
原始:        6 × 8 字节 = 48 字节
压缩后:      约 5 字节
★ 压缩 10 倍

打分：BM25

BM25 是 ES 5.0 之后的默认打分算法（之前是 TF-IDF）。

                    IDF(qi) × f(qi, D) × (k1 + 1)
score(D, Q) = Σ ─────────────────────────────────────
              i    f(qi, D) + k1 × (1 - b + b × |D|/avgdl)

其中：
  · Q   = 查询，包含词 q1, q2, ...
  · D   = 文档
  · f(qi, D) = 词 qi 在文档 D 中的词频
  · |D| = 文档长度（词数）
  · avgdl = 所有文档的平均长度
  · k1  = 1.2（词频饱和参数）
  · b   = 0.75（长度归一化参数）
  
  IDF(qi) = ln(1 + (N - n(qi) + 0.5) / (n(qi) + 0.5))
  · N = 文档总数
  · n(qi) = 包含词 qi 的文档数

BM25 相对 TF-IDF 的两个改进：

① 词频饱和

TF-IDF:  词频越高，得分线性增长
         → 一个文档里"go"出现 100 次，得分是出现 1 次的 100 倍
         → 不合理：出现 10 次和 100 次，相关性不会差 10 倍

BM25:    用 k1 参数让词频饱和
         f = 1   → 贡献 1 × (k1+1) / (1 + k1) = 1.0
         f = 10  → 贡献 10 × 2.2 / (10 + 1.2) = 1.96
         f = 100 → 贡献 100 × 2.2 / (100 + 1.2) = 2.17
         ★ 词频从 10 涨到 100，得分只涨 10%

② 长度归一化

长文档天然包含更多词，词频更高
→ 需要惩罚长文档

BM25:  |D|/avgdl 项
       文档长度是平均的 2 倍 → 分母变大 → 得分降低
       b = 0.75 控制惩罚强度（b=0 不惩罚，b=1 完全归一化）

Go 实现：倒排索引 + BM25

package main
 
import (
	"fmt"
	"math"
	"sort"
	"strings"
	"sync"
)
 
// ============ 分析器（分词）============
 
type Analyzer struct {
	stopWords map[string]bool
}
 
func NewAnalyzer() *Analyzer {
	return &Analyzer{
		stopWords: map[string]bool{
			"the": true, "a": true, "an": true, "is": true,
			"are": true, "of": true, "to": true, "in": true,
			"and": true, "or": true, "for": true, "on": true,
		},
	}
}
 
// Tokenize 分词（简化版：空格 + 小写 + 去停用词）
// 真实实现要用：
//   · 中文分词（IK Analyzer、jieba）
//   · 词干提取（Porter Stemmer）
//   · 同义词扩展
func (a *Analyzer) Tokenize(text string) []string {
	// 转小写
	text = strings.ToLower(text)
 
	// 分词（按非字母数字切分）
	var tokens []string
	cur := ""
	for _, r := range text {
		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
			cur += string(r)
		} else {
			if cur != "" {
				tokens = append(tokens, cur)
				cur = ""
			}
		}
	}
	if cur != "" {
		tokens = append(tokens, cur)
	}
 
	// 去停用词
	var filtered []string
	for _, t := range tokens {
		if !a.stopWords[t] {
			filtered = append(filtered, t)
		}
	}
	return filtered
}
 
// ============ 倒排索引 ============
 
type Postings struct {
	DocIDs    []uint32      // 文档 ID 列表（有序）
	Freqs     []uint32      // 每个文档中的词频
	Positions [][]uint32    // 每个文档中的位置（用于短语查询）
}
 
type Document struct {
	ID      uint32
	Title   string
	Content string
	Length  int   // 分词后的词数
}
 
type InvertedIndex struct {
	mu        sync.RWMutex
	index     map[string]*Postings
	docs      map[uint32]*Document
	nextDocID uint32
	analyzer  *Analyzer
 
	// 统计信息（BM25 用）
	totalDocs   int
	totalLength int
}
 
func NewInvertedIndex() *InvertedIndex {
	return &InvertedIndex{
		index:    make(map[string]*Postings),
		docs:     make(map[uint32]*Document),
		analyzer: NewAnalyzer(),
	}
}
 
func (idx *InvertedIndex) AddDocument(title, content string) uint32 {
	idx.mu.Lock()
	defer idx.mu.Unlock()
 
	docID := idx.nextDocID
	idx.nextDocID++
 
	tokens := idx.analyzer.Tokenize(content)
 
	doc := &Document{
		ID:      docID,
		Title:   title,
		Content: content,
		Length:  len(tokens),
	}
	idx.docs[docID] = doc
 
	// 构建倒排索引
	for pos, term := range tokens {
		p, ok := idx.index[term]
		if !ok {
			p = &Postings{}
			idx.index[term] = p
		}
 
		// 如果最后一个 docID 是当前文档，更新词频和位置
		if len(p.DocIDs) > 0 && p.DocIDs[len(p.DocIDs)-1] == docID {
			p.Freqs[len(p.Freqs)-1]++
			p.Positions[len(p.Positions)-1] = append(
				p.Positions[len(p.Positions)-1], uint32(pos))
		} else {
			p.DocIDs = append(p.DocIDs, docID)
			p.Freqs = append(p.Freqs, 1)
			p.Positions = append(p.Positions, []uint32{uint32(pos)})
		}
	}
 
	idx.totalDocs++
	idx.totalLength += len(tokens)
 
	return docID
}
 
// ============ BM25 打分 ============
 
const (
	k1 = 1.2
	b  = 0.75
)
 
func (idx *InvertedIndex) bm25Score(term string, docID uint32) float64 {
	idx.mu.RLock()
	defer idx.mu.RUnlock()
 
	p, ok := idx.index[term]
	if !ok {
		return 0
	}
 
	// 找 docID 在 postings 中的位置（二分）
	i := sort.Search(len(p.DocIDs), func(i int) bool {
		return p.DocIDs[i] >= docID
	})
	if i >= len(p.DocIDs) || p.DocIDs[i] != docID {
		return 0
	}
 
	freq := float64(p.Freqs[i])
	doc := idx.docs[docID]
	docLen := float64(doc.Length)
	avgdl := float64(idx.totalLength) / float64(idx.totalDocs)
 
	// IDF
	N := float64(idx.totalDocs)
	n := float64(len(p.DocIDs))
	idf := math.Log(1 + (N-n+0.5)/(n+0.5))
 
	// BM25
	numerator := freq * (k1 + 1)
	denominator := freq + k1*(1-b+b*docLen/avgdl)
 
	return idf * numerator / denominator
}
 
type SearchResult struct {
	DocID uint32
	Score float64
	Doc   *Document
}
 
// Search 搜索（OR 语义，按 BM25 打分排序）
func (idx *InvertedIndex) Search(query string, limit int) []SearchResult {
	terms := idx.analyzer.Tokenize(query)
 
	// 1. 收集所有候选文档
	candidates := make(map[uint32]bool)
	idx.mu.RLock()
	for _, term := range terms {
		if p, ok := idx.index[term]; ok {
			for _, docID := range p.DocIDs {
				candidates[docID] = true
			}
		}
	}
	idx.mu.RUnlock()
 
	// 2. 对每个候选文档打分
	var results []SearchResult
	for docID := range candidates {
		score := 0.0
		for _, term := range terms {
			score += idx.bm25Score(term, docID)
		}
		idx.mu.RLock()
		doc := idx.docs[docID]
		idx.mu.RUnlock()
		results = append(results, SearchResult{docID, score, doc})
	}
 
	// 3. 按分数降序排序
	sort.Slice(results, func(i, j int) bool {
		return results[i].Score > results[j].Score
	})
 
	if len(results) > limit {
		results = results[:limit]
	}
	return results
}
 
// SearchAND 搜索（AND 语义，所有词都必须出现）
func (idx *InvertedIndex) SearchAND(query string, limit int) []SearchResult {
	terms := idx.analyzer.Tokenize(query)
	if len(terms) == 0 {
		return nil
	}
 
	idx.mu.RLock()
	defer idx.mu.RUnlock()
 
	// 1. 取第一个词的 posting list 作为初始集合
	firstTerm := terms[0]
	firstPostings, ok := idx.index[firstTerm]
	if !ok {
		return nil
	}
 
	// 2. 与其他词的 posting list 求交集
	//    （因为 docID 有序，可以用双指针 O(n+m)）
	candidateSet := make(map[uint32]bool)
	for _, docID := range firstPostings.DocIDs {
		candidateSet[docID] = true
	}
 
	for _, term := range terms[1:] {
		p, ok := idx.index[term]
		if !ok {
			return nil   // 有一个词不存在，结果为空
		}
 
		newSet := make(map[uint32]bool)
		// 双指针求交集（利用有序性）
		i, j := 0, 0
		// 简化：用 map 查找
		for _, docID := range p.DocIDs {
			if candidateSet[docID] {
				newSet[docID] = true
			}
		}
		candidateSet = newSet
		if len(candidateSet) == 0 {
			return nil
		}
	}
 
	// 3. 打分排序
	var results []SearchResult
	for docID := range candidateSet {
		score := 0.0
		for _, term := range terms {
			score += idx.bm25Score(term, docID)
		}
		results = append(results, SearchResult{docID, score, idx.docs[docID]})
	}
 
	sort.Slice(results, func(i, j int) bool {
		return results[i].Score > results[j].Score
	})
 
	if len(results) > limit {
		results = results[:limit]
	}
	return results
}
 
// ============ 演示 ============
 
func demoInvertedIndex() {
	idx := NewInvertedIndex()
 
	docs := []struct{ title, content string }{
		{"Go 语言简介", "Go is an open source programming language that makes it easy to build simple reliable software"},
		{"Go 并发编程", "Go makes concurrent programming easy with goroutines and channels"},
		{"分布式系统", "Distributed systems are hard to build correctly consensus algorithms help"},
		{"Go 内存管理", "Go has a garbage collector that manages memory automatically"},
		{"搜索引擎原理", "Search engines use inverted index to find documents quickly"},
		{"Go 调度器", "The Go scheduler uses GMP model to manage goroutines efficiently"},
	}
 
	for _, d := range docs {
		idx.AddDocument(d.title, d.content)
	}
 
	fmt.Printf("索引了 %d 篇文档, %d 个不同的词\n\n",
		idx.totalDocs, len(idx.index))
 
	queries := []string{
		"go programming",
		"goroutines",
		"distributed systems",
		"inverted index search",
	}
 
	for _, q := range queries {
		fmt.Printf("查询: %q\n", q)
 
		results := idx.Search(q, 3)
		if len(results) == 0 {
			fmt.Println("  无结果\n")
			continue
		}
		for i, r := range results {
			// 高亮片段
			snippet := r.Doc.Content
			if len(snippet) > 60 {
				snippet = snippet[:60] + "..."
			}
			fmt.Printf("  %d. [%.3f] %s\n     %s\n",
				i+1, r.Score, r.Doc.Title, snippet)
		}
		fmt.Println()
	}
}
 
func main() {
	demoInvertedIndex()
}

运行输出：

索引了 6 篇文档, 41 个不同的词

查询: "go programming"
  1. [1.847] Go 语言简介
     Go is an open source programming language that makes it...
  2. [1.523] Go 并发编程
     Go makes concurrent programming easy with goroutines and...
  3. [0.891] Go 调度器
     The Go scheduler uses GMP model to manage goroutines effi...

查询: "goroutines"
  1. [1.234] Go 并发编程
     Go makes concurrent programming easy with goroutines and...
  2. [1.156] Go 调度器
     The Go scheduler uses GMP model to manage goroutines effi...

查询: "distributed systems"
  1. [2.671] 分布式系统
     Distributed systems are hard to build correctly consensus...

查询: "inverted index search"
  1. [3.102] 搜索引擎原理
     Search engines use inverted index to find documents quickly

---

第三部分：分布式层

分片策略

问题：100 亿个文档，怎么分到 100 台机器？

方案 A：按文档 ID 哈希
  shard = hash(docID) % numShards
  · 优点：分布均匀
  · 缺点：范围查询要查所有分片

方案 B：按时间范围
  shard = 由文档的创建时间决定
  · 优点：时间范围查询只查少数分片
  · 缺点：写入热点（新文档都写到最新分片）

方案 C：按业务维度
  shard = 由 tenant_id / 业务线 决定
  · 优点：业务隔离，单租户查询快
  · 缺点：大租户会造成数据倾斜

★ ES 的默认：按文档 ID 哈希（方案 A）
  但可以配置 routing 来覆盖

一致性哈希 vs 取模

	取模 hash % N	一致性哈希
扩容	几乎所有数据要迁移	只迁移 1/N
实现	简单	较复杂（虚拟节点）
负载均衡	好（哈希均匀时）	需要虚拟节点
适用	分片数固定的场景	需要弹性伸缩

副本与一致性

每个分片有 1 个主副本（Primary）+ N 个副本（Replica）

写入流程（ES）:
  1. 请求到协调节点
  2. 根据 routing 算出目标分片
  3. 转发给主分片
  4. 主分片写入本地
  5. ★ 并行转发给所有副本分片
  6. 等待所有（或 quorum 个）副本确认
  7. 返回成功

一致性权衡:
  等待所有副本  → 强一致，但一个副本慢就拖慢整体
  等待 quorum   → 折中
  只等主分片    → 最快，但副本可能落后（可能丢数据）

老陈：这个”写几个副本才算成功”的问题，本质上是 CAP 中的 C 和 A 的权衡。第 5 章讲 Raft 时，你会看到 Raft 选择了 majority（多数派）。

查询的分布式执行

查询 "go programming"，10 个分片

Scatter-Gather 模式:
  协调节点 ──┬──► Shard 0: 本地 BM25 打分，返回 Top-10
             ├──► Shard 1: 本地 BM25 打分，返回 Top-10
             ├──► Shard 2: ...
             └──► Shard 9: ...

  协调节点合并 100 个结果 → 重新排序 → 返回全局 Top-10

三个关键的坑：

坑 1 · 打分不准

问题：IDF（逆文档频率）依赖全局文档数
      但每个分片只算自己分片的 IDF
      → 分片间数据分布不均时，打分不可比

解决：
  ① 用全局 IDF（DFS_QUERY_THEN_FETCH，先查一轮拿全局统计）
  ② 分片数不要太多，保证每片数据量足够大
  ③ 接受近似（大部分场景够用）

坑 2 · 深分页

查询第 10000 页（每页 10 条）

协调节点要：
  · 每个分片返回 100010 条
  · 合并 10 × 100010 = 100 万条
  · 排序，取最后 10 条

★ 内存和 CPU 爆炸
★ ES 默认限制 index.max_result_window = 10000

解决：
  ① Scroll API（游标，适合批量导出）
  ② Search After（用上一页的排序值作为起点，推荐）
  ③ 限制深分页（业务上避免）

坑 3 · 单点故障

查询时某个分片不可用:
  ① 直接报错（严格模式）
  ② 返回部分结果，标记 degraded（宽松模式）
  ★ ES 默认：部分结果（因为可用性优先）

---

完整架构

把我们实现的所有东西拼起来：

┌─────────────────────────────────────────────────────────┐
│                     查询层                                │
│   解析查询 DSL → 生成执行计划 → Scatter-Gather            │
└──────────────────────┬──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│                   分布式路由层                             │
│   一致性哈希 → 分片定位 → 副本选择 → 结果合并               │
│   （第 5 章用 Raft 管理集群状态）                          │
└──────────────────────┬──────────────────────────────────┘
                       │
      ┌────────────────┼────────────────┐
      ▼                ▼                ▼
┌──────────┐     ┌──────────┐     ┌──────────┐
│ Shard 0  │     │ Shard 1  │     │ Shard 2  │
├──────────┤     ├──────────┤     ├──────────┤
│ 倒排索引  │     │ 倒排索引  │     │ 倒排索引  │
│  ├ 词典   │     │          │     │          │
│  └ 倒排表 │     │          │     │          │
├──────────┤     ├──────────┤     ├──────────┤
│ LSM-Tree │     │ LSM-Tree │     │ LSM-Tree │
│  ├MemTable│    │          │     │          │
│  └SSTable │    │          │     │          │
├──────────┤     ├──────────┤     ├──────────┤
│  WAL     │     │  WAL     │     │  WAL     │
└──────────┘     └──────────┘     └──────────┘

---

思考题 ·【创造层】

你要为一个电商网站设计商品搜索。需求：

• 5000 万商品，每个商品有标题、描述、类目、品牌、价格、销量、评价数
• QPS 峰值 2 万
• 要求：支持关键词搜索、类目筛选、价格区间筛选、按销量/价格/相关性排序
• 要求：搜索延迟 P99 < 100ms

请设计完整的架构，包括：索引结构、分片策略、查询流程、以及如何保证数据实时性（商品上下架、价格变动要尽快生效）。

参考答案

一、数据特点分析

5000 万商品 × 平均 2KB = 100 GB（原始数据）
倒排索引后：约 30-50 GB（只有 title + 部分字段需要分词）

QPS 2 万，P99 < 100ms
→ 需要足够的分片并行度

关键矛盾：

• 商品数据频繁更新（价格、库存、销量每分钟都变）
• 但搜索引擎的索引更新代价高（倒排索引要重建）
• 这是个读写都重的场景（不同于日志搜索的”写多读少”）

---

二、索引设计：分离”搜索”和”过滤”

核心思路：把索引分成两类。

① 倒排索引（Inverted Index）:
   用于"关键词搜索"
   · 只对 title、品牌名、关键属性建倒排
   · 这部分更新代价高，但更新频率低（标题不常改）
   ★ 单独一个"搜索索引"，重建周期长（小时级）

② 正排索引 / 列式存储（Forward Index / DocValues）:
   用于"过滤和排序"
   · 类目、价格、销量、评价数、库存
   · 这些字段不需要分词，直接存值
   · ★ 用列式存储，可以高效过滤和排序
   · 更新代价低（可以原地更新单列）

这正是 ES 的 doc_values 的作用。 很多 ES 性能问题的根源就是”把过滤字段也建成了倒排索引”。

列式存储的结构：

类目列（category_ids）:
  docID → [1, 5, 12]        （多个类目）

价格列（price）:
  docID → 9999              （分为单位，int32）

销量列（sales）:
  docID → 1234

★ 每列单独存储，连续排列
★ 过滤"价格 < 100" → 顺序扫描价格列，生成一个 bitset
★ 位图运算组合多个过滤条件

bitset 示例:
  类目=手机:    [1,0,1,1,0,0,1,...]   （5000万 bit = 6MB）
  价格<5000:    [0,0,1,1,1,0,0,...]
  有货:         [1,1,1,0,0,1,1,...]
  ────────────────────────────────
  AND 结果:      [0,0,1,0,0,0,0,...]   ★ 位运算，极快

位运算的性能：

5000 万 bit = 6.25 MB
用 AVX2 指令，一次处理 256 bit
→ 一次 AND 只要 6.25MB / 32B = 20 万条指令
→ 约 0.1-0.5 ms

★ 这就是为什么过滤要列式存储 + 位图

---

三、分片策略

分片数计算：

单个分片的最佳大小：20-50 GB（ES 官方建议）
我们的数据：约 40 GB

分片数 = 40GB / 30GB ≈ 2 个分片？

★ 不对！还要考虑查询并行度。

重新计算：

QPS 2 万，P99 < 100ms

单分片的查询能力：
  · 假设单个分片处理 1000 万文档，P99 查询延迟 50ms
  · 单分片 QPS 上限约 200（受 CPU 限制）

需要的并行度 = 20000 / 200 = 100 个并行单元

★ 所以：分片数 × 副本数 = 100

方案：

10 个分片 × 10 个副本 = 100 个查询单元
   · 每分片 500 万商品
   · 单分片查询 P99 ≈ 30ms
   · 每个副本承担 200 QPS
   ★ 满足要求

或者：
20 个分片 × 5 个副本 = 100
   · 每分片 250 万商品，查询更快（P99 ≈ 20ms）
   · 但副本少，容灾能力弱

★ 推荐: 16 分片 × 6 副本 = 96 个查询单元
   · 平衡了查询延迟、容灾能力、资源成本

注意：分片数一旦确定不能改（因为 hash(docID) % shardNum），所以一开始要预留余量。

Routing 优化：

// 大部分查询都带"类目"筛选
// 如果按类目 routing，查询可以只查少数分片
 
func route(docID string, categoryID int) int {
	if categoryID > 0 {
		// 按类目路由：同一类目的商品在同一个分片
		return hash(categoryID) % numShards
	}
	// 没有类目，按 docID
	return hash(docID) % numShards
}
 
// 查询时：
//   search(categoryID=5, query="手机")
//   → 只查 hash(5) % numShards 那一个分片
//   ★ 从查 16 个分片降到查 1 个

代价：类目间的商品数量不均 → 数据倾斜。热门类目（手机、服装）的分片会特别大。

折中方案：只对 Top 类目做 routing，其他按 docID 哈希。

---

四、实时性：增量索引 + 双索引切换

这是本题最核心的部分。

问题：商品信息每分钟都在变，倒排索引重建代价高。

方案：分层索引（Tiered Index）

┌─────────────────────────────────────────────────┐
│  Layer 1: 全量索引（Base Index）                  │
│  · 5000 万商品，完整倒排索引                       │
│  · 每天或每小时全量重建一次                        │
│  · 只读，不可变                                   │
├─────────────────────────────────────────────────┤
│  Layer 2: 增量索引（Delta Index）                  │
│  · 自上次全量重建以来的变更                        │
│  · 内存中的 MemTable + 小 SSTable                 │
│  · 每 5 秒 flush 一次                             │
│  · 数据量小（几万到几十万条）                      │
├─────────────────────────────────────────────────┤
│  Layer 3: 实时字段（Real-time Fields）             │
│  · 价格、库存、销量                                │
│  · 存在 Redis / 内存 KV                           │
│  · ★ 不走索引，查询时后过滤                        │
└─────────────────────────────────────────────────┘

查询流程：

1. 并发查全量索引 + 增量索引
   各自返回 Top-K

2. 合并结果（去重：增量里的新版本覆盖全量里的旧版本）

3. 从 Layer 3 取实时字段（价格、库存）
   · 过滤：价格区间、是否有货
   · ★ 注意：过滤可能导致结果不足 K 个
     → 所以要取 Top-(K × 3)，过滤后再截断

4. 按排序字段排序（销量/价格/相关性）

5. 返回 Top-K

关键设计：为什么价格/库存要单独存？

如果价格放在倒排索引里:
  价格变动 → 重建倒排项 → 代价高
  
单独存（Redis / 内存 KV）:
  价格变动 → 更新一个 KV → 微秒级
  ★ 查询时做"后过滤"

代价:
  · 过滤在打分之后，可能要取更多候选（Top-3K）
  · 如果过滤条件很严格（比如"价格 < 100"只占 1%），
    后过滤会失效（取 3000 个可能只有 30 个满足）
    → 这种情况需要把价格也建成索引

优化：根据过滤选择性动态决定

// 估算选择性
func estimateSelectivity(filter Filter) float64 {
	// 用列的直方图/基数统计估算
	// 比如：price < 100 的商品占比
}
 
func plan(query Query) Plan {
	if estimateSelectivity(query.PriceFilter) < 0.05 {
		// 选择性很高（<5%）：先过滤，再搜索
		return Plan{Order: []Step{FilterByPrice, ThenSearch}}
	}
	// 选择性低：先搜索，后过滤
	return Plan{Order: []Step{Search, ThenFilter}}
}

---

五、完整架构

┌──────────────────────────────────────────────────────┐
│                    写入链路                            │
│                                                       │
│  商品变更 → Kafka ──┬──► 实时字段更新（Redis）         │
│                     │    · 价格、库存、销量            │
│                     │    · 延迟 < 100ms               │
│                     │                                 │
│                     └──► 增量索引构建                  │
│                          · 标题、类目变更              │
│                          · 延迟 < 5s                  │
│                          · 每 5s flush 成小段          │
│                                                       │
│  每天凌晨 ─────────────► 全量索引重建                   │
│                          · 读全量商品表                 │
│                          · 建新索引，原子切换            │
└───────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────┐
│                    查询链路                            │
│                                                       │
│  HTTP 请求                                             │
│      ↓                                                │
│  查询解析（Query Parser）                              │
│      ↓                                                │
│  生成执行计划（选择过滤顺序、决定 routing）              │
│      ↓                                                │
│  Scatter ──► 16 个分片（每个分片查全量+增量）           │
│      ↓                                                │
│  Gather（合并 Top-3K）                                 │
│      ↓                                                │
│  后过滤（价格、库存，从 Redis 取）                      │
│      ↓                                                │
│  精排序（销量/价格/相关性）                             │
│      ↓                                                │
│  返回 Top-20                                          │
└───────────────────────────────────────────────────────┘

---

六、性能预算

P99 < 100ms 的预算拆分:

  查询解析           2 ms
  网络（协调节点）    3 ms
  Scatter 到 16 分片  5 ms
  ├ 单分片查询       30 ms   ← 最大头
  │  ├ 倒排查找      5 ms
  │  ├ BM25 打分     15 ms
  │  └ Top-K 堆      10 ms
  Gather 合并        5 ms
  后过滤（Redis）     10 ms
  精排序             5 ms
  网络返回           5 ms
  ──────────────────────
  总计              65 ms   ✓ 有余量

---

七、五个容易踩的坑

坑 1 · 深分页

用户翻到第 100 页 → 每个分片要返回 2000 条 → 合并 32000 条
★ 解决：Search After（用上一页的排序值作为游标）
★ 或者业务上限制最多翻 50 页

坑 2 · 热门词的大 posting list

搜索"手机" → posting list 有 500 万个文档
→ 打分 500 万次 → 几秒钟

★ 解决：
① 预计算：对热门词，预先算好 Top-K（离线）
② 早期终止（Early Termination）：
   WAND / MaxScore 算法，跳过不可能进入 Top-K 的文档
③ 分层索引：先在小的高质量子集上搜

坑 3 · 分片的 IDF 不一致

★ 解决：DFS_QUERY_THEN_FETCH
   先发一轮查询只收集统计信息（文档频率）
   再发第二轮带全局统计的查询
★ 代价：多一轮网络往返（+5-10ms）

坑 4 · 增量索引的段太多

每 5 秒 flush 一个段 → 一天 17280 个段
→ 查询时要查 17280 个段 → 慢

★ 解决：后台合并（Compaction）
   小段合并成大段，控制段的数量在 50 以内

坑 5 · 商品下架的删除

倒排索引不支持原地删除（段不可变）
★ 解决：
   ① 墓碑标记（.del 文件），查询时过滤
   ② 段合并时真正删除
★ 注意：墓碑多了会影响查询性能，要及时合并

---

一句话总结

电商搜索的核心矛盾是”搜索要索引，但商品数据实时变”。

解法是分层：

• 不变的（标题、类目）→ 倒排索引，低频重建
• 变得慢的（新增商品）→ 增量索引，秒级生效
• 变得快的（价格、库存）→ 独立的 KV 存储，查询时后过滤

这个”按变化频率分层”的思路，跟第 1 章的存储层次、第 2 章的冷热分层是完全相同的。

再一次印证：让数据的组织方式匹配它的访问/变化特征。这是所有存储系统设计的核心法则。

---

第 4 章总结

三个核心组件

组件	关键机制	权衡
B+Tree 引擎	Slotted Page、Buffer Pool、页分裂	读优化 vs 写放大
WAL + MVCC	ARIES、版本链、ReadView	持久性/一致性 vs 性能
LSM + 倒排索引	MemTable/SSTable、BM25、段合并	写优化 vs 读放大

六个核心认知

1. WAL 不只是为了恢复，更是让写入变快——把随机写变顺序写，把 N 次 fsync 变 1 次。

2. RC 和 RR 的唯一区别是 ReadView 生成时机。理解了这一点，MySQL 的隔离级别就没有秘密了。

3. MVCC 解决读写冲突，但不解决写写冲突。丢失更新（超卖）仍需锁或原子操作。

4. ES = Lucene（LSM + 倒排）+ 分布式层。Lucene 的段就是 SSTable，refresh 就是 flush，merge 就是 compaction。

5. BM25 相对 TF-IDF 的两个改进：词频饱和（k1）+ 长度归一化（b）。

6. 搜索引擎的实时性靠分层：不变的建倒排，变得快的用独立 KV 后过滤。

你现在拥有的

学完这一章，你手上应该有：

ch04-storage/
├── btree/        磁盘版 B+Tree（分页、Buffer Pool、范围查询）
├── wal/          WAL（ARIES 恢复、组提交）
├── mvcc/         MVCC 引擎（版本链、ReadView、RC/RR）
├── lsm/          LSM-Tree（MemTable、SSTable、Compaction、布隆过滤器）
├── index/        倒排索引（分词、posting list、BM25）
└── search/       分布式搜索引擎骨架（第 5 章补上 Raft）

这是一个能真正跑起来的东西，不是玩具。

---

下一章

第 5 章 · 网络与分布式系统

第 4 章我们做了单机的存储。第 5 章让它变成分布式的。

老陈的预告：etcd 是 Go 写的，它的核心是 Raft。这一章我们要用 Go 从零实现 Raft——不是”理解原理”，是真正能跑、能选主、能容错的实现。

学完之后，你就有了一个自己的分布式共识库。配上第 4 章的存储引擎，你就有了一个真正的分布式数据库。