add: 布隆过滤器 & HeavyKeeper 笔记

2026-08-24 03:01:51 +00:00
parent a6d950a4c8
commit 1e733e13f0
3 changed files with 607 additions and 0 deletions
@@ -0,0 +1,258 @@
+# 布隆过滤器
+
+!!! note "💡 一句话概述"
+    布隆过滤器是一种空间高效的概率数据结构，用于判断元素**是否可能存在**——说不存在则一定不存在，说存在则可能不存在。
+
+---
+
+## 🔑 核心概念
+
+1. **位数组（Bit Array）**：底层数据结构是一个 m 位的 bit 数组，初始全为 0。
+2. **多个哈希函数**：使用 k 个独立的哈希函数，每个元素映射到 k 个 bit 位。
+3. **假阳性（False Positive）**：不存在的元素可能被所有哈希函数命中的位恰好被其他元素置为 1，导致误判。
+4. **无假阴性（No False Negative）**：已插入的元素，其所有位一定为 1，判断"存在"时不会漏报。
+
+---
+
+## 📝 详细说明
+
+### 工作原理
+
+```
+插入元素 x：
+  h1(x) = 3   →  bit[3]  = 1
+  h2(x) = 7   →  bit[7]  = 1
+  h3(x) = 11  →  bit[11] = 1
+
+查询元素 y：
+  h1(y) = 3   →  bit[3]  = 1  ✓
+  h2(y) = 5   →  bit[5]  = 0  ✗  →  y 一定不存在
+
+查询元素 z：
+  h1(z) = 3   →  bit[3]  = 1  ✓  (被 x 置的)
+  h2(z) = 7   →  bit[7]  = 1  ✓  (被 x 置的)
+  h3(z) = 11  →  bit[11] = 1  ✓  (被 x 置的)
+  → z 可能存在（假阳性！实际未插入过）
+```
+
+### 假阳性率
+
+假阳性率由三个参数决定：位数组大小 `m`、哈希函数个数 `k`、已插入元素数 `n`：
+
+$$p \approx \left(1 - e^{-kn/m}\right)^k$$
+
+给定预期元素数 `n` 和可接受误判率 `p`，最优参数为：
+
+| 参数 | 公式 |
+|------|------|
+| 位数组大小 m | $m = -\frac{n \ln p}{(\ln 2)^2}$ |
+| 哈希函数个数 k | $k = \frac{m}{n} \ln 2$ |
+
+**示例**：100 万元素、1% 误判率 → m ≈ 958 万 bit（约 1.14 MB）、k ≈ 7。
+
+### 布隆过滤器 vs 其他结构
+
+| | 布隆过滤器 | HashSet | Counting Bloom Filter |
+|---|-----------|---------|----------------------|
+| 空间 | ★★★★★ 极省 | ★★ 需存完整元素 | ★★★ 约为布隆 4x |
+| 查询 | O(k) | O(1) 均摊 | O(k) |
+| 插入 | O(k) | O(1) 均摊 | O(k) |
+| 删除 | ❌ 不支持 | ✅ | ✅ |
+| 假阳性 | 有 | 无 | 有 |
+
+---
+
+## 💻 代码示例
+
+### 基础实现
+
+```go
+package bloom
+
+import (
+	"hash/fnv"
+	"math"
+)
+
+type BloomFilter struct {
+	bits []uint64    // 位数组，用 uint64 切片
+	k    int         // 哈希函数个数
+	m    uint        // 位数组总大小
+}
+
+// New 根据预期元素数和误判率创建布隆过滤器
+func New(expectedItems uint, falsePositiveRate float64) *BloomFilter {
+	m := optimalM(expectedItems, falsePositiveRate)
+	k := optimalK(m, expectedItems)
+	// 每个 uint64 有 64 个 bit
+	size := (m + 63) / 64
+	return &BloomFilter{
+		bits: make([]uint64, size),
+		k:    k,
+		m:    m,
+	}
+}
+
+func optimalM(n uint, p float64) uint {
+	return uint(-float64(n) * math.Log(p) / (math.Ln2 * math.Ln2))
+}
+
+func optimalK(m, n uint) int {
+	return int(math.Round(float64(m) / float64(n) * math.Ln2))
+}
+
+// Add 插入元素
+func (bf *BloomFilter) Add(data []byte) {
+	for i := 0; i < bf.k; i++ {
+		pos := bf.hash(data, i)
+		idx := pos / 64
+		bit := pos % 64
+		bf.bits[idx] |= 1 << bit
+	}
+}
+
+// Contains 判断元素是否可能存在
+func (bf *BloomFilter) Contains(data []byte) bool {
+	for i := 0; i < bf.k; i++ {
+		pos := bf.hash(data, i)
+		idx := pos / 64
+		bit := pos % 64
+		if bf.bits[idx]&(1<<bit) == 0 {
+			return false // 一定不存在
+		}
+	}
+	return true // 可能存在
+}
+
+// hash 使用双重哈希技巧生成第 i 个哈希值
+func (bf *BloomFilter) hash(data []byte, i int) uint {
+	h1 := fnv.New32a()
+	h1.Write(data)
+	hash1 := h1.Sum32()
+
+	h2 := fnv.New32()
+	h2.Write(data)
+	hash2 := h2.Sum32()
+
+	// h_i(x) = h1(x) + i * h2(x)
+	combined := uint(hash1) + uint(i)*uint(hash2)
+	return combined % bf.m
+}
+```
+
+### 使用示例：缓存穿透防护
+
+```go
+package main
+
+import (
+	"fmt"
+	"strconv"
+
+	"github.com/bits-and-blooms/bloom/v3"
+)
+
+func main() {
+	// 100万用户，1% 误判率
+	filter := bloom.NewWithEstimates(1_000_000, 0.01)
+
+	// 服务启动时：从 DB 加载所有合法用户 ID
+	userIDs := []int64{1, 2, 3, 100, 200, 500, 999}
+	for _, id := range userIDs {
+		filter.AddString(strconv.FormatInt(id, 10))
+	}
+
+	// 查询时：先过布隆过滤器
+	testIDs := []int64{1, 50, 100, 888, 999}
+	for _, id := range testIDs {
+		key := strconv.FormatInt(id, 10)
+		if !filter.TestString(key) {
+			fmt.Printf("用户 %d → 一定不存在，跳过 DB 查询\n", id)
+		} else {
+			fmt.Printf("用户 %d → 可能存在，继续查缓存/DB\n", id)
+		}
+	}
+}
+```
+
+### Redis 布隆过滤器
+
+```go
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/redis/go-redis/v9"
+)
+
+func main() {
+	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	ctx := context.Background()
+
+	// 创建布隆过滤器（需 RedisBloom 模块）
+	rdb.Do(ctx, "BF.RESERVE", "users:filter", 0.01, 1_000_000)
+
+	// 添加元素
+	rdb.Do(ctx, "BF.ADD", "users:filter", "user:1001")
+
+	// 批量添加
+	rdb.Do(ctx, "BF.MADD", "users:filter", "user:1002", "user:1003", "user:1004")
+
+	// 查询
+	exists, _ := rdb.Do(ctx, "BF.EXISTS", "users:filter", "user:1001").Bool()
+	fmt.Println("user:1001 exists:", exists) // true
+
+	exists, _ = rdb.Do(ctx, "BF.EXISTS", "users:filter", "user:9999").Bool()
+	fmt.Println("user:9999 exists:", exists) // 可能 false
+}
+```
+
+---
+
+## ⚠️ 常见陷阱
+
+!!! warning "不支持删除"
+    布隆过滤器无法删除元素——将位置 0 可能影响其他元素的判断。需要删除场景请用 Counting Bloom Filter（每 位改为计数器）或 Cuckoo Filter。
+
+!!! warning "容量超出后误判率飙升"
+    插入元素远超预期 `n` 时，假阳性率急剧上升。生产环境建议预留 20~30% 冗余，或使用 Scalable Bloom Filter 自动扩容。
+
+!!! warning "哈希函数必须独立均匀"
+    哈希函数质量差会导致分布不均匀、误判率高于理论值。推荐使用双重哈希（Double Hashing）或 MurmurHash3 系列。
+
+!!! warning "跨语言/跨系统序列化"
+    布隆过滤器的位数组直接序列化后，在另一端反序列化时必须使用**完全相同的 m、k、哈希函数**，否则查询结果无意义。
+
+---
+
+## 🏋️ 练习题
+
+??? question "练习 1：1000 万元素、0.1% 误判率需要多少内存？几个哈希函数？"
+    代入公式：m = -n·ln(p) / (ln2)² = -10⁷·ln(0.001) / (0.693)² ≈ 143.8M bit ≈ 17.1 MB，k = (m/n)·ln2 ≈ 10。
+
+    ??? success "答案"
+        位数组 m ≈ 1.438 亿 bit ≈ **17.1 MB**，哈希函数 k ≈ **10** 个。
+        相比 HashSet 存储 1000 万个 int64（约 76 MB），空间节省约 4.4x。
+
+??? question "练习 2：为什么说布隆过滤器「删除一个元素」很危险？"
+    删除需要把对应的 k 个位置零，但这些位可能被其他元素共享。置零后，其他本该"存在"的元素会变成"不存在"——产生了假阴性。
+
+    ??? success "答案"
+        布隆过滤器的多个元素共享 bit 位。删除元素 A 时把其 k 个位置 0，若元素 B 恰好也映射到其中某个位，B 的判断就会从"可能存在"变为"一定不存在"，违反了"无假阴性"的核心保证。解决方案：Counting Bloom Filter（4x 空间）、Cuckoo Filter（支持删除、空间更优）。
+
+??? question "练习 3：Scalable Bloom Filter 如何解决容量超限问题？"
+    SBF 在当前过滤器填满后，自动创建一个新的更大的布隆过滤器（容量按指数增长，误判率按比例收紧），查询时遍历所有子过滤器。总体误判率 = 各子过滤器误判率之积，渐进趋近于 0。
+
+    ??? success "答案"
+        SBF 维护一个布隆过滤器链：每个子过滤器容量递增、误判率递减。插入时写入当前活跃的过滤器；查询时遍历所有子过滤器，任一命中即返回"可能存在"。总体误判率是各子过滤器误判率的乘积，因此即使单个小过滤器误判率较高，整体仍可控。代价是查询时间随子过滤器数量线性增长。
+
+---
+
+## 🔗 相关链接
+
+- [Bloom Filter — Wikipedia](https://en.wikipedia.org/wiki/Bloom_filter) — 原理与数学推导
+- [go-bloom 库](https://github.com/bits-and-blooms/bloom) — Go 布隆过滤器实现
+- [RedisBloom 模块](https://redis.io/docs/stack/bloom/) — Redis 原生布隆过滤器
+- [Cuckoo Filter](https://www.cs.cmu.edu/~dga/papers/cuckoo-conext2014.pdf) — 支持删除的替代方案
@@ -0,0 +1,347 @@
+# HeavyKeeper
+
+!!! note "💡 一句话概述"
+    HeavyKeeper 是一种基于概率的数据结构，用极小空间在高速数据流中识别并近似统计**大象流（Heavy Hitters）**，同时通过衰减机制淘汰老鼠流。
+
+---
+
+## 🔑 核心概念
+
+1. **大象流 vs 老鼠流**：大象流是出现频次远高于平均的流（如热门 URL、攻击源 IP），老鼠流是频次很低的流。识别大象流是网络监控和异常检测的核心问题。
+2. **衰减计数器（Exponential Decay）**：每次访问计数器时以概率 `p` 将计数值减 1，让低频流自然衰减归零，高频流因持续到达而稳定在高位。
+3. **多级哈希表**：使用 `d` 行、每行 `w` 个桶，每个桶存储流 ID + 衰减计数器。同一流 ID 在多行中可能存在冲突，取最大计数器值作为估计。
+
+---
+
+## 📝 详细说明
+
+### 背景：为什么需要 HeavyKeeper
+
+| 方案 | 优势 | 劣势 |
+|------|------|------|
+| 精确计数（HashMap） | 100% 准确 | 内存随流数线性增长，无法应对海量高速流 |
+| Count-Min Sketch | 空间固定 | 无法区分大象流和老鼠流的累积误差，老鼠流噪声大 |
+| Space-Saving | 空间固定 | 需预知 top-k，淘汰策略对突发流量不友好 |
+| **HeavyKeeper** | 空间固定 + 衰减淘汰老鼠流 | 概率近似，有少量假阳性 |
+
+### 算法流程
+
+```
+数据结构：d × w 的二维表
+每个桶：(flowID, count)
+
+插入流 x：
+  for i = 0 to d-1:
+    pos = h_i(x) mod w
+    if bucket[pos].flowID == x:
+      bucket[pos].count++         // 命中：直接增加
+    elif bucket[pos].flowID == 空:
+      bucket[pos] = (x, 1)       // 空桶：直接插入
+    else:
+      // 冲突：以概率 p 衰减当前计数
+      if random() < p:
+        bucket[pos].count--
+        if bucket[pos].count == 0:
+          bucket[pos] = (x, 1)   // 衰减归零，替换为新流
+
+查询流 x 的频次：
+  return max{ bucket[h_i(x) mod w].count | bucket[h_i(x) mod w].flowID == x, i=0..d-1 }
+
+Top-k 查询：
+  收集所有桶中 count > threshold 的 (flowID, count)，排序取前 k
+```
+
+### 衰减概率的影响
+
+| 衰减概率 p | 效果 |
+|-----------|------|
+| p = 0 | 无衰减，退化为 Count-Min Sketch 变体 |
+| p 过大 | 大象流也被严重衰减，估计偏差大 |
+| p 适中（通常 0.01~0.1） | 老鼠流快速淘汰，大象流保持稳定 |
+
+**直觉**：老鼠流偶尔被衰减一次就从 1 → 0（淘汰），大象流的计数远大于 1，偶尔 -1 后很快 +1 补回。
+
+### 参数选择参考
+
+| 场景 | d（行数） | w（列数/桶数） | p（衰减概率） | 内存 |
+|------|----------|--------------|-------------|------|
+| 1Gbps 网络监控 | 5 | 65536 | 0.05 | ~2.5 MB |
+| 10Gbps 核心网络 | 5 | 131072 | 0.03 | ~5 MB |
+| Web 访问日志 | 4 | 32768 | 0.1 | ~1 MB |
+
+---
+
+## 💻 代码示例
+
+### 基础实现
+
+```go
+package heavykeeper
+
+import (
+	"hash/fnv"
+	"math/rand"
+)
+
+type Bucket struct {
+	FlowID []byte
+	Count  uint64
+}
+
+type HeavyKeeper struct {
+	table [][]Bucket
+	d     int     // 行数（哈希函数数）
+	w     int     // 每行桶数
+	p     float64 // 衰减概率
+}
+
+func New(d, w int, decayProbability float64) *HeavyKeeper {
+	table := make([][]Bucket, d)
+	for i := range table {
+		table[i] = make([]Bucket, w)
+	}
+	return &HeavyKeeper{
+		table: table,
+		d:     d,
+		w:     w,
+		p:     decayProbability,
+	}
+}
+
+func (hk *HeavyKeeper) hash(row int, data []byte) int {
+	h := fnv.New32a()
+	h.Write(data)
+	h.Write([]byte{byte(row)})
+	return int(h.Sum32()) % hk.w
+}
+
+// Insert 插入一个流
+func (hk *HeavyKeeper) Insert(flowID []byte) {
+	for i := 0; i < hk.d; i++ {
+		pos := hk.hash(i, flowID)
+		bucket := &hk.table[i][pos]
+
+		if bucket.Count == 0 {
+			// 空桶：直接插入
+			bucket.FlowID = make([]byte, len(flowID))
+			copy(bucket.FlowID, flowID)
+			bucket.Count = 1
+		} else if bytesEqual(bucket.FlowID, flowID) {
+			// 命中：计数 +1
+			bucket.Count++
+		} else {
+			// 冲突：以概率 p 衰减
+			if rand.Float64() < hk.p && bucket.Count > 0 {
+				bucket.Count--
+				if bucket.Count == 0 {
+					// 衰减归零：替换为新流
+					bucket.FlowID = make([]byte, len(flowID))
+					copy(bucket.FlowID, flowID)
+					bucket.Count = 1
+				}
+			}
+		}
+	}
+}
+
+// Query 查询流的估计频次，返回 0 表示未找到
+func (hk *HeavyKeeper) Query(flowID []byte) uint64 {
+	var maxCount uint64
+	for i := 0; i < hk.d; i++ {
+		pos := hk.hash(i, flowID)
+		bucket := &hk.table[i][pos]
+		if bytesEqual(bucket.FlowID, flowID) && bucket.Count > maxCount {
+			maxCount = bucket.Count
+		}
+	}
+	return maxCount
+}
+
+// TopK 返回估计频次最高的 k 个流
+func (hk *HeavyKeeper) TopK(k int) []Item {
+	seen := make(map[string]uint64)
+	for i := 0; i < hk.d; i++ {
+		for j := 0; j < hk.w; j++ {
+			b := &hk.table[i][j]
+			if b.Count > 0 {
+				key := string(b.FlowID)
+				if b.Count > seen[key] {
+					seen[key] = b.Count
+				}
+			}
+		}
+	}
+
+	items := make([]Item, 0, len(seen))
+	for id, count := range seen {
+		items = append(items, Item{FlowID: []byte(id), Count: count})
+	}
+	sortDesc(items)
+	if len(items) > k {
+		items = items[:k]
+	}
+	return items
+}
+
+type Item struct {
+	FlowID []byte
+	Count  uint64
+}
+
+func bytesEqual(a, b []byte) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func sortDesc(items []Item) {
+	for i := 1; i < len(items); i++ {
+		for j := i; j > 0 && items[j].Count > items[j-1].Count; j-- {
+			items[j], items[j-1] = items[j-1], items[j]
+		}
+	}
+}
+```
+
+### 实战：Nginx 访问日志 Top-K 热门 IP
+
+```go
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"net"
+	"os"
+	"strings"
+
+	"your-project/heavykeeper"
+)
+
+func main() {
+	// 4 行、65536 桶、5% 衰减 ≈ 2MB 内存
+	hk := heavykeeper.New(4, 65536, 0.05)
+
+	// 从 stdin 读取 Nginx access log
+	scanner := bufio.NewScanner(os.Stdin)
+	for scanner.Scan() {
+		line := scanner.Text()
+		ip := extractIP(line)
+		if ip != nil {
+			hk.Insert(ip)
+		}
+	}
+
+	// 输出 Top 10 访问量最高的 IP
+	top := hk.TopK(10)
+	fmt.Println("Top 10 IPs:")
+	for i, item := range top {
+		fmt.Printf("  %2d. %-15s ~%d requests\n", i+1, item.FlowID, item.Count)
+	}
+}
+
+func extractIP(line string) []byte {
+	// Nginx 默认格式: $remote_addr - ...
+	parts := strings.SplitN(line, " ", 2)
+	if len(parts) < 1 {
+		return nil
+	}
+	if net.ParseIP(parts[0]) == nil {
+		return nil
+	}
+	return []byte(parts[0])
+}
+```
+
+### 实时流式监控（带时间窗口）
+
+```go
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"your-project/heavykeeper"
+)
+
+func main() {
+	hk := heavykeeper.New(5, 32768, 0.05)
+
+	// 模拟流量：2 个大象流 + 大量老鼠流
+	go func() {
+		for {
+			// 大象流
+			hk.Insert([]byte("192.168.1.100"))
+			hk.Insert([]byte("10.0.0.5"))
+
+			// 老鼠流
+			for i := 0; i < 100; i++ {
+				hk.Insert([]byte(fmt.Sprintf("172.16.%d.%d", i/256, i%256)))
+			}
+		}
+	}()
+
+	// 每秒输出 Top-K
+	ticker := time.NewTicker(time.Second)
+	for range ticker.C {
+		top := hk.TopK(3)
+		fmt.Println("--- Top 3 ---")
+		for i, item := range top {
+			fmt.Printf("  %d. %s: ~%d\n", i+1, item.FlowID, item.Count)
+		}
+	}
+}
+```
+
+---
+
+## ⚠️ 常见陷阱
+
+!!! warning "衰减概率 p 需要调参"
+    p 太小则老鼠流衰减慢、占位久；p 太大则大象流也被压制、估计偏低。建议从 0.05 开始，根据流量特征微调。
+
+!!! warning "冲突导致大象流被替换"
+    极端情况下多个大象流哈希冲突，导致互相衰减替换。增加行数 `d` 可降低冲突概率，代价是内存翻倍。
+
+!!! warning "Top-K 结果可能遗漏"
+    某些大象流恰好被高频老鼠流抢占桶位，可能不在 Top-K 中。可通过增大桶数 `w` 或降低阈值来缓解。
+
+!!! warning "单次插入非 O(1)"
+    每次插入需要遍历 d 行计算哈希，d 通常 4~5，开销不大但仍需注意极高 QPS 场景。
+
+---
+
+## 🏋️ 练习题
+
+??? question "练习 1：为什么 HeavyKeeper 用衰减而非直接淘汰？"
+    直接淘汰（如 LRU）需要维护数据结构的顺序关系，复杂度高且无法自然区分大象流和老鼠流。衰减让老鼠流自然归零，大象流因持续到来而"免疫"偶发的 -1，无需额外状态。
+
+    ??? success "答案"
+        衰减是概率化的"软淘汰"：老鼠流计数为 1，一次衰减就从 1→0 被清出；大象流计数可能是 10000，一次 -1 无伤大雅且很快被后续 +1 补回。这种非对称性使得衰减天然过滤老鼠流、保留大象流，且实现只需一行 `if rand() < p { count-- }`。
+
+??? question "练习 2：将 HeavyKeeper 和布隆过滤器串联能解决什么问题？"
+    布隆过滤器负责"某流是否见过"，HeavyKeeper 负责"某流的频次"。串联后：先用布隆过滤器过滤全新流（不在 HeavyKeeper 中查询），再用 HeavyKeeper 统计已见流的频次，减少无效查询。
+
+    ??? success "答案"
+        布隆过滤器判断"流是否首次出现"：若一定未见过，直接放行无需查 HeavyKeeper；若可能见过，进 HeavyKeeper 查询/更新。这样 HeavyKeeper 的桶位只被重复出现的流占用，减少老鼠流抢占大象流桶位的冲突。两者互补：布隆过滤器零假阴性保证不漏，HeavyKeeper 衰减机制保证大象流不被老鼠流淹没。
+
+??? question "练习 3：如果需要对历史数据"忘却"（只统计最近 N 秒的流量），HeavyKeeper 的衰减概率 p 应如何设置？"
+    引入时间维度的衰减：设时间窗口 T 秒内期望将计数衰减为原来的 1/e，则每次操作的衰减概率 p 应满足 (1-p)^(ops_in_T) ≈ 1/e，其中 ops_in_T 是 T 秒内该桶的平均操作次数。
+
+    ??? success "答案"
+        静态 p 无法精确实现时间窗口衰减。更好的做法是改为**指数加权移动平均**：每次到达时 `count = count * α + 1`（0 < α < 1），α 越小遗忘越快；或引入全局时间戳，定期对所有桶做 `count *= β`（β < 1）。这样不再依赖操作频率，时间维度独立可控。
+
+---
+
+## 🔗 相关链接
+
+- [HeavyKeeper 论文](https://conferences.sigcomm.org/sigcomm/2018/files/papers/pim-workshop/HeavyKeeper.pdf) — SIGCOMM 2018 原始论文
+- [Count-Min Sketch](https://en.wikipedia.org/wiki/Count%E2%80%93min_sketch) — 基础频次估计结构
+- [Top-K 算法综述](https://www.cs.ucsb.edu/~suri/psdir/esa09.pdf) — 流式 Top-K 的多种方案对比
@@ -89,3 +89,5 @@ nav:
      - 缓存击穿雪崩穿透: architecture/cache/cache-breakdown-avalanche-penetration.md
  - 算法:
    - algorithm/index.md
+    - 布隆过滤器: algorithm/bloom-filter.md
+    - HeavyKeeper: algorithm/heavykeeper.md