本文共 9898 字，大约阅读时间需要 32 分钟。

判存业务

什么是

概念：

1.缓存雪崩

对于系统 A，假设每天高峰期每秒 5000 个请求，本来缓存在高峰期可以扛住每秒 4000 个请求，但是缓存机器意外发生了全盘宕机。缓存挂了，此时 1 秒 5000 个请求全部落数据库，数据库必然扛不住，它会报一下警，然后就挂了。此时，如果没有采用什么特别的方案来处理这个故障，DBA 很着急，重启数据库，但是数据库立马又被新的流量给打死了。

解释二：

缓存雪崩是指缓存中数据大批量到过期时间，而查询数据量巨大，引起数据库压力过大甚至down机。和缓存击穿不同的是，        缓存击穿指并发查同一条数据，缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库

缓存雪崩的事前事中事后的解决方案如下。

• 事前：redis 高可用，主从+哨兵，redis cluster，避免全盘崩溃。
• 事中：本地 ehcache 缓存 + hystrix 限流&降级，避免 MySQL 被打死。
• 事后：redis 持久化，一旦重启，自动从磁盘上加载数据，快速恢复缓存数据。

用户发送一个请求，系统 A 收到请求后，先查本地 ehcache 缓存，如果没查到再查 redis。如果 ehcache 和 redis 都没有，再查数据库，将数据库中的结果，写入 ehcache 和 redis 中。

限流组件，可以设置每秒的请求，有多少能通过组件，剩余的未通过的请求，怎么办？走降级！可以返回一些默认的值，或者友情提示，或者空白的值。

好处：

• 数据库绝对不会死，限流组件确保了每秒只有多少个请求能通过。
• 只要数据库不死，就是说，对用户来说，2/5 的请求都是可以被处理的。
• 只要有 2/5 的请求可以被处理，就意味着你的系统没死，对用户来说，可能就是点击几次刷不出来页面，但是多点几次，就可以刷出来一次。

2.缓存穿透

 缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求，如发起为id为“-1”的数据或id为特别大不存在的数据。这时的用户很可能是攻击者，攻击会导致数据库压力过大（一个用户），请求每次都“视缓存于无物”，直接查询数据库

  解决方案：

    a.接口层增加校验，如用户鉴权校验，id做基础校验，id<=0的直接拦截；

    b.从缓存取不到的数据，在数据库中也没有取到，这时也可以将key-value对写为key-null，缓存有效时间可以设置短点，如30秒（设置太长会导致正常情况也没法使用）。这样可以防止攻击用户反复用同一个id暴力攻击     c.每次系统 A 从数据库中只要没查到，就写一个空值到缓存里去，比如 set -999 UNKNOWN。然后设置一个过期时间，这样的话，下次有相同的 key 来访问的时候，在缓存失效之前，都可以直接从缓存中取数据。

3.缓存击穿

缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力（多个用户）

解决方案：

1. 设置热点数据永远不过期。
2. 基于 redis or zookeeper 实现互斥锁，等待第一个请求构建完缓存之后，再释放锁，进而其它请求才能通过该 key 访问数据   

互斥锁参考代码如下：

4.布隆过滤器

针对上述问题，我们引入布隆过滤器

当出现缓存穿透问题时，相当于redis不存在了，被击穿了，对于这种情况很好解决，我们可以在redis缓存一个空字符串或者特殊字符串，比如&&，下次我们去redis中查询的时候，当取到的值是空或者&&，我们就知道这个值在数据库中是没有的，就不会在去数据库中查询，ps：这里缓存不存在key的时候一定要设置过期时间，不然当数据库已经新增了这一条记录的时候，这样会导致缓存和数据库不一致的情况

上面这个是重复查询同一个不存在的值的情况，如果应用每次查询的不存在的值是不一样的呢？即使你每次都缓存特殊字符串也没用，因为它的值不一样，比如我们的数据库用户id是111，112，113，114依次递增，但是别人要攻击你，故意拿-100，-936，-545这种乱七八糟的key来查询，这时候redis和数据库这种值都是不存在的，人家每次拿的key也不一样，你就算缓存了也没用，这时候数据库的压力是相当大，比上面这种情况可怕的多，怎么办呢，这时候我们今天的主角布隆过滤器就登场了（黑客攻击）

