避免缓存击穿的利器之BloomFilter

缓存穿透问题

当用户想要查询一个数据，发现缓存中没有，出现缓存未命中，于是转向持久层数据库查询发现也没有，于是本次查询失败，且不会加入缓存。这就会导致，下次再用相同的条件查询时，由于缓存未命中依旧会查数据库，当并发流量很大时，会把DB打垮，这就是缓存穿透。

缓存穿透是指查询一个根本不存在的数据，缓存和存储层都没有命中，通常当存储层查不到数据时就不会写入缓存。

public Object getItemById(long id) {
	//从缓存中查
	Object item = cache.get(id);
	if (item != null) {
		return item;
	}
	item = itemDao.findById(id);
	//如果这里查询DB为空，则出现缓存穿透
	if (item != null) {
		cache.put(id, item);
	}
	return item;
}

缓存穿透将导致不存在的数据每次请求都要到存储层去查询，失去了缓存保护DB的意义，由于DB不具备高并发性甚至可能造成DB挂掉。

解决方案一：缓存空对象

顾名思义，就是DB查询结果为null时，将null也缓存起来，防止下次查询时穿透缓存。但这种方式存在漏洞，如果恶意制造许多不存在的id来查询，那么缓存中的键值就会很多，需要更多的内存空间，恶意攻击时会打爆缓存，比较有效的方法是针对null的缓存要设置较短的过期时间，让其自动剔除。

public Object getItemById(long id) {
	//从缓存中查
	Object item = cache.get(id);
	if (item != null) {
		return item;
	}
	item = itemDao.findById(id);
	//如缓存了空对象，就设置过期时间为5S
	if (item != null) {
		cache.put(id, item);
	} else {
		cache.put(id, item, 60 * 5);
	}
	return item;
}

解决方案二：布隆过滤器

布隆过滤器（英语：Bloom Filter）是1970年由一个叫布隆的小伙子提出的。它实际上是由一个很长的二进制向量和一系列随机映射函数构成。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询效率都远远超过一般的算法，缺点是有一定的误识别率和删除困难。Bloom Filter的高效是有一定代价的：在判断一个元素是否属于某个集合时，有可能会把不属于这个集合的元素误认为属于这个集合(false positive)。因此，Bloom Filter不适合那些“零错误”的应用场合。而在能容忍低错误率的应用场合下，Bloom Filter通过极少的错误换取了存储空间的极大节省。

Bloom Filter原理

Bloom Filter的原理是，当一个元素被加入集合时，通过K个散列函数将这个元素映射成一个位数组中的K个点，把它们置为1。检索时，我们只要看看这些点是不是都是1就（大约）知道集合中有没有它了：如果这些点有任何一个0，则被检元素一定不在；如果都是1，则被检元素很可能存在。这就是布隆过滤器的基本思想。

Bloom Filter跟单哈希函数Bit-Map不同之处在于：Bloom Filter使用了K个哈希函数，每个元素要跟K个bit对应，从而降低了冲突的概率。

Bloom Filter是由一个长度为m比特的位数组（bit array）与K个哈希函数（hash function）组成的数据结构。位数组均初始化为0，所有哈希函数都可以分别把输入数据尽量均匀地散列。

• 当要插入一个元素时，将其数据分别输入K个哈希函数，产生K个哈希值。以哈希值作为位数组中的下标，将所有K个对应的比特置为1。
• 当要判断一个元素是否存在时，同样将其数据输入哈希函数，然后检查对应的K个比特。如果有任意一个比特为0，表明该元素一定不在集合中。如果所有比特均为1，表明该集合有（较大的）可能性在集合中。

上图展示出一个m=28、K=3的Bloom Filter示例。集合中的a、b、c三个元素通过3个不同的哈希函数散列到位数组中。当查询元素e时，因为有一个比特为0，因此e一定不在该集合中。当查询元素d时，尽管所有比特都为1，但由于哈希冲突，可能判断d很可能在集合中，也可能不在。

那么Bloom Filter的诞生契机是什么呢？我们平常在检测集合中是否存在某元素时，都会采用比较的方法。考虑以下情况：

• 如果集合用线性表存储，查找的时间复杂度为O(N)。
• 如果用平衡二叉查找树（如AVL树、红黑树）存储，时间复杂度为O(logN)。
• 如果用哈希表存储，并用链地址法与平衡二叉查找树解决哈希冲突（参考JDK8中的HashMap实现方法），时间复杂度也要有O[log(N/M)]，M为哈希分桶数。

总而言之，当集合中元素的数量极多（百/千万级甚至更多）时，不仅查找会变得很慢，而且占用的空间也会大到无法想象。而布隆过滤器就是解决这个矛盾的利器。

Bloom Filter的优缺点

Bloom filter的优点是显而易见的：

• 不需要存储数据本身，只用比特表示，因此空间占用相对于传统方式有巨大的优势，并且能够保密数据；
• 时间效率也较高，插入和查询的时间复杂度均为O(k)；
• 哈希函数之间相互独立，可以在硬件指令层面并行计算。

它的缺点也同样明显：

• 存在误判，可能要查到的元素并没有在容器中，但是hash之后得到的K个位置上值都是1，不适用于任何要求100%准确率的场景。
• 只能插入和查询元素，不能删除元素，一个放入容器的元素映射到bit数组的K个位置上是1，删除的时候不能简单的直接置为0，可能会影响其他元素的判断。
• 哈希函数个数K越多，误判概率越低；位数组长度m越大，误判概率越低；已插入元素的个数n越大，误判概率越高。

Guava中的布隆过滤器实现

布隆过滤器有许多实现与优化，Guava中就提供了一种Bloom Filter的实现。在使用Bloom Filter时，绕不过的两点是预估数据量n以及期望的误判率fpp，在实现Bloom Filter时，绕不过的两点就是hash函数的选取以及bit数组的大小。

对于一个确定的场景，我们预估要存的数据量为n，期望的误判率为fpp，然后需要计算我们需要的Bit数组的大小m，以及hash函数的个数K，并选择hash函数。

要使用BloomFilter，需要引入guava包：

<dependency>
	<groupId>com.google.guava</groupId>
	<artifactId>guava</artifactId>
	<version>23.0</version>
</dependency>

Guava中，布隆过滤器的实现主要涉及到2个类，BloomFilter和BloomFilterStrategies，首先来看一下BloomFilter的成员变量。需要注意的是不同Guava版本的BloomFilter实现不同，以下是它的4个成员变量。

  /** guava实现的以CAS方式设置每个bit位的bit数组 */
  private final LockFreeBitArray bits;

  /** hash函数的个数 */
  private final int numHashFunctions;

  /** guava中将对象转换为byte的通道 */
  private final Funnel<? super T> funnel;

  /**
   * 将byte转换为n个bit的策略，也是bloomfilter hash映射的具体实现
   */
  private final Strategy strategy;

Guava Bloom Filter测试代码如下，往过滤器中放一百万个数，然后去验证这一百万个数是否能存在于过滤器中，再找另外1000个不存在于过滤器中的数字，验证误判率有多大。

public class TestBloomFilter {

    private static final int TOTAL = 1000000;
    private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), TOTAL);

    public static void main(String[] args) {
        for (int i=0; i < TOTAL;i++) {
            bloomFilter.put(i);
        }

        // 匹配已在过滤器中的值，是否有匹配不上的
        for (int i = 0; i < TOTAL;i++) {
            if (!bloomFilter.mightContain(i)) {
                System.out.println("误判，已存在于在过滤器中的值被判断不存在~~~");
            }
        }

        // 匹配不在过滤器中的10000个值，有多少匹配出来
        int count = 0;
        for (int i = TOTAL; i < TOTAL + 10000; i++) {
            if (bloomFilter.mightContain(i)) {
                count++;
            }
        }
        System.out.println("10000个不存在于过滤器中的数中，误判为存在的数量：" + count + "， 错误率：" + count / 10000.0);
    }
}

运行结果如下：

10000个不存在于过滤器中的数中，误判为存在的数量：320， 错误率：0.032

Process finished with exit code 0

运行结果表示，遍历这一百万个在过滤器中的数时，都被识别出来了。一万个不在过滤器中的数，误伤了320个，错误率是0.032。

看下BloomFilter的源码，BloomFilter一共四个create方法，不过最终都是走向第四个。

public static <T> BloomFilter<T> create(Funnel<? super T> funnel, int expectedInsertions) {
	return create(funnel, (long) expectedInsertions);
}

public static <T> BloomFilter<T> create(Funnel<? super T> funnel, long expectedInsertions) {
	return create(funnel, expectedInsertions, 0.03); // FYI, for 3%, we always get 5 hash functions
}

public static <T> BloomFilter<T> create(
  Funnel<? super T> funnel, long expectedInsertions, double fpp) {
	return create(funnel, expectedInsertions, fpp, BloomFilterStrategies.MURMUR128_MITZ_64);
}

@VisibleForTesting
static <T> BloomFilter<T> create(
  Funnel<? super T> funnel, long expectedInsertions, double fpp, Strategy strategy) {
  	
	checkNotNull(funnel);
	checkArgument(
	    expectedInsertions >= 0, "Expected insertions (%s) must be >= 0", expectedInsertions);
	checkArgument(fpp > 0.0, "False positive probability (%s) must be > 0.0", fpp);
	checkArgument(fpp < 1.0, "False positive probability (%s) must be < 1.0", fpp);
	checkNotNull(strategy);

	if (expectedInsertions == 0) {
		expectedInsertions = 1;
	}
	/*
	 * TODO(user): Put a warning in the javadoc about tiny fpp values, since the resulting size
	 * is proportional to -log(p), but there is not much of a point after all, e.g.
	 * optimalM(1000, 0.0000000000000001) = 76680 which is less than 10kb. Who cares!
	 */
	long numBits = optimalNumOfBits(expectedInsertions, fpp);
	int numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);
	try {
		return new BloomFilter<T>(new LockFreeBitArray(numBits), numHashFunctions, funnel, strategy);
	} catch (IllegalArgumentException e) {
		throw new IllegalArgumentException("Could not create BloomFilter of " + numBits + " bits", e);
	}
}

最终的create方法接受4个参数：funnel是插入数据的Funnel，expectedInsertions是期望插入的元素总个数n，fpp即期望误判率p，strategy即哈希策略。由上可知，位数组的长度m和哈希函数的个数K分别通过optimalNumOfBits()方法和optimalNumOfHashFunctions()方法来估计。

在BloomFilterStrategies枚举中定义了两种哈希策略，都基于著名的MurmurHash算法，默认使用MURMUR128_MITZ_64。

Guava为了提高效率，自己实现了LockFreeBitArray来提供bit数组的无锁设置和读取，LockFreeBitArray类的部分源码如下：

static final class LockFreeBitArray {
	private static final int LONG_ADDRESSABLE_BITS = 6;
	//采用原子类型AtomicLongArray作为位数组的存储
	final AtomicLongArray data;
	//LongAddable类型的计数器，用来统计置为1的比特数
	private final LongAddable bitCount;

	LockFreeBitArray(long bits) {
		this(new long[Ints.checkedCast(LongMath.divide(bits, 64, RoundingMode.CEILING))]);
	}

	// Used by serialization
	LockFreeBitArray(long[] data) {
		checkArgument(data.length > 0, "data length is zero!");
		this.data = new AtomicLongArray(data);
	  	this.bitCount = LongAddables.create();
	  	long bitCount = 0;
	  	for (long value : data) {
	    	bitCount += Long.bitCount(value);
	  	}
	  	this.bitCount.add(bitCount);
	}

	/** Returns true if the bit changed value. */
	boolean set(long bitIndex) {
		if (get(bitIndex)) {
	    	return false;
	  	}

	  	int longIndex = (int) (bitIndex >>> LONG_ADDRESSABLE_BITS);
	  	long mask = 1L << bitIndex; // only cares about low 6 bits of bitIndex

		long oldValue;
		long newValue;
		// 经典的CAS自旋重试
		do {
			oldValue = data.get(longIndex);
			newValue = oldValue | mask;
			if (oldValue == newValue) {
			  return false;
			}
		} while (!data.compareAndSet(longIndex, oldValue, newValue));
		
		bitCount.increment();
		return true;
	}
	/*
	 * 2的6次方等于64
	 * 将下标i无符号右移6位就可以获得data数组中对应的位置，再在其基础上左移i位就可以取得对应的比特了
	 */
	boolean get(long bitIndex) {
	  return (data.get((int) (bitIndex >>> 6)) & (1L << bitIndex)) != 0;
	}
}

采用原子类型AtomicLongArray作为位数组的存储的，确实不需要加锁。采用AtomicLongArray除了有并发上的优势之外，更主要的是它可以表示非常长的位数组。一个长整型数占用64bit，因此data[0]可以代表第0-63bit，data[1]代表64-127bit，data[2]代表128~191bit……依次类推。这样设计的话，将下标i无符号右移6位就可以获得data数组中对应的位置，再在其基础上左移i位就可以取得对应的比特了。

布隆过滤器防止缓存穿透

缓存null对象并设置较短的过期时间，让其自动失效，这种防止缓存穿透的方案不是最佳的实践方案。在了解了布隆过滤器之后，我们可以在商品持久化完成之后，将其商品id加入过滤器当中，每次进行查询时先确认要查询的id是否在过滤器中，如果不在，则说明id为非法id，就不需要进行后续的步骤了。

public Object getItemViaBloomFilter(long id) {
	if (!bloomFilter.mightContain(id)) {
		return null;
	}
	//从缓存中查
	Object item = cache.get(id);
	if (item != null) {
		return item;
	}
	item = itemDao.findById(id);
	//如缓存了空对象，就设置过期时间为5S
	if (item != null) {
		cache.put(id, item);
	} else {
		cache.put(id, item, 60 * 5);
	}
	return item;
}

布隆过滤器的其他适用场景

数据库防止穿库，Google Bigtable、Apache HBase和Apache Cassandra以及Postgresql使用BloomFilter来减少不存在的行或列的磁盘查找。避免代价高昂的磁盘查找会大大提高数据库查询操作的性能。

WEB拦截器，相同请求拦截防止被攻击。用户第一次请求，将请求参数放入BloomFilter中，当第二次请求时，先判断请求参数是否被BloomFilter命中。

恶意地址检测，chrome浏览器检查是否是恶意地址。首先针对本地BloomFilter检查任何URL，并且仅当BloomFilter返回肯定结果时才对所执行的URL进行全面检查。