【Redis】高阶使用（三）：缓存穿透，缓存击穿，缓存雪崩及解决方案

1.缓存穿透

缓存穿透是指查询一个根本不存在的数据，缓存层和存储层都不会命中，通常出于容错的考虑，如果从存储层查不到数据则不写入缓存层。

缓存穿透将导致不存在的数据每次请求都要到存储层去查询，失去了缓存保护后端存储的意义。造成缓存穿透的基本原因有两个：

自身业务代码或者数据出现问题。
一些恶意攻击、爬虫等造成大量空命中。

缓存穿透问题解决方案：

方案一：缓存空对象

String get(String key) { 
    // 从缓存中获取数据 
    String cacheValue = cache.get(key); 
    // 缓存为空 
    if (StringUtils.isBlank(cacheValue)) { 
        // 从存储中获取 
        String storageValue = storage.get(key); 
        cache.set(key, storageValue); 
        // 如果存储数据为空， 需要设置一个过期时间(300秒)
        if (storageValue == null) { 
            cache.expire(key, 60 * 5); 1
        } 
        return storageValue; 
    } else { 
        // 缓存非空 
        return cacheValue; 
    } 
}

方案二：布隆过滤器

对于恶意攻击，向服务器请求大量不存在的数据造成的缓存穿透，还可以用布隆过滤器先做一次过滤，对于不存在的数据布隆过滤器一般都能够过滤掉，不让请求再往后端发送。当布隆过滤器说某个值存在时，这个值可能不存在；当它说不存在时，那就肯定不存在。

布隆过滤器就是一个大型的位数组和几个不一样的无偏 hash 函数。所谓无偏就是能够把元素的 hash 值算得比较均匀。

向布隆过滤器中添加 key 时，会使用多个 hash 函数对 key 进行 hash 算得一个整数索引值然后对位数组长度进行取模运算得到一个位置，每个 hash 函数都会算得一个不同的位置。再把位数组的这几个位置都置为 1 就完成了 add 操作。

向布隆过滤器询问 key 是否存在时，跟 add 一样，也会把 hash 的几个位置都算出来，看看位数组中这几个位置是否都为 1，只要有一个位为 0，那么说明布隆过滤器中这个key 不存在。

如果都是 1，这并不能说明这个 key 就一定存在，只是极有可能存在，因为这些位被置为 1 可能是因为其它的 key 存在所致。如果这个位数组比较稀疏，这个概率就会很大，如果这个位数组比较拥挤，这个概率就会降低。

这种方法适用于数据命中不高、数据相对固定、实时性低（通常是数据集较大）的应用场景，代码维护较为复杂，但是缓存空间占用很少。

可以用guava包自带的布隆过滤器，引入依赖：

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

示例伪代码：

import com.google.common.hash.BloomFilter;

// 初始化布隆过滤器
// 1000：期望存入的数据个数，0.001：期望的误差率 
BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf‐8")), 1000, 0.001);

//把所有数据存入布隆过滤器 
void init(){ 
    for (String key: keys) { 
        bloomFilter.put(key); 
    } 
}

String get(String key) { 1
    // 从布隆过滤器这一级缓存判断下key是否存在 
    Boolean exist = bloomFilter.mightContain(key); 
    if(!exist){ 
        return ""; 
    } 
    // 从缓存中获取数据 
    String cacheValue = cache.get(key); 
    // 缓存为空 
    if (StringUtils.isBlank(cacheValue)) { 
        // 从存储中获取 
        String storageValue = storage.get(key); 
        cache.set(key, storageValue); 
        // 如果存储数据为空， 需要设置一个过期时间(300秒) 
        if (storageValue == null) { 
            cache.expire(key, 60 * 5); 
        } 
        return storageValue; 
    } else { 
        // 缓存非空 
        return cacheValue;  
    } 
}

更多关于布隆过滤器的可以看这篇：【数据结构】布隆过滤器：BloomFilter原理及Java实现…

2.缓存击穿

开发人员使用“缓存+过期时间”的策略既可以加速数据读写，又保证数据的定期更新，这种模式基本能够满足绝大部分需求。但是有两个问题如果同时出现，可能就会对应用造成致命的危害：

当前key是一个热点key（例如一个热门的娱乐新闻），并发量非常大。
重建缓存不能在短时间完成，可能是一个复杂计算，例如复杂的SQL、多次IO、多个依赖等。

在缓存失效的瞬间，有大量线程来重建缓存，造成后端负载加大，甚至可能会让应用崩溃。要解决这个问题主要就是要避免大量线程同时重建缓存。我们可以利用互斥锁来解决，此方法只允许一个线程重建缓存，其他线程等待重建缓存的线程执行完，重新从缓存获取数据即可。

示例伪代码：

String get(String key) { 
    // 从Redis中获取数据 
    String value = redis.get(key); 
    // 如果value为空， 则开始重构缓存 
    if (value == null) { 
        // 只允许一个线程重建缓存， 使用nx， 并设置过期时间ex 
        String mutexKey = "mutext:key:" + key; 
        if (redis.set(mutexKey, "1", "ex 180", "nx")) { 
            // 从数据源获取数据 
            value = db.get(key); 
            // 回写Redis， 并设置过期时间 1
            redis.setex(key, timeout, value);
            // 删除key_mutex 
            redis.delete(mutexKey); 
        }
        // 其他线程休息50毫秒后重试 
        else { 
            Thread.sleep(50); 
            get(key); 
        } 
    } 
    return value; 
}

3.缓存雪崩

缓存雪崩是指，我们设置缓存时采用了相同的过期时间，所以大批量缓存在同一时间失效，导致流量会像奔逃的野牛一样，打向后端存储层。

由于缓存层承载着大量请求，有效地保护了存储层，但是如果缓存层由于某些原因不能提供服务(比如超大并发过来，缓存层支撑不住，或者由于缓存设计不好，类似大量请求访问bigkey，导致缓存能支撑的并发急剧下降)，于是大量请求都会达到存储层，存储层的调用量会暴增，造成存储层也会级联宕机的情况。

缓存雪崩是因为大面积的缓存失效，打崩了DB。而缓存击穿是值一个热点Key在不停的扛着大并发，当这个Key失效的瞬间，持续的大并发就穿破缓存，直接请求数据库。（也可以理解成缓存雪崩说的是很多key，而缓存击穿是某个热点key）

对于这种情况我们在批量增加缓存时好将这一批数据的缓存过期时间设置为一个时间段内的不同时间。

示例伪代码：

String get(String key) { 
    // 从缓存中获取数据 
    String cacheValue = cache.get(key); 
    // 缓存为空 
    if (StringUtils.isBlank(cacheValue)) { 
        // 从存储中获取 
        String storageValue = storage.get(key); 
        cache.set(key, storageValue);
        // 设置一个过期时间(300到600之间的一个随机数) 
        int expireTime = new Random().nextInt(300) + 300; 
        if (storageValue == null) { 
            cache.expire(key, expireTime); 13  } 
        return storageValue; 
    } else { 
        // 缓存非空 
        return cacheValue; 
    }
}