跳转至

计数布隆过滤器

💡 一句话概述

Counting Bloom Filter 将标准布隆过滤器的位数组升级为计数器数组，从而原生支持删除操作，代价是约 4 倍的空间开销。

🔑 核心概念

  1. 计数器数组（Counter Array）：每个槽位不再只存 0/1，而是存一个计数值（通常 4 bit 或 8 bit）。
  2. 插入即 +1：元素映射的 k 个槽位计数器各自加一。
  3. 删除即 -1：元素映射的 k 个槽位计数器各自减一。
  4. 查询 > 0：所有映射槽位的计数器均大于零时，判定元素可能存在。

📝 详细说明

与标准布隆过滤器对比

graph LR
    subgraph SBF["标准布隆过滤器"]
        direction TB
        SB["位数组"] --> SI["插入: 置 1"]
        SB --> SQ["查询: 全为 1?"]
        SB --> SD["删除: ❌"]
    end

    subgraph CBF["计数布隆过滤器"]
        direction TB
        CB["计数器数组"] --> CI["插入: +1"]
        CB --> CQ["查询: 全 > 0?"]
        CB --> CD["删除: -1 ✅"]
    end

    SBF -.->|"约 4x 空间"| CBF
特性 标准布隆过滤器 计数布隆过滤器
存储单位 1 bit/槽 4 bit（或 8 bit）/槽
空间开销 1x ~4x（4 bit 计数器）
插入 置 1 计数器 +1
删除 ❌ 计数器 -1 ✅
查询 所有位 = 1？ 所有计数器 > 0？
假阳性 有 有（相同概率）
假阴性 无 无

删除原理

以 4 bit 计数器为例（每个槽位可表示 0~15）：

初始状态: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]

插入 A（h1=1, h2=5, h3=9）:
[0, 1, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0]

插入 B（h1=1, h2=3, h3=7）:    ← 注意 h1 与 A 冲突
[0, 2, 0, 1, 0, 1, 0, 1, 0, 1, 0, 0]

删除 A（h1=1, h2=5, h3=9）→ 各计数器 -1:
[0, 1, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0]

查询 B（h1=1→1, h2=3→1, h3=7→1）→ 全部 > 0 ✅ 仍能正确判断存在

对比标准布隆过滤器：如果直接将 bit[1] 清零，B 的查询就会产生假阴性。

计数器溢出风险

4 bit 计数器最大值为 15。如果同一个槽位被超过 15 个元素命中，计数器将溢出归零，导致后续操作出错。

实际风险极低：对于 k 个哈希函数、m 个槽位的布隆过滤器，某个特定槽位的计数期望为 \(n \cdot k / m\)。以 100 万元素、1% 误判率为例（m ≈ 958 万、k ≈ 7），单槽计数期望仅约 0.73，远低于 15。

保险策略：

  • 使用 4 bit 计数器（0~15）适合绝大多数场景
  • 极端密集场景使用 8 bit 计数器（0~255）
  • 计数器达到上限时视为错误，触发告警或全量重建

💻 代码示例

Go 实现

package countingbloom

import (
    "hash/fnv"
    "math"
)

type CountingBloomFilter struct {
    counters []uint8  // 计数器数组，每个用 4 bit（低 4 位）
    k        int      // 哈希函数个数
    m        uint     // 计数器个数
}

// New 根据预期元素数和误判率创建计数布隆过滤器
func New(expectedItems uint, falsePositiveRate float64) *CountingBloomFilter {
    m := optimalM(expectedItems, falsePositiveRate)
    k := optimalK(m, expectedItems)
    return &CountingBloomFilter{
        counters: make([]uint8, (m+1)/2), // 每个 uint8 存 2 个 4-bit 计数器
        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))
}

// getCounter 读取第 idx 个 4-bit 计数器
func (cbf *CountingBloomFilter) getCounter(idx uint) uint8 {
    byteIdx := idx / 2
    if idx%2 == 0 {
        return cbf.counters[byteIdx] & 0x0F // 低 4 位
    }
    return cbf.counters[byteIdx] >> 4 // 高 4 位
}

// setCounter 写入第 idx 个 4-bit 计数器
func (cbf *CountingBloomFilter) setCounter(idx uint, val uint8) {
    byteIdx := idx / 2
    if idx%2 == 0 {
        cbf.counters[byteIdx] = (cbf.counters[byteIdx] & 0xF0) | (val & 0x0F)
    } else {
        cbf.counters[byteIdx] = (cbf.counters[byteIdx] & 0x0F) | ((val & 0x0F) << 4)
    }
}

// Add 插入元素
func (cbf *CountingBloomFilter) Add(data []byte) {
    for i := 0; i < cbf.k; i++ {
        pos := cbf.hash(data, i)
        cur := cbf.getCounter(pos)
        if cur < 15 { // 防止溢出
            cbf.setCounter(pos, cur+1)
        }
    }
}

// Remove 删除元素
func (cbf *CountingBloomFilter) Remove(data []byte) {
    for i := 0; i < cbf.k; i++ {
        pos := cbf.hash(data, i)
        cur := cbf.getCounter(pos)
        if cur > 0 {
            cbf.setCounter(pos, cur-1)
        }
    }
}

// Contains 判断元素是否可能存在
func (cbf *CountingBloomFilter) Contains(data []byte) bool {
    for i := 0; i < cbf.k; i++ {
        pos := cbf.hash(data, i)
        if cbf.getCounter(pos) == 0 {
            return false
        }
    }
    return true
}

func (cbf *CountingBloomFilter) hash(data []byte, i int) uint {
    h1 := fnv.New32a()
    h1.Write(data)
    hash1 := h1.Sum32()
    h2 := fnv.New32()
    h2.Write(data)
    hash2 := h2.Sum32()
    return (uint(hash1) + uint(i)*uint(hash2)) % cbf.m
}

使用示例

package main

import (
    "fmt"

    cbloom "your-module/countingbloom"
)

func main() {
    // 100万元素，1% 误判率
    cbf := cbloom.New(1_000_000, 0.01)

    // 插入
    cbf.Add([]byte("user:1001"))
    cbf.Add([]byte("user:1002"))
    cbf.Add([]byte("user:1003"))

    fmt.Println("user:1001 exists:", cbf.Contains([]byte("user:1001"))) // true
    fmt.Println("user:9999 exists:", cbf.Contains([]byte("user:9999"))) // false

    // 删除
    cbf.Remove([]byte("user:1001"))
    fmt.Println("user:1001 after delete:", cbf.Contains([]byte("user:1001"))) // false

    // 不存在的元素删除是安全的（计数器不会下溢）
    cbf.Remove([]byte("user:9999")) // 无副作用
}

⚠️ 常见陷阱

计数器溢出

4 bit 计数器上限为 15。当溢出发生时，后续删除操作无法正确递减，可能导致元素永远无法被"判不存在"。生产环境应监控计数器高位是否触及上限。

空间开销是标准布隆的 4 倍

4 bit 计数器 = 标准布隆每个 bit 位 × 4。100 万元素、1% 误判率的标准布隆约 1.14 MB，CBF 约 4.57 MB。如果空间敏感，考虑 Cuckoo Filter。

删除不存在的元素是"空操作"但不报错

对未插入的元素调用 Remove，计数器已经为 0 不会下溢，但也不会给出"元素不存在"的提示。业务层需要额外逻辑确保只在确认存在时才删除。

🏋️ 练习题

练习 1：100 万元素、1% 误判率的 CBF 需要多少内存？

标准布隆过滤器 m ≈ 958 万 bit ≈ 1.14 MB。CBF 每个计数器 4 bit，空间为 4 × 958 万 bit ≈ 4.57 MB。

答案

约 4.57 MB（4 bit 计数器场景）。是同参数标准布隆过滤器的约 4 倍。

练习 2：如果某个 4 bit 计数器已经等于 15，再插入一个映射到同一槽位的元素会发生什么？

计数器不能超过 15，忽略此次增量。后续删除该元素时计数器仍然 15，无法正确递减。

答案

计数器溢出。实现中通常做饱和处理（不再 +1），但这意味着该槽位"锁死"在 15——即使所有映射到该槽的元素都被删除，计数器也不会归零，导致假阳性无法消除。生产环境应监控饱和计数器数量，接近阈值时触发重建。

练习 3：为什么 CBF 不用 1 bit 计数器（即标准布隆过滤器）？1 bit 计数器"能减到 0"不也行吗？

1 bit 计数器值域只有 {0, 1}。两个元素映射到同一位后，该位为 1；删除其中一个后置 0，另一个元素就查不到了——这正是标准布隆过滤器不能删除的原因。

答案

1 bit 计数器无法区分"被几个元素占用"。值为 1 时可能是 1 个元素映射，也可能是多个元素映射；置 0 后所有占用该位的元素都会受影响。只有用 ≥2 bit 的计数器才能记录"有多少个元素占用了这个槽位"，从而安全地执行 -1 删除。

🔗 相关链接