布隆过滤器使用bit数组映射关键字key，对于在一个超大的集合中判断是否存在某个key能够起到很好的效果。但是缺点很明显：容易误报。也就是本来不存在的key，可能告诉你它存在。

一、布隆过滤器

根据上图来说明布隆过滤器的原理 ：

1）布隆过滤有一个m位（这里是10个）的bit数组（或者称bitmap），bit数组初始化为全0，并且有k个（这里是3个）hash函数。

2）当我们分别存储abc，hello，world三个关键字，然后分别对这些key进行hash（每个hash都执行），得出结果如上图第三张表所示，然后对bit数组对应的bit位置1，置1后的结果如第二张表所示。

3）当我们对现有数据进行查询的时候，比如查询myworld，经过hash计算后得出0， 2，7，我们发现bit数组下标为2时，这一位为0，说明myworld一定不存在。再比如查询myabc，经过hash计算后得出0， 5，7，经过bit数组对比，发现0，5，7都是1，但是myabc却不存在，所以myabc属于误报！

总结：

1）布隆过滤器，对于不存在关键值能给出肯定结果，但是对于存在关键值只能说可能存在（存在一定误报）。这个主要是因为hash函数特征所决定的。

2）对于那么如何确定是否真的存在还是误报呢？应该交给业务模块再次进行确定查找。

3）如果保证误报率最低呢？有一个公式：k = （m/n）*ln2。（k为hash函数个数，m为bit数组大小，n为插入元素个数，ln2约等于0.69），当存在k个hash函数时误报率最低。至于为什么是这个公式，大家可自行查询相关资料。

二、leveldb中的布隆过滤器

2.1、构造方法

private: size_t bits_per_key_; size_t k_; /* 表示需要k_个hash函数 最大30个*/ public: explicit BloomFilterPolicy(int bits_per_key) : bits_per_key_(bits_per_key) { // We intentionally round down to reduce probing cost a little bit k_ = static_cast<size_t>(bits_per_key * 0.69); // 0.69 =~ ln(2) if (k_ < 1) k_ = 1; if (k_ > 30) k_ = 30; }

2.2、CreateFilter

/** * 生成过滤器 * @param keys 数组 * @param n 数组keys元素个数 * @param dst 保存bitmap */ virtual void CreateFilter(const Slice* keys, int n, std::string* dst) const { // Compute bloom filter size (in both bits and bytes) size_t bits = n * bits_per_key_; //表示bit数组中包含多少位 // For small n, we can see a very high false positive rate. Fix it // by enforcing a minimum bloom filter length. if (bits < 64) bits = 64;//避免误报率过高 size_t bytes = (bits + 7) / 8;//8bit对齐 bits = bytes * 8; const size_t init_size = dst->size(); dst->resize(init_size + bytes, 0); dst->push_back(static_cast<char>(k_)); // Remember # of probes in filter char* array = &(*dst)[init_size]; for (int i = 0; i < n; i++) { // Use double-hashing to generate a sequence of hash values. // See analysis in [Kirsch,Mitzenmacher 2006]. // 对一个key进行两次hash uint32_t h = BloomHash(keys[i]); const uint32_t delta = (h >> 17) | (h << 15); // Rotate right 17 bits //对于一个hash值（这里的hash可能很大数字）,通过for循环模拟k个hash场景 for (size_t j = 0; j < k_; j++) { const uint32_t bitpos = h % bits;//对bit数组求余 获取存储的bit array[bitpos/8] |= (1 << (bitpos % 8)); h += delta; } } }

2.3、KeyMayMatch

/** * 匹配key是存在当前过滤器中 * @param key 关键值 * @param bloom_filter 过滤器 * @return false 表示不存在 * true 表示key可能存在 是否真的存在还需要对key进行比较 */ virtual bool KeyMayMatch(const Slice& key, const Slice& bloom_filter) const { const size_t len = bloom_filter.size(); if (len < 2) return false; const char* array = bloom_filter.data(); const size_t bits = (len - 1) * 8; // Use the encoded k so that we can read filters generated by // bloom filters created using different parameters. const size_t k = array[len-1]; if (k > 30) { // Reserved for potentially new encodings for short bloom filters. // Consider it a match. return true; } uint32_t h = BloomHash(key); const uint32_t delta = (h >> 17) | (h << 15); // Rotate right 17 bits for (size_t j = 0; j < k; j++) { const uint32_t bitpos = h % bits; if ((array[bitpos/8] & (1 << (bitpos % 8))) == 0) return false; h += delta; } return true; }

三、布隆过滤器在leveldb中的应用

那么布隆过滤器在leveldb应用到什么方面呢？看过之前的博客，应该有印象的，在介绍存储结构中一个的block叫做filter block。这部分处理就是布隆过滤器应用之处，可参考《leveldb深度剖析-存储结构(2)》。下面具体看看是如何应用的。

3.1、开启过滤策略

leveldb接口默认不开启过滤策略（不是很清楚为啥？），在创建对象Options时候指定了为NULL（options.cc）。

Options::Options() : comparator(BytewiseComparator()), create_if_missing(false), error_if_exists(false), paranoid_checks(false), env(Env::Default()), info_log(NULL), write_buffer_size(4<<20), // 4M max_open_files(1000), block_cache(NULL), block_size(4096), block_restart_interval(16), //重启点间隔为16 max_file_size(2<<20), //2M compression(kSnappyCompression), reuse_logs(false), filter_policy(NULL) { }

 所以我们在创建Options对象需要单独设置一下filter_policy对象才可以。

3.2、添加key

std::string keys_; // Flattened key contents std::vector<size_t> start_; // Starting index in keys_ of each key /** * 将key添加到FilterBlock中 * @param key InternalKey 不包含value */ void FilterBlockBuilder::AddKey(const Slice& key) { Slice k = key; start_.push_back(keys_.size());//记录每个key的偏移位置 keys_.append(k.data(), k.size());//将key追加到字符串keys_末尾 }

3.3、Finish

Finish主要是格式化之前保存的key的，这里就会涉及布隆过滤器的。

Slice FilterBlockBuilder::Finish() { if (!start_.empty()) { GenerateFilter();//生成过滤器数据 } // Append array of per-filter offsets // 经过上面GenerateFilter操作后result_保存操作结果 const uint32_t array_offset = result_.size(); for (size_t i = 0; i < filter_offsets_.size(); i++) { PutFixed32(&result_, filter_offsets_[i]); } PutFixed32(&result_, array_offset); result_.push_back(kFilterBaseLg); // Save encoding parameter in result return Slice(result_);//返回后 写入ldb文件 }

 3.4、GenerateFilter

void FilterBlockBuilder::GenerateFilter() { const size_t num_keys = start_.size();//start_保存这这个block存储的key 多个 if (num_keys == 0) { // Fast path if there are no keys for this filter filter_offsets_.push_back(result_.size()); return; } // Make list of keys from flattened key structure start_.push_back(keys_.size()); // Simplify length computation tmp_keys_.resize(num_keys); for (size_t i = 0; i < num_keys; i++) { const char* base = keys_.data() + start_[i]; size_t length = start_[i+1] - start_[i]; tmp_keys_[i] = Slice(base, length); } // Generate filter for current set of keys and append to result_. // 调用布隆过滤器生成数据 filter_offsets_.push_back(result_.size()); policy_->CreateFilter(&tmp_keys_[0], static_cast<int>(num_keys), &result_); tmp_keys_.clear(); keys_.clear(); start_.clear(); }

3.5、查找匹配

查询的时候需要创建FilterBlockReader对象才可以进行查询，调用FilterBlockReader的KeyMayMatch方法即可，具体如下：

bool FilterBlockReader::KeyMayMatch(uint64_t block_offset, const Slice& key) { uint64_t index = block_offset >> base_lg_; if (index < num_) { uint32_t start = DecodeFixed32(offset_ + index*4); uint32_t limit = DecodeFixed32(offset_ + index*4 + 4); if (start <= limit && limit <= static_cast<size_t>(offset_ - data_)) { Slice filter = Slice(data_ + start, limit - start); return policy_->KeyMayMatch(key, filter);//InternalFilterPolicy::KeyMayMatch 布隆过滤器 } else if (start == limit) { // Empty filters do not match any keys return false; } } return true; // Errors are treated as potential matches }

这里就很明显是调用布隆过滤器进行查询匹配。那么上层业务是如何调到这里呢？

/** * 在table cache中查找k * @param options 选项 * @param k 待查找key * @param arg 回调函数参数 * @param saver 回调函数 */ Status Table::InternalGet(const ReadOptions& options, const Slice& k, void* arg, void (*saver)(void*, const Slice&, const Slice&)) { Status s; Iterator* iiter = rep_->index_block->NewIterator(rep_->options.comparator);//这里index_block类型是Block iiter->Seek(k);//查找 在索引块中查找 Block.cc if (iiter->Valid()) { Slice handle_value = iiter->value(); FilterBlockReader* filter = rep_->filter; BlockHandle handle; /** * 如果存在过滤模块 则先在过滤模块中通过布隆过滤器进行确定 如果布隆过滤器不存在 那么文件中一定不存在 * if 返回true表示不存在,if返回false表示可能存在 需要再次进行文件中查找即进入else中 */ if (filter != NULL && handle.DecodeFrom(&handle_value).ok() && !filter->KeyMayMatch(handle.offset(), k)) { // Not found } else {//在数据块中进行查找 iiter->value存储的是BlockHandle Iterator* block_iter = BlockReader(this, options, iiter->value()); block_iter->Seek(k); if (block_iter->Valid()) {//表示找到 (*saver)(arg, block_iter->key(), block_iter->value()); } s = block_iter->status(); delete block_iter; } } if (s.ok()) { s = iiter->status(); } delete iiter; return s; }

 四、总结

其实布隆过滤器思想比较简单，我们需要了解相关特性即可。