从一道面试题说起

问：如何在海量元素中（例如 10 亿无序、不定长、不重复）快速判断一个元素是否存在？

好，我们最简单的想法就是把这么多数据放到数据结构里去，比如List、Map、Tree，一搜不就出来了吗，比如map.get(),我们假设一个元素1个字节的字段，10亿的数据大概需要 900G 的内存空间，这个对于普通的服务器来说是承受不了的，当然面试官也不希望听到你这个答案，因为太笨了吧，我们肯定是要用一种好的方法，巧妙的方法来解决，这里引入一种节省空间的数据结构，位图，他是一个有序的数组，只有两个值，0 和 1。0代表不存在，1代表存在。

如果我们要把一个key映射到布隆过滤器中，就需要利用所有哈希函数对这个key进行映射，得到一系列的哈希值，把这些哈希值当成数据下标，把下标对应的数组元素置为1。

有了这个屌炸天的东西，现在我们还需要一个映射关系，你总得知道某个元素在哪个位置上吧，然后在去看这个位置上是0还是1，怎么解决这个问题呢，那就要用到哈希函数，用哈希函数有两个好处，第一是哈希函数无论输入值的长度是多少，得到的输出值长度是固定的，第二是他的分布是均匀的，如果全挤的一块去那还怎么区分，比如MD5、SHA-1这些就是常见的哈希算法。

我们通过哈希函数计算以后就可以到相应的位置去找是否存在了，我们看红色的线，24和147经过哈希函数得到的哈希值是一样的，我们把这种情况叫做哈希冲突或者哈希碰撞。哈希碰撞是不可避免的，我们能做的就是降低哈希碰撞的概率，第一种是可以扩大维数组的长度或者说位图容量，因为我们的函数是分布均匀的，所以位图容量越大，在同一个位置发生哈希碰撞的概率就越小。但是越大的位图容量，意味着越多的内存消耗，所以我们想想能不能通过其他的方式来解决，第二种方式就是经过多几个哈希函数的计算，你想啊，24和147现在经过一次计算就碰撞了，那我经过5次，10次，100次计算还能碰撞的话那真的是缘分了，你们可以在一起了，但也不是越多次哈希函数计算越好，因为这样很快就会填满位图，而且计算也是需要消耗时间，所以我们需要在时间和空间上寻求一个平衡。。

布隆过滤器

当然，这个事情早就有人研究过了，在 1970 年的时候，有一个叫做布隆的前辈对于判断海量元素中元素是否存在的问题进行了研究，也就是到底需要多大的位图容量和多少个哈希函数，它发表了一篇论文，提出的这个容器就叫做布隆过滤器。

大家来看下这个图，我们看集合里面3个元素，现在我们要存了，比如说a，经过f1(a)，f2(a)，f3(a)经过三个哈希函数的计算，在相应的位置上存入1，元素b，c也是通过这三个函数计算放入相应的位置。当取的时候，元素a通过f1(a)函数计算，发现这个位置上是1，没问题，第二个位置也是1，第三个位置上也是 1，这时候我们说这个a在布隆过滤器中是存在的，没毛病，同理我们看下面的这个d，通过三次计算发现得到的结果也都是1，那么我们能说d在布隆过滤器中是存在的吗，显然是不行的，我们仔细看d得到的三个1其实是f1(a)，f1(b)，f2(c)存进去的，并不是d自己存进去的，这个还是哈希碰撞导致的，我们把这种本来不存在布隆过滤器中的元素误判为存在的情况叫做假阳性（False Positive Probability，FPP）。

我们再来看另一个元素，e 元素。我们要判断它在容器里面是否存在，一样地要用这三个函数去计算。第一个位置是 1，第二个位置是 1，第三个位置是 0。那么e元素能不能判断是否在布隆过滤器中？答案是肯定的，e一定不存在。你想啊，如果e存在的话，他存进去的时候这三个位置都置为1，现在查出来有一个位置是0，证明他没存进去啊。。通过上面这张图加说明，我们得出两个重要的结论

