From 1e733e13f00d89ac796625622e27771f75fc3d5c Mon Sep 17 00:00:00 2001 From: wonder Date: Mon, 24 Aug 2026 03:01:51 +0000 Subject: [PATCH] =?UTF-8?q?add:=20=E5=B8=83=E9=9A=86=E8=BF=87=E6=BB=A4?= =?UTF-8?q?=E5=99=A8=20&=20HeavyKeeper=20=E7=AC=94=E8=AE=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/algorithm/bloom-filter.md | 258 ++++++++++++++++++++++++ docs/algorithm/heavykeeper.md | 347 +++++++++++++++++++++++++++++++++ mkdocs.yml | 2 + 3 files changed, 607 insertions(+) create mode 100644 docs/algorithm/bloom-filter.md create mode 100644 docs/algorithm/heavykeeper.md diff --git a/docs/algorithm/bloom-filter.md b/docs/algorithm/bloom-filter.md new file mode 100644 index 0000000..7cd5cd8 --- /dev/null +++ b/docs/algorithm/bloom-filter.md @@ -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< 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 的多种方案对比 diff --git a/mkdocs.yml b/mkdocs.yml index c73f137..b7b925f 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -89,3 +89,5 @@ nav: - 缓存击穿雪崩穿透: architecture/cache/cache-breakdown-avalanche-penetration.md - 算法: - algorithm/index.md + - 布隆过滤器: algorithm/bloom-filter.md + - HeavyKeeper: algorithm/heavykeeper.md