当我们使用主流数据结构如 Lists, Maps, Sets, Trees等等时，我可以得到确切的结果，无论这个数据存在或是不存在。概率数据结构能够提供一种基于内存的，快速的查找出一种可能而非确切的结果。这种概率数据结构就是布隆过滤器（Bloom filter）。 布隆过滤器可以检查值是“可能在集合中”还是“绝对不在集合中”。
Bloom过滤器实质上由长度为m的位向量或位列表（仅包含0或1位值的列表）组成，最初所有值均设置为0，如下所示：
在哈希表中，我们使用单个哈希函数，因此仅获得单个索引作为输出。 但是对于Bloom过滤器，我们将使用多个哈希函数，这将为我们提供多个索引。
如上图，对于给定的输入“geeks”，我们的3个散列函数将提供3个不同的输出1、4和7。我们已对其进行了标记。
对于另一个输入“nerd”，哈希函数给我们3、4和5。您可能已经注意到，索引“ 4”已经被先前的“geeks”输入标记了。（你肯定又疑问，继续看）
我们已经用两个输入填充了位向量，现在我们可以检查它是否存在。我们该怎么做？ 我们将使用3个哈希函数对 要搜索的值 进行哈希处理，然后查看保留的结果是什么。 比如，搜索“cat”，我们的哈希函数这次给了我们1、3和7。如下图：
而且我们可以看到所有索引都已标记为1。这意味着我们可以说，“也许’cat’已经插入到我们的列表中了”。但事实并非如此。所以，出了什么问题？ 实际上，没有错。问题是，“误报”（false positive）就是这种情况。 布隆过滤器告诉我们，似乎之前已经插入了“cat”，因为应该用“cat”标记的索引已经被标记了（尽管是由于其他不同的数据）。 那么，是的，它有什么帮助？ 好吧，让我们考虑一下，如果“猫”给我们的输出是1，6，7而不是1，3，7。那会发生什么呢？我们可以看到，在3个索引中，有6个是“ 0”，这意味着它没有被任何先前的输入标记。这显然意味着以前从未插入过“cat”，如果是的话，则不可能有6成为“ 0”，对吧？这样，bloom过滤器可以“确定”地判断列表中是否没有数据。 简而言之： 如果我们搜索一个值，并且发现该值的任何哈希索引为“ 0”，那么该值肯定不在列表中。 如果所有散列索引均为“ 1”，则“也许”搜索到的值在列表中。
1）一个元素如果判断结果为存在的时候元素不一定存在，但是判断结果为不存在的时候则一定不存在。也就是说布隆过滤器只能判断数据是否一定不存在，而无法判断数据是否一定存在。 2) 布隆过滤器可以添加元素，但是不能删除元素。因为删掉元素会导致误判率增加。
相比于其它的数据结构，布隆过滤器在空间和时间方面都有巨大的优势。布隆过滤器存储空间和插入/查询时间都是常数 ，另外，散列函数相互之间没有关系，方便由硬件并行实现。布隆过滤器不需要存储元素本身，在某些对保密要求非常严格的场合有优势。 布隆过滤器可以表示全集，其它任何数据结构都不能； 由于存放的不是完整的数据，所以占用的内存很少，而且新增，查询速度够快。
但是布隆过滤器的缺点和优点一样明显。误算率是其中之一。随着存入的元素数量增加，误算率随之增加。随着数据的增加，误判率随之增加；无法做到删除数据；只能判断数据是否一定不存在，而无法判断数据是否一定存在。 但是如果元素数量太少，则使用散列表足矣。 另外，一般情况下不能从布隆过滤器中删除元素。我们很容易想到把位数组变成整数数组，每插入一个元素相应的计数器加 1, 这样删除元素时将计数器减掉就可以了。然而要保证安全地删除元素并非如此简单。首先我们必须保证删除的元素的确在布隆过滤器里面。这一点单凭这个过滤器是无法保证的。另外计数器回绕也会造成问题。
在程序的世界中，布隆过滤器是程序员的一把利器，利用它可以快速地解决项目中一些比较棘手的问题。 如网页 URL 去重、垃圾邮件识别、大集合中重复元素的判断和缓存穿透等问题。 布隆过滤器的典型应用有：
Google Bigtable，HBase 和 Cassandra 以及 Postgresql 使用BloomFilter来减少不存在的行或列的磁盘查找。避免代价高昂的磁盘查找会大大提高数据库查询操作的性能。
比如抖音或头条，当然会导致一定的误判，但不会让用户看到重复的内容。
一般判断用户是否在缓存中，如果在则直接返回结果，不在则查询db，如果来一波冷数据，会导致缓存大量击穿，造成雪崩效应，这时候可以用布隆过滤器当缓存的索引，只有在布隆过滤器中，才去查询缓存，如果没查询到，则穿透到db。如果不在布隆器中，则直接返回。
如果相同请求则拦截，防止重复被攻击。用户第一次请求，将请求参数放入布隆过滤器中，当第二次请求时，先判断请求参数是否被布隆过滤器命中。可以提高缓存命中率。Squid 网页代理缓存服务器在 cache digests 中就使用了布隆过滤器。Google Chrome浏览器使用了布隆过滤器加速安全浏览服务
缓存穿透：当请求数据库中不存在的数据，这时候所有的请求都会打到数据库上，这种情况就是缓存穿透。如果当请求较多的话，这将会严重浪费数据库资源甚至导致数据库假死。 我们经常会把一部分数据放在Redis等缓存，比如产品详情。这样有查询请求进来，我们可以根据产品Id直接去缓存中取数据，而不用读取数据库，这是提升性能最简单，最普遍，也是最有效的做法。一般的查询请求流程是这样的：先查缓存，有缓存的话直接返回，如果缓存中没有，再去数据库查询，然后再把数据库取出来的数据放入缓存，一切看起来很美好。但是如果现在有大量请求进来，而且都在请求一个不存在的产品Id，会发生什么？既然产品Id都不存在，那么肯定没有缓存，没有缓存，那么大量的请求都怼到数据库，数据库的压力一下子就上来了，还有可能把数据库打死。 所以缓存穿透就是当请求数据库中不存在的数据，这时候所有的请求都会打到数据库上，这种情况就是缓存穿透。如果当请求较多的话，这将会严重浪费数据库资源甚至导致数据库假死。 虽然有很多办法都可以解决这问题，但是 “布隆过滤器”就可以解决（缓解）缓存穿透问题。至于为什么说是“缓解”，看下去你就明白了。
这种技术在缓存之前再加一层屏障，里面存储目前数据库中存在的所有key。 当业务系统有查询请求的时候，首先去BloomFilter中查询该key是否存在。若不存在，则说明数据库中也不存在该数据，因此缓存都不要查了，直接返回null。若存在，则继续执行后续的流程，先前往缓存中查询，缓存中没有的话再前往数据库中的查询。
redis伪代码： String get(String key){ String value = redis.get(key); if (value == null) { if (!bloomFilter.mightContain(key)) { //不存在则返回 return null; } else { //可能存在则查数据库 value = db.get(key); redis.set(key, value); } } return value; }现在有大量的数据，而这些数据的大小已经远远超出了服务器的内存，现在再给你一个数据，如何判断给你的数据在不在其中。如果服务器的内存足够大，那么用HashMap是一个不错的解决方案，理论上的时间复杂度可以达到O(1)，但是现在数据的大小已经远远超出了服务器的内存，所以无法使用HashMap，这个时候就可以使用“布隆过滤器”来解决这个问题。但是还是同样的，会有一定的“误判率”。
比如123456这种被成为是弱密码。 可以在启动项目的时候，将弱密码库中的所有数据查出来放到Bloom filter中， 当校验是否为弱密码时，先去Bloom filter中校验，如果不存在，则表明为非弱密码，如果存在，由于会出现误报（false positive）情况，这个时候，可以去库里面查询是否真的存在然后返回结果。这样可以大大提供效率。 如下图表示如密码校验流程：
关于布隆过滤器，我们不需要自己实现，谷歌已经帮我们实现好了。
<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>27.0.1-jre</version> </dependency>guava 的布隆过滤器，封装的非常好，使用起来非常简洁方便。 例： 预估数据量1w，错误率需要减小到万分之一。使用如下代码进行创建。
public static void main(String[] args) { // 1.创建符合条件的布隆过滤器 // 预期数据量10000，错误率0.0001 BloomFilter<CharSequence> bloomFilter = BloomFilter.create(Funnels.stringFunnel( Charset.forName("utf-8")),10000, 0.0001); // 2.将一部分数据添加进去 for (int i = 0; i < 5000; i++) { bloomFilter.put("" + i); } System.out.println("数据写入完毕"); // 3.测试结果 for (int i = 0; i < 10000; i++) { if (bloomFilter.mightContain("" + i)) { System.out.println(i + "存在"); } else { System.out.println(i + "不存在"); } } }我们使用guava实现布隆过滤器是把数据放在本地内存中，无法实现布隆过滤器的共享，我们还可以把数据放在redis中，用 redis来实现布隆过滤器，我们要使用的数据结构是bitmap，你可能会有疑问，redis支持五种数据结构：String，List，Hash，Set，ZSet，没有bitmap呀。没错，实际上bitmap的本质还是String。 要用redis来实现布隆过滤器，我们需要自己设计映射函数，自己度量二进制向量的长度，这对我来说，无疑是一个不可能完成的任务，只能借助搜索引擎，下面直接放出代码把。
public class RedisMain { static final int expectedInsertions = 100;//要插入多少数据 static final double fpp = 0.01;//期望的误判率 //bit数组长度 private static long numBits; //hash函数数量 private static int numHashFunctions; static { numBits = optimalNumOfBits(expectedInsertions, fpp); numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits); } public static void main(String[] args) { Jedis jedis = new Jedis("192.168.0.109", 6379); for (int i = 0; i < 100; i++) { long[] indexs = getIndexs(String.valueOf(i)); for (long index : indexs) { jedis.setbit("codebear:bloom", index, true); } } for (int i = 0; i < 100; i++) { long[] indexs = getIndexs(String.valueOf(i)); for (long index : indexs) { Boolean isContain = jedis.getbit("codebear:bloom", index); if (!isContain) { System.out.println(i + "肯定没有重复"); } } System.out.println(i + "可能重复"); } } /** * 根据key获取bitmap下标 */ private static long[] getIndexs(String key) { long hash1 = hash(key); long hash2 = hash1 >>> 16; long[] result = new long[numHashFunctions]; for (int i = 0; i < numHashFunctions; i++) { long combinedHash = hash1 + i * hash2; if (combinedHash < 0) { combinedHash = ~combinedHash; } result[i] = combinedHash % numBits; } return result; } private static long hash(String key) { Charset charset = Charset.forName("UTF-8"); return Hashing.murmur3_128().hashObject(key, Funnels.stringFunnel(charset)).asLong(); } //计算hash函数个数 private static int optimalNumOfHashFunctions(long n, long m) { return Math.max(1, (int) Math.round((double) m / n * Math.log(2))); } //计算bit数组长度 private static long optimalNumOfBits(long n, double p) { if (p == 0) { p = Double.MIN_VALUE; } return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2))); } }运行结果：
88可能重复 89可能重复 90可能重复 91可能重复 92可能重复 93可能重复 94可能重复 95可能重复 96可能重复 97可能重复 98可能重复 99可能重复==不过Redis 官方提供的布隆过滤器到了 Redis 4.0 提供了插件功能。==布隆过滤器作为一个插件加载到 Redis Server 中，给 Redis 提供了强大的布隆去重功能。 在已安装 Redis 的前提下，安装 RedisBloom，有两种方式 直接编译进行安装
git clone https://github.com/RedisBloom/RedisBloom.git cd RedisBloom make #编译 会生成一个rebloom.so文件 redis-server --loadmodule /path/to/rebloom.so #运行redis时加载布隆过滤器模块 redis-cli # 启动连接容器中的 redis 客户端验证使用Docker进行安装
docker pull redislabs/rebloom:latest # 拉取镜像 docker run -p 6379:6379 --name redis-redisbloom redislabs/rebloom:latest #运行容器 docker exec -it redis-redisbloom bash redis-cli使用 布隆过滤器基本指令： • bf.add 添加元素到布隆过滤器 • bf.exists 判断元素是否在布隆过滤器 • bf.madd 添加多个元素到布隆过滤器，bf.add 只能添加一个 • bf.mexists 判断多个元素是否在布隆过滤器
127.0.0.1:6379> bf.add user Tom (integer) 1 127.0.0.1:6379> bf.add user John (integer) 1 127.0.0.1:6379> bf.exists user Tom (integer) 1 127.0.0.1:6379> bf.exists user Linda (integer) 0 127.0.0.1:6379> bf.madd user Barry Jerry Mars 1) (integer) 1 2) (integer) 1 3) (integer) 1 127.0.0.1:6379> bf.mexists user Barry Linda 1) (integer) 1 2) (integer) 0我们只有这几个参数，肯定不会有误判，当元素逐渐增多时，就会有一定的误判了。 上面使用的布隆过滤器只是默认参数的布隆过滤器，它在我们第一次 add 的时候自动创建。 Redis 还提供了自定义参数的布隆过滤器，bf.reserve 过滤器名 error_rate initial_size • error_rate：允许布隆过滤器的错误率，这个值越低过滤器的位数组的大小越大，占用空间也就越大 • initial_size：布隆过滤器可以储存的元素个数，当实际存储的元素个数超过这个值之后，过滤器的准确率会下降 但是这个操作需要在 add 之前显式创建。如果对应的 key 已经存在，bf.reserve 会报错
127.0.0.1:6379> bf.reserve user 0.01 100 (error) ERR item exists 127.0.0.1:6379> bf.reserve topic 0.01 1000 OKJava 的 Redis 客户端比较多，有些还没有提供指令扩展机制，已知的 Redisson 和 lettuce 是可以使用布隆过滤器的，我们这里用 Redisson
public class RedissonBloomFilterDemo { public static void main(String[] args) { Config config = new Config(); config.useSingleServer().setAddress("redis://127.0.0.1:6379"); RedissonClient redisson = Redisson.create(config); RBloomFilter<String> bloomFilter = redisson.getBloomFilter("user"); // 初始化布隆过滤器，预计统计元素数量为55000000，期望误差率为0.03 bloomFilter.tryInit(55000000L, 0.03); bloomFilter.add("Tom"); bloomFilter.add("Jack"); System.out.println(bloomFilter.count()); //2 System.out.println(bloomFilter.contains("Tom")); //true System.out.println(bloomFilter.contains("Linda")); //false } }假如现在收集了10亿个邮箱黑名单，当用户注册或者登录请求的时候，我们需要判断用户是否属于这10亿个黑名单邮箱中，如果存在则需要对该用户进行拦截。 这个需求咋一听很简单，无非就是yes or no的判断逻辑，难点在于数据量很大，而且巨大，如何把这个10亿个元素存储起来都是一个问题，还要在这10亿个元素之间去检索。通过BloomFilter的学习，我们就可以选择使用布隆过滤器来存储我们的10亿条数据，
import com.google.common.hash.BloomFilter; import com.google.common.hash.Funnel; import com.google.common.hash.PrimitiveSink; import org.apache.commons.codec.Charsets; public class BloomDemo { private static final BloomFilter<String> dealIdBloomFilter = BloomFilter.create(new Funnel<String>() { private static final long serialVersionUID = 1L; @Override public void funnel(String arg0, PrimitiveSink arg1) { arg1.putString(arg0, Charsets.UTF_8); } }, 1024*1024*32,0.0001); public static void main(String[] args) { dealIdBloomFilter.put("123"); dealIdBloomFilter.put("456"); if (dealIdBloomFilter.mightContain("123")){ System.out.println("yes"); } } }其中put方法是往布隆过滤器中添加元素，mightContain方法是检测元素是否在布隆过滤器中。
单机版本的布隆过滤器是没有任何问题的，也可以实现我们的需求，但是我们在生产环境都是集群部署的，也就意味着每台机器我们都需要存储10亿条记录，如果遇到系统发布，那还得重新初始化存储一份，为此我们通过guava和redis实现了分布式版本的布隆过滤器，代码如下：
import com.google.common.base.Preconditions; import com.google.common.hash.Funnel; import com.google.common.hash.Hashing; import org.springframework.data.redis.core.RedisTemplate; public class RedisBloom<T> { private int numHashFunctions; private int bitSize; private Funnel<T> funnel; private RedisTemplate redisTemplate; private RedisBloom(){ } /** * * @param funnel * @param expectedInsertions 预计元数总量大小 * @param fpp 允许的误差，例如 0.0001，允许万分之一的误差 * @param redisTemplate */ public RedisBloom(Funnel<T> funnel, int expectedInsertions, double fpp, RedisTemplate redisTemplate) { Preconditions.checkArgument(funnel != null, "funnel不能为空"); this.funnel = funnel; bitSize = optimalNumOfBits(expectedInsertions, fpp); numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, bitSize); this.redisTemplate = redisTemplate; } private int[] murmurHashOffset(T value) { int[] offset = new int[numHashFunctions]; long hash64 = Hashing.murmur3_128().hashObject(value, funnel).asLong(); int hash1 = (int) hash64; int hash2 = (int) (hash64 >>> 32); for (int i = 1; i <= numHashFunctions; i++) { int nextHash = hash1 + i * hash2; if (nextHash < 0) { nextHash = ~nextHash; } offset[i - 1] = nextHash % bitSize; } return offset; } /** * 计算bit数组长度 */ private int optimalNumOfBits(long n, double p) { if (p == 0) { p = Double.MIN_VALUE; } return (int) (-n * Math.log(p) / (Math.log(2) * Math.log(2))); } /** * 计算hash方法执行次数 */ private int optimalNumOfHashFunctions(long n, long m) { return Math.max(1, (int) Math.round((double) m / n * Math.log(2))); } /** * 根据给定的布隆过滤器添加值 */ public void addByBloomFilter(String key, T value) { int[] offset = this.murmurHashOffset(value); for (int i : offset) { redisTemplate.opsForValue().setBit(key, i, true); } } /** * 根据给定的布隆过滤器判断值是否存在 */ public boolean includeByBloomFilter(String key, T value) { int[] offset = murmurHashOffset(value); for (int i : offset) { if (!redisTemplate.opsForValue().getBit(key, i)) { return false; } } return true; } }RedisBloom是实现功能的关键，包含了计算bitmap的核心算法，其实大部分代码都是来源于Guava库里面的BloomFilterStrategies类，但是因为这个类是专门为Guava的BloomFilter类使用的，所以没有对外暴露一些重要的算法逻辑。
再来看怎么结合redis一起使用BloomFilter，addByBloomFilter，往redis里面添加元素includeByBloomFilter，检查元素是否在redis bloomFilter里面。
测试用例如下：
import com.google.common.hash.Funnel; import com.google.common.hash.PrimitiveSink; import org.apache.commons.codec.Charsets; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.test.context.junit4.SpringRunner; import javax.annotation.PostConstruct; @RunWith(SpringRunner.class) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) public class RedisBloomTest { @Autowired private RedisTemplate redisTemplate; private static RedisBloom<String> redisBloom; @PostConstruct public void init(){ redisBloom = new RedisBloom<>(new Funnel<String>() { @Override public void funnel(String s, PrimitiveSink primitiveSink) { primitiveSink.putString(s, Charsets.UTF_8); } }, 1024, 0.0001, redisTemplate); } @Test public void testBloom() { String key = "black:list"; redisBloom.addByBloomFilter(key, "1@qq.com"); redisBloom.addByBloomFilter(key, "2@qq.com"); boolean isExist = redisBloom.includeByBloomFilter(key, "1@qq.com"); boolean isExist2 = redisBloom.includeByBloomFilter(key, "2@qq.com"); boolean isExist3 = redisBloom.includeByBloomFilter(key, "3@qq.com"); System.out.println(isExist); System.out.println(isExist2); System.out.println(isExist3); } }运行结果：
true true false以上。
