如何实现布隆过滤器？

1.布隆过滤器的场景

在Redis 缓存击穿（失效）、缓存穿透、缓存雪崩怎么解决？中我们说到可以使用布隆过滤器避免「缓存穿透」。
你会说我们只要记录了每个用户看过的历史记录，每次推荐的时候去查询数据库过滤存在的数据实现去重。
实际上，如果历史记录存储在关系数据库里，去重就需要频繁地对数据库进行 exists 查询，当系统并发量很高时，数据库是很难扛住压力的。
我们不能使用缓存 将历史数据存在Redis中，这么多的历史记录那要浪费多大的内存空间，所以这个时候我们就能使用布隆过滤器去解决这种去重问题。又快又省内存，互联网开发必备杀招！
当我们遇到数据量比较大，又需要去重的时候就可以考虑布隆过滤器，如下场景：

• 解决 Redis 缓存穿透问题（面试重点）；
  • 利用布隆过滤器我们可以预先把数据查询的主键，比如用户 ID 或文章 ID 缓存到过滤器中。当根据 ID 进行数据查询的时候，我们先判断该 ID 是否存在，若存在的话，则进行下一步处理。若不存在的话，直接返回，这样就不会触发后续的数据库查询。需要注意的是缓存穿透不能完全解决，我们只能将其控制在一个可以容忍的范围内。
• 邮件过滤，使用布隆过滤器实现邮件黑名单过滤；
• 反垃圾邮件，从数十亿个垃圾邮件列表中判断某邮箱是否垃圾邮箱；
• 网页爬虫对 URL 去重，避免爬取相同的 URL 地址；
• Google Chrome 使用布隆过滤器识别恶意 URL；
• Medium 使用布隆过滤器避免推荐给用户已经读过的文章；
• 推荐过的新闻不再推荐；
• Google BigTable，Apache HBbase 和 Apache Cassandra 使用布隆过滤器减少对不存在的行和列的查找。

2.布隆过滤器的概述

布隆过滤器 (Bloom Filter)是由 Burton Howard Bloom 于 1970 年提出，它是一种 space efficient 的概率型数据结构，用于判断一个元素是否在集合中。
当布隆过滤器说，某个数据存在时，这个数据可能不存在；当布隆过滤器说，某个数据不存在时，那么这个数据一定不存在。
哈希表也能用于判断元素是否在集合中，但是布隆过滤器只需要哈希表的 1/8 或 1/4 的空间复杂度就能完成同样的问题。
布隆过滤器可以插入元素，但不可以删除已有元素。
其中的元素越多，false positive rate(误报率)越大，但是 false negative (漏报)是不可能的。

3.布隆过滤器原理

BloomFilter 的算法是，首先分配一块内存空间做 bit 数组，数组的 bit 位初始值全部设为 0。加入元素时，采用 k 个相互独立的 Hash 函数计算，然后将元素 Hash 映射的 K 个位置全部设置为 1。检测 key 是否存在，仍然用这 k 个 Hash 函数计算出 k 个位置，如果位置全部为 1，则表明 key 存在，否则不存在。
如下图所示：

布隆过滤器原理
哈希函数会出现碰撞，所以布隆过滤器会存在误判。
这里的误判率是指，BloomFilter 判断某个 key 存在，但它实际不存在的概率，因为它存的是 key 的 Hash 值，而非 key 的值。
所以有概率存在这样的 key，它们内容不同，但多次 Hash 后的 Hash 值都相同。
对于 BloomFilter 判断不存在的 key ，则是 100% 不存在的，反证法，如果这个 key 存在，那它每次 Hash 后对应的 Hash 值位置肯定是 1，而不会是 0。布隆过滤器判断存在不一定真的存在。
布隆过滤器（Bloom Filter）本质上是由长度为 m 的位向量或位列表（仅包含 0 或 1 位值的列表）组成，最初所有的值均设置为 0，如下图所示。

为了将数据项添加到布隆过滤器中，我们会提供 K 个不同的哈希函数，并将结果位置上对应位的值置为 “1”。在前面所提到的哈希表中，我们使用的是单个哈希函数，因此只能输出单个索引值。而对于布隆过滤器来说，我们将使用多个哈希函数，这将会产生多个索引值。

如上图所示，当输入 “semlinker” 时，预设的 3 个哈希函数将输出 2、4、6，我们把相应位置 1。假设另一个输入 ”kakuqo“，哈希函数输出 3、4 和 7。你可能已经注意到，索引位 4 已经被先前的 “semlinker” 标记了。此时，我们已经使用 “semlinker” 和 ”kakuqo“ 两个输入值，填充了位向量。当前位向量的标记状态为：

当对值进行搜索时，与哈希表类似，我们将使用 3 个哈希函数对 ”搜索的值“ 进行哈希运算，并查看其生成的索引值。假设，当我们搜索 ”fullstack“ 时，3 个哈希函数输出的 3 个索引值分别是 2、3 和 7：

从上图可以看出，相应的索引位都被置为 1，这意味着我们可以说 ”fullstack“ 可能已经插入到集合中。事实上这是误报的情形，产生的原因是由于哈希碰撞导致的巧合而将不同的元素存储在相同的比特位上。幸运的是，布隆过滤器有一个可预测的误判率（FPP）：

• n 是已经添加元素的数量（总的元素数量）
• k 哈希的次数
• m 布隆过滤器的长度（如比特数组的大小）

极端情况下，当布隆过滤器没有空闲空间时（满），每一次查询都会返回 true 。这也就意味着 m 的选择取决于期望预计添加元素的数量 n ，并且 m 需要远远大于 n 。
实际情况中，布隆过滤器的长度 m 可以根据给定的误判率（FPP）的和期望添加的元素个数 n 的通过如下公式计算：

FPP越小说明了m的值越大 来实现m远大于n实现比特数组的长度更长。
了解完上述的内容之后，我们可以得出一个结论，当我们搜索一个值的时候，若该值经过 K 个哈希函数运算后的任何一个索引位为 ”0“，那么该值肯定不在集合中。但如果所有哈希索引值均为 ”1“，则只能说该搜索的值可能存在集合中。

为什么不允许删除元素？

删除意味着需要将对应的 k 个 bits 位置设置为 0，其中有可能是其他元素对应的位。
因此 remove 会引入 false negative，这是绝对不被允许的。

4.布隆过滤器优缺点

4.1.布隆过滤器优点

1. 增加和查询元素的时间复杂度为：O(K), (K为哈希函数的个数，一般比较小)，与数据量大小无关
2. 哈希函数相互之间没有关系，方便硬件并行运算
3. 布隆过滤器不需要存储元素本身，在某些对保密要求比较严格的场合有很大优势
4. 在能够承受一定的误判时，布隆过滤器比其他数据结构有这很大的空间优势
5. 数据量很大时，布隆过滤器可以表示全集，其他数据结构不能
6. 使用同一组散列函数的布隆过滤器可以进行交、并、差运算
7. 空间效率和查询时间都比一般的算法要好的多

4.2.布隆过滤器缺点

1. 有误判率，即存在假阳性(False Position)，即不能准确判断元素是否在集合中(补救方法：再建立一个白名单，存储可能会误判的数据)
2. 不能获取元素本身
3. 一般情况下不能从布隆过滤器中删除元素
4. 如果采用计数方式删除，可能会存在计数回绕问题
5. 存在一定的误识别率和删除困难

5.布隆过滤器的应用

布隆过滤器有很多实现和优化，由 Google 开发著名的 Guava 库就提供了布隆过滤器（Bloom Filter）的实现。在基于 Maven 的 Java 项目中要使用 Guava 提供的布隆过滤器，只需要引入以下环境：

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>30.0-jre</version>
</dependency>

在导入 Guava 库后，我们新建一个 BloomFilterDemo 类，在 main 方法中我们通过 BloomFilter.create 方法来创建一个布隆过滤器，接着我们初始化 1 百万条数据到过滤器中，然后在原有的基础上增加 10000 条数据并判断这些数据是否存在布隆过滤器中：

package cn.wen;import com.google.common.base.Charsets;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;/*** 布隆过滤器测试*/
public class BloomFilterDemo {public static void main(String[] args) {int total = 1000000; // 总数量BloomFilter<CharSequence> bf =BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), total,0.0002);// 初始化 1000000 条数据到过滤器中for (int i = 0; i < total; i++) {bf.put("" + i);}// 判断值是否存在过滤器中int count = 0;for (int i = 0; i < total + 10000; i++) {if (bf.mightContain("" + i)) {count++;}}System.out.println("已匹配数量 " + count);}}

当以上代码运行后，控制台会输出以下结果：

已匹配数量 1000309

很明显以上的输出结果已经出现了误报，因为相比预期的结果多了 309 个元素，误判率为：

309/(1000000 + 10000) * 100 ≈ 0.030594059405940593

如果要提高匹配精度的话，我们可以在创建布隆过滤器的时候设置误判率 fpp：

BloomFilter<CharSequence> bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), total, 0.0002);

在 BloomFilter 内部，误判率 fpp 的默认值是 0.03：

// com/google/common/hash/BloomFilter.class
public static <T> BloomFilter<T> create(Funnel<? super T> funnel, long expectedInsertions) {return create(funnel, expectedInsertions, 0.03D);
}

在重新设置误判率为 0.0002 之后，我们重新运行程序，这时控制台会输出以下结果：

已匹配数量 1000003

通过观察以上的结果，可知误判率 fpp 的值越小，匹配的精度越高。当减少误判率 fpp 的值，需要的存储空间也越大，所以在实际使用过程中需要在误判率和存储空间之间做个权衡。
方案1：开发定时任务，每隔几个小时，自动创建一个新的布隆过滤器数组，替换老的，有点CopyOnWriteArrayList的味道
方案2：布隆过滤器增加一个等长的数组，存储计数器，主要解决冲突问题，每次删除时对应的计数器减一，如果结果为0，更新主数组的二进制值为0

6、手写布隆过滤器

/**
* @ClassName: BloomFilter
* @Author: 小飞
* @Date: 2023/5/17 21:50
* @Description: 布隆过滤器
*/
public class BloomFilter {// 默认大小private static final int DEFAULT_SIZE = Integer.MAX_VALUE;// 最小的大小private static final int MIN_SIZE = 1000;// 大小为默认大小private int SIZE = DEFAULT_SIZE;// hash函数的种子因子private static final int[] HASH_SEEDS = new int[]{3, 5, 7, 11, 13, 17, 19, 23, 29, 31};// 位数组，0/1,表示特征private BitSet bitSet = null;// hash函数private HashFunction[] hashFunctions = new HashFunction[HASH_SEEDS.length];// 无参数初始化public BloomFilter() {// 按照默认大小init();}// 带参数初始化public BloomFilter(int size) {// 大小初始化小于最小的大小if (size >= MIN_SIZE) {SIZE = size;}init();}// 初始化private void init() {// 初始化位大小bitSet = new BitSet(SIZE);// 初始化hash函数for (int i = 0; i < HASH_SEEDS.length; i++) {hashFunctions[i] = new HashFunction(SIZE, HASH_SEEDS[i]);}}// 添加元素，相当于把元素的特征添加到位数组中public void add(Object value) {for (HashFunction f : hashFunctions) {// 将hash值计算出来的值为truebitSet.set(f.hash(value), true);}}// 判断元素的特征是否存在于位数组public boolean contains(Object value) {boolean result = true;for (HashFunction f : hashFunctions) {result = result && bitSet.get(f.hash(value));// hash函数只要有一个计算出为false，则直接返回if (!result) {return false;}}return true;}// hash函数public static class HashFunction {// 位数组大小private int size;// hash种子private int seed;public HashFunction(int size, int seed) {this.size = size;this.seed = seed;}// hash函数public int hash(Object value) {if (value == null) {return 0;} else {// hash值int hash1 = value.hashCode();// 高位的hash值int hash2 = hash1 >>> 16;// 合并hash值 结合高低位特征int combineHash = hash1 ^ hash2;// 相乘取余return Math.abs(combineHash * seed) % size;}}}public static void main(String[] args) {Integer num1 = new Integer(12321);Integer num2 = new Integer(12345);BloomFilter myBloomFilter =new BloomFilter();System.out.println(myBloomFilter.contains(num1));System.out.println(myBloomFilter.contains(num2));myBloomFilter.add(num1);myBloomFilter.add(num2);System.out.println(myBloomFilter.contains(num1));System.out.println(myBloomFilter.contains(num2));}
}

上面未提供Hash函数的数量，默认值实现hash函数。

// 带参数初始化
public BloomFilter(int num,double rate) {// 计算位数组的大小this.SIZE = (int) (-num * Math.log(rate) / Math.pow(Math.log(2), 2));// hash 函数个数this.s = (int) (this.SIZE * Math.log(2) / num);// 初始化位数组this.bitSet = new BitSet(SIZE);
}

上面的构造函数就是对误差率的初始化值。