从容器的角度来说：

• 如果布隆过滤器判断元素在集合中存在，不一定存在

• 如果布隆过滤器判断不存在，一定不存在

从元素的角度来说：

• 如果元素实际存在，布隆过滤器一定判断存在

• 如果元素实际不存在，布隆过滤器可能判断存在

布隆过滤器的缺点

因为哈希函数可能是存在冲突的。在发生冲突的情况下，如果刚好所有的值都被置为1了，就会产生误判。为了能做到在时间和空间上的高效率，布隆过滤器牺牲了判断的准确率。所以对于不能接受误判的业务场景，通常需要在通过布隆过滤器之后再做一层判断。

另外布隆过滤器还有一个缺点就是删除比较困难，假如要删除一个key的话，不能简单的把这个key映射到的bit位都简单的置为0，因为哈希冲突的情况下，会影响其它key的判断。如果要删除的话，可以参考Counting Bloom Filter。

 

实战：使用Guava提供的布隆过滤器

首先需要引入依赖，在pom里面加上：

      
    
     com.google.guava
       
    
     guava
       
    
     29.0-jre
    
   

 测试误判率

import com.google.common.hash.BloomFilter;import com.google.common.hash.Funnels;​public class BloomFilterTest {   public static void main(String[] args) {       int total = 100000; //总数量​       BloomFilter
   
     bf = BloomFilter.create(Funnels.integerFunnel(), total);       // 初始化数据到布隆过滤器中       for (int i = 0; i < total; i++) {           bf.put(i);       }       // 判断值是否存在过滤器中       int count = 0;       for (int i = total; i < total + 10000; i++) {           if (bf.mightContain(i)) {               count++;           }       }       System.out.println("误判的数量 ～ " + count);   }}
   

布隆过滤器还是判断了286个存在，这个就是误判，原因上面也说过了，所以布隆过滤器不是万能的，但是他能帮我们抵挡掉大部分不存在的数据已经很不错了，已经减轻数据库很多压力了，另外误判率0.02是在初始化布隆过滤器的时候我们自己设的，如果不设默认是0.03，我们自己设的时候千万不能设0!

Redis实现布隆过滤器

上面使用guava实现布隆过滤器是把数据放在本地内存中，我们项目往往是分布式的，我们还可以把数据放在redis中，用redis来实现布隆过滤器，这就需要我们自己设计映射函数，自己度量二进制向量的长度

/** * 布隆过滤器核心类 * * @param 
   
     * @author jack xu */public class BloomFilterHelper
    
      {    private int numHashFunctions;    private int bitSize;    private Funnel
     
       funnel;    public BloomFilterHelper(int expectedInsertions) {        this.funnel = (Funnel
      
       ) Funnels.stringFunnel(Charset.defaultCharset());        bitSize = optimalNumOfBits(expectedInsertions, 0.03);        numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, bitSize);    }    public BloomFilterHelper(Funnel
       
         funnel, int expectedInsertions, double fpp) {        this.funnel = funnel;        bitSize = optimalNumOfBits(expectedInsertions, fpp);        numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, bitSize);    }    public 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)));    }}
       
      
     
    
   

/** * redis操作布隆过滤器 * * @param 
   
     * @author xhj */public class RedisBloomFilter
    
      {    @Autowired    private RedisTemplate redisTemplate;    /**     * 删除缓存的KEY     *     * @param key KEY     */    public void delete(String key) {        redisTemplate.delete(key);    }    /**     * 根据给定的布隆过滤器添加值，在添加一个元素的时候使用，批量添加的性能差     *     * @param bloomFilterHelper 布隆过滤器对象     * @param key               KEY     * @param value             值     * @param 
     
                     泛型，可以传入任何类型的value     */    public 
      
        void add(BloomFilterHelper
       
         bloomFilterHelper, String key, T value) {        int[] offset = bloomFilterHelper.murmurHashOffset(value);        for (int i : offset) {            redisTemplate.opsForValue().setBit(key, i, true);        }    }    /**     * 根据给定的布隆过滤器添加值，在添加一批元素的时候使用，批量添加的性能好，使用pipeline方式(如果是集群下，请使用优化后RedisPipeline的操作)     *     * @param bloomFilterHelper 布隆过滤器对象     * @param key               KEY     * @param valueList         值，列表     * @param 
        
          泛型，可以传入任何类型的value */ public 
         
           void addList(BloomFilterHelper
          
            bloomFilterHelper, String key, List
           
             valueList) { redisTemplate.executePipelined(new RedisCallback
            
             () { @Override public Long doInRedis(RedisConnection connection) throws DataAccessException { connection.openPipeline(); for (T value : valueList) { int[] offset = bloomFilterHelper.murmurHashOffset(value); for (int i : offset) { connection.setBit(key.getBytes(), i, true); } } return null; } }); } /** * 根据给定的布隆过滤器判断值是否存在 * * @param bloomFilterHelper 布隆过滤器对象 * @param key KEY * @param value 值 * @param 
             
               泛型，可以传入任何类型的value * @return 是否存在 */ public 
              
                boolean contains(BloomFilterHelper
               
                 bloomFilterHelper, String key, T value) { int[] offset = bloomFilterHelper.murmurHashOffset(value); for (int i : offset) { if (!redisTemplate.opsForValue().getBit(key, i)) { return false; } } return true; }}
               
              
             
            
           
          
         
        
       
      
     
    
   

测试类

public static void main(String[] args) {        RedisBloomFilter redisBloomFilter = new RedisBloomFilter();        int expectedInsertions = 1000;        double fpp = 0.1;        redisBloomFilter.delete('bloom');        BloomFilterHelper
   
     bloomFilterHelper = new BloomFilterHelper<>(Funnels.stringFunnel(Charset.defaultCharset()), expectedInsertions, fpp);        int j = 0;        // 添加100个元素        List
    
      valueList = new ArrayList<>();        for (int i = 0; i < 100; i++) {            valueList.add(i + '');        }        long beginTime = System.currentTimeMillis();        redisBloomFilter.addList(bloomFilterHelper, 'bloom', valueList);        long costMs = System.currentTimeMillis() - beginTime;        log.info('布隆过滤器添加{}个值，耗时：{}ms', 100, costMs);        for (int i = 0; i < 1000; i++) {            boolean result = redisBloomFilter.contains(bloomFilterHelper, 'bloom', i + '');            if (!result) {                j++;            }        }        log.info('漏掉了{}个,验证结果耗时：{}ms', j, System.currentTimeMillis() - beginTime);    }
    
   

注意这里用的是addList，他的底层是pipelining管道，而add方法的底层是一个个for循环的setBit，这样的速度效率是很慢的，但是他能有返回值，知道是否插入成功，而pipelining是不知道的，所以具体选择用哪一种方法看你的业务场景，以及需要插入的速度决定。

布隆过滤器工作位置

第一步是将数据库所有的数据加载到布隆过滤器。第二步当有请求来的时候先去布隆过滤器查询，如果bf说没有，第三步直接返回。如果bf说有，在往下走之前的流程。ps：另外guava的数据加载中只有put方法

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

• 网页爬虫对URL去重，避免爬取相同的 URL 地址；

• 反垃圾邮件，从数十亿个垃圾邮件列表中判断某邮箱是否垃圾邮箱；

• Google Chrome 使用布隆过滤器识别恶意 URL；

• Medium 使用布隆过滤器避免推荐给用户已经读过的文章；

• Google BigTable，Apache HBbase 和 Apache Cassandra使用布隆过滤器减少对不存在的行和列的查找。

在工作中也可以应用，比如鉴权服务，当用户登录的时候可以先用布隆过滤器判断下，而不是直接去redis、数据库查（防止黑客攻击）

          缓存穿透、缓存击穿、缓存雪崩区别和解决方案

        缓存穿透、缓存击穿、缓存雪崩区别和解决方案

     布隆过滤器

     漫画：什么是布隆过滤器

   布隆过滤器

转载地址：http://lpulf.baihongyu.com/