当前位置：首页 > news >正文

Redis布隆过滤器的学习(六）

news 2025/7/20 12:34:20

一、布隆过滤器

1.1、为什么需要布隆过滤器

当需要判断一个元素是否在某个集合的业务场景，一般想到的是将集合中的所有元素保存起来，然后通过比较确定。链表、树、哈希表等数据结构都是这种思路，但是随着集合中的元素增加，存储空间也会线性增长，最终会达到这些数据结构的瓶颈，同时检索速度也会越来越慢。

比如现在数据库中存在50亿个电话号码，现有10万个电话号码，判断这10万个电话号码是否在数据库中。

如果用一般思路来做，以一个电话号码占8字节计算，50亿个电话号码就需要40GB内存，这个内存太大了；如果就放在磁盘上，对磁盘进行IO访问，速度又会慢许多了，所以就需要布隆过滤器

1.2、布隆过滤器是什么？

是由一个初值为0的bit数组和多个哈希函数构成，用来快速判断集合中是否存在某个元素。

***注意：***布隆过滤器是一种类似set的数据结构，只是统计结果在巨量数据下有点小瑕疵，不够完美，但能够接受。

1.3、设计思想：

目的	减少内存占用
方式	不保存数据信息，只是在内存中做一个是否存在的标记flag

1.4、布隆过滤器能干什么？

1. 高效地插入和查询，占用空间少，返回的结果是不确定性+不够完美(即一个元素如果判断结果，存在时，元素不一定存在，但是判断结果为不存在时，则一定不存在，因为哈希函数，会产生哈希冲突，会导致多个元素占同一个位置)
1. 布隆过滤器可以添加元素，但是不能删除元素，因为会涉及hash判断依据，删除元素会导致误判率增加(元素3，4，5占据同一个位置，删除元素3，会导致元素4和5误判为存在)

1.5、布隆过滤器原理：

原理：是由一个初值为0的bit数组和多个哈希函数构成，实质上是一个大型位数组和几个不同的无偏hash函数(无偏代表分布均匀)。
添加key时：使用多个hash函数对key进行计算，每个hash函数都会得到一个不同的位置，再对bit数组长度进行取模运算得到一个位置，最后将这个位置都置为1.
查询key时：只要有其中一位是0就表示这个key不存在，但如果都是1，则表示这个key可能存在。

注意： 正是基于布隆过滤器的快速检测特性，可以在把数据写入数据库时，使用布隆过滤器做个标记；当缓存缺失后，应用查询数据库时，可以通过查询布隆过滤器快速判断数据是否存在，如果不存在就不用再去查询数据库了；这样一来，即使发生缓存穿透，数据库也不会被频繁访问，也不会影响数据库正常运行。

1.6、模拟哈希冲突：

/** 模拟哈希冲突，只使用一种hash函数 **/
// 哈希表节点结构
struct HashNode 
{int key;int value;HashNode* next;HashNode(int k, int v) : key(k), value(v), next(nullptr) {}
};// 哈希表类
class HashTable 
{
private:vector<list<HashNode*>> table;int size;public:HashTable(int s) : size(s){table.resize(s);}// 哈希函数int hashFunction(int key) {return key % size;}// 插入键值对void insert(int key, int value) {int index = hashFunction(key);list<HashNode*>& bucket = table[index];// 检查键是否已存在for (auto it = bucket.begin(); it != bucket.end(); ++it) {if ((*it)->key == key) {(*it)->value = value;  // 更新值return;}}// 插入新节点HashNode* newNode = new HashNode(key, value);bucket.push_back(newNode);}// 打印哈希表内容void printTable(){for (int i = 0; i < size; ++i) {cout << "Position " << i << ": ";list<HashNode*>& bucket = table[i];for (auto it = bucket.begin(); it != bucket.end(); ++it) {cout << "(" << (*it)->key << ", " << (*it)->value << ") -> ";}cout << "nullptr" << endl;}}// 析构函数，释放内存~HashTable() {for (int i = 0; i < size; ++i) {list<HashNode*>& bucket = table[i];for (auto it = bucket.begin(); it != bucket.end(); ++it) {delete* it;}}}
};int main() 
{HashTable ht(5);  // 创建一个大小为5的哈希表ht.insert(1, 100);ht.insert(2, 200);ht.insert(3, 300);ht.insert(4, 400);ht.insert(5, 500);ht.insert(6, 150);ht.insert(10, 400);ht.printTable();return 0;
}

在这里插入图片描述

可以看到只有一个hash函数时，冲突的概率还是挺大的，所以需要多个hash函数，来减少冲突的概率。

1.7、布隆过滤器典型使用场景：

1. 解决缓存击穿和redis的bitmap结合使用
- 在缓存层和应用层之间增加布隆过滤器，将所有可能的查询key先写入到布隆过滤器中。当用户请求到来时，首先判断该key是否存在于布隆过滤器中，如果不在就直接返回不存在，不再访问数据库；如果在，才去查询redis，如果redis里面没有，再去查询MySQL数据库。
1. 黑名单校验：比如垃圾邮件的拦截
1. 白名单校验：比如安全连接网址

1.8、实现简易布隆过滤器：

假设有这么一个场景：

有10万个邮件账号
其中有5000个是黑名单账号
利用布隆过滤器，判断哪些邮件账号是黑名单，哪些不是
这就不添加对MySQL的查询，简化代码

/** 设计布隆类 **/
class BloomFilter {
public:BloomFilter(const std::string& redis_host, int redis_port,const std::string& key, const std::string& password = "",size_t expected_items = 10000, double false_positive_rate = 0.01);~BloomFilter();// 连接 Redisvoid connect_redis(const std::string& redis_host, int redis_port);// 向布隆过滤器中添加元素void add(const std::string& element);// 检查元素是否可能存在于布隆过滤器中bool searchKey(const std::string& element);// 获取布隆过滤器的统计信息void print_stats() const;private:// 计算最优的位图大小和哈希函数数量void calculate_optimal_parameters(size_t n, double p);// 初始化哈希函数void initialize_hash_functions();// 计算位位置size_t calculate_bit_position(const std::string& element, size_t hash_index) const;redisContext* m_redis_conn;             // Redis 连接std::string m_redis_key;                // Redis 键名const std::string m_redis_password;     // Redis 密码size_t m_bitmap_size;                   // 位图大小（位数）size_t m_num_hashes;                    // 哈希函数数量double m_false_positive_rate;           // 误判率std::vector<std::function<size_t(const std::string&)>> m_hash_functions;
};

/** 计算最优位图大小和哈希函数数量 **/void BloomFilter::calculate_optimal_parameters(size_t n, double p) 
{// 计算位图大小 m = - (n * ln(p)) / (ln(2)^2)m_bitmap_size = static_cast<size_t>(-(n * log(p)) / (log(2) * log(2)));// 确保位图大小至少为1if (m_bitmap_size == 0) m_bitmap_size = 1;// 计算哈希函数数量 k = (m / n) * ln(2)m_num_hashes = static_cast<size_t>(ceil((m_bitmap_size / static_cast<double>(n)) * log(2)));// 至少需要1个哈希函数if (m_num_hashes == 0) m_num_hashes = 1;m_false_positive_rate = p;
}/*** @brief 初始化哈希函数，哈希函数越多，位图空间越大，那么哈希冲突就会越低，误判率越低。*/
void BloomFilter::initialize_hash_functions() 
{// 使用不同的种子创建多个哈希函数for (size_t i = 0; i < m_num_hashes; ++i) {m_hash_functions.emplace_back([seed = i](const string& str) -> size_t {// 简单的哈希函数实现 (FNV-1a算法)const size_t prime = 0x100000001b3;size_t hash = 0xcbf29ce484222325 ^ seed;for (char c : str) {hash ^= static_cast<size_t>(c);hash *= prime;}return hash;});}
}/*** @brief 向布隆过滤器中添加元素** @param element 要添加的元素*/
void BloomFilter::add(const string& element) 
{for (size_t i = 0; i < m_num_hashes; ++i) {size_t bit_position = calculate_bit_position(element, i);redisCommand(m_redis_conn, "SETBIT %s %d 1", m_redis_key.c_str(), static_cast<int>(bit_position));}
}/*** @brief 检查元素是否可能存在于布隆过滤器中** @param element 要检查的元素* @return true 元素可能存在（可能有误判）* @return false 元素绝对不存在*/
bool BloomFilter::searchKey(const std::string& element)
{for (size_t i = 0; i < m_num_hashes; ++i) {size_t bit_position = calculate_bit_position(element, i);redisReply* reply = static_cast<redisReply*>(redisCommand(m_redis_conn, "GETBIT %s %d", m_redis_key.c_str(), static_cast<int>(bit_position)));if (reply == nullptr) {std::cerr << "Redis command failed" << std::endl;continue;}// 如果任何一个位为0，则元素一定不存在if (reply->integer == 0) {freeReplyObject(reply);return false;}freeReplyObject(reply);}// 所有位都为1，元素可能存在return true;
}

在这里插入图片描述