首页 技术设计 数据库 redis

Redis缓存穿透、击穿、雪崩，解决方案重磅来袭！

Redis 经常用于系统中的缓存，这样可以解决目前 IO 设备无法满足互联网应用海量的读写请求的问题。

# 一、缓存穿透

缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求，如发起 id 为 - 1 的数据或者特别大的不存在的数据。有可能是黑客利用漏洞攻击从而去压垮应用的数据库。

# 1. 常见解决方案

对于缓存穿透问题，常见的解决方案有以下三种：

  • 验证拦截：接口层进行校验，如鉴定用户权限，对 ID 之类的字段做基础的校验，如 id<=0 的字段直接拦截；
  • 缓存空数据：当数据库查询到的数据为空时，也将这条数据进行缓存，但缓存的有效性设置得要较短，以免影响正常数据的缓存；
public Student getStudentsByID(Long id) {
    
    // 从 Redis 中获取学生信息
    Student student = redisTemplate.opsForValue()
        .get(String.valueOf(id));
    if (student != null) {
        return student;
    }
    
    // 从数据库查询学生信息，并存入 Redis
    student = studentDao.selectByStudentId(id);
    if (student != null) {
        redisTemplate.opsForValue()
            .set(String.valueOf(id), student, 60, TimeUnit.MINUTES);
    } else {
        // 即使不存在，也将其存入缓存中
        redisTemplate.opsForValue()
            .set(String.valueOf(id), null, 60, TimeUnit.SECONDS);
    }
    
    return student;
}
  • 使用布隆过滤器：布隆过滤器是一种比较独特数据结构，有一定的误差。当它指定一个数据存在时，它不一定存在，但是当它指定一个数据不存在时，那么它一定是不存在的。

# 2. 布隆过滤器

布隆过滤器是一种比较特殊的数据结构，有点类似与 HashMap，在业务中我们可能会通过使用 HashMap 来判断一个值是否存在，它可以在 O(1) 时间复杂度内返回结果，效率极高，但是受限于存储容量，如果可能需要去判断的值超过亿级别，那么 HashMap 所占的内存就很可观了。

而 BloomFilter 解决这个问题的方案很简单。首先用多个 bit 位去代替 HashMap 中的数组，这样的话储存空间就下来了，之后就是对 Key 进行多次哈希，将 Key 哈希后的值所对应的 bit 位置为 1。

当判断一个元素是否存在时，就去判断这个值哈希出来的比特位是否都为 1，如果都为 1，那么可能存在，也可能不存在（如下图 F）。但是如果有一个 bit 位不为 1，那么这个 Key 就肯定不存在。

img

注意： BloomFilter 并不支持删除操作，只支持添加操作。这一点很容易理解，因为你如果要删除数据，就得将对应的 bit 位置为 0，但是你这个 Key 对应的 bit 位可能其他的 Key 也对应着。

# 3. 缓存空数据与布隆过滤器的比较

上面对这两种方案都进行了简单的介绍，缓存空数据与布隆过滤器都能有效解决缓存穿透问题，但使用场景有着些许不同；

  • 当一些恶意攻击查询查询的 key 各不相同，而且数量巨多，此时缓存空数据不是一个好的解决方案。因为它需要存储所有的 Key，内存空间占用高。并且在这种情况下，很多 key 可能只用一次，所以存储下来没有意义。所以对于这种情况而言，使用布隆过滤器是个不错的选择；
  • 而对与空数据的 Key 数量有限、Key 重复请求效率较高的场景而言，可以选择缓存空数据的方案。

# 二、缓存击穿

缓存击穿是指当前热点数据存储到期时，多个线程同时并发访问热点数据。因为缓存刚过期，所有并发请求都会到数据库中查询数据。

# 1. 解决方案

  • 将热点数据设置为永不过期；
  • 加互斥锁：互斥锁可以控制查询数据库的线程访问，但这种方案会导致系统的吞吐量下降，需要根据实际情况使用。
public String get(key) {
    String value = redis.get(key);
    if (value == null) { // 代表缓存值过期
        // 设置 3min 的超时，防止 del 操作失败的时候，下次缓存过期一直不能 load db
        if (redis.setnx(key_mutex, 1, 3 * 60) == 1) {  // 代表设置成功
            value = db.get(key);
            redis.set(key, value, expire_secs);
            redis.del(key_mutex);
        } else {  // 这个时候代表同时候的其他线程已经 load db 并回设到缓存了，这时候重试获取缓存值即可
            sleep(50);
            get(key);  // 重试
        }
    } else {
        return value;      
    }
}

# 三、缓存雪崩

缓存雪崩发生有几种情况，比如大量缓存集中在或者缓存同时在大范围中失效，出现了大量请求去访问数据库，从而导致 CPU 和内存过载，甚至停机。

与缓存击穿的区别在于这里针对很多 key 缓存，前者则是某一个 key。

缓存正常从 Redis 中获取，示意图如下：

img

缓存失效瞬间示意图如下：

img

一个简单的雪崩过程：

  1. Redis 集群产生了大面积故障；
  2. 缓存失败，此时仍有大量请求去访问 Redis 缓存服务器；
  3. 在大量 Redis 请求失败后，这些请求将会去访问数据库；
  4. 由于应用的设计依赖于数据库和 Redis 服务，很快就会造成服务器集群的雪崩，最终导致整个系统的瘫痪。

# 1. 解决方案

  • 【事前】高可用缓存：高可用缓存是防止出现整个缓存故障。即使个别节点，机器甚至机房都关闭，系统仍然可以提供服务，Redis 哨兵 (Sentinel) 和 Redis 集群 (Cluster) 都可以做到高可用；
  • 【事中】缓存降级（临时支持）：当访问次数急剧增加导致服务出现问题时，我们如何确保服务仍然可用。在国内使用比较多的是 Hystrix，它通过熔断、降级、限流三个手段来降低雪崩发生后的损失。只要确保数据库不死，系统总可以响应请求，每年的春节 12306 我们不都是这么过来的吗？只要还可以响应起码还有抢到票的机会；
  • 【事后】Redis 备份和快速预热：Redis 数据备份和恢复、快速缓存预热。

缓存失效时的雪崩效应对底层系统的冲击非常可怕！大多数系统设计者考虑用加锁或者队列的方式保证来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。还有一个简单方案就时讲缓存失效时间分散开，比如我们可以在原有的失效时间基础上增加一个随机值，比如 1-5 分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。

加锁排队，伪代码如下：

// 伪代码
public object GetProductListNew() {
    int cacheTime = 30;
    String cacheKey = "product_list";
    String lockKey = cacheKey;
    String cacheValue = CacheHelper.get(cacheKey);
    if (cacheValue != null) {
        return cacheValue;
    } else {
        synchronized(lockKey) {
            cacheValue = CacheHelper.get(cacheKey);
            if (cacheValue != null) {
                return cacheValue;
            } else {
              // 这里一般是 sql 查询数据
                cacheValue = GetProductListFromDB(); 
                CacheHelper.Add(cacheKey, cacheValue, cacheTime);
            }
        }
        return cacheValue;
    }
}
加锁排队只是为了减轻数据库的压力，并没有提高系统吞吐量。假设在高并发下，缓存重建期间 key 是锁着的，这是过来 1000 个请求 999 个都在阻塞的。同样会导致用户等待超时，这是个治标不治本的方法！
注意：加锁排队的解决方式分布式环境的并发问题，有可能还要解决分布式锁的问题；线程还会被阻塞，用户体验很差！因此，在真正的高并发场景下很少使用！
随机值伪代码：
// 伪代码
public object GetProductListNew() {
    int cacheTime = 30;
    String cacheKey = "product_list";
    // 缓存标记
    String cacheSign = cacheKey + "_sign";
    String sign = CacheHelper.Get(cacheSign);
    // 获取缓存值
    String cacheValue = CacheHelper.Get(cacheKey);
    if (sign != null) {
        return cacheValue; // 未过期，直接返回
    } else {
        CacheHelper.Add(cacheSign, "1", cacheTime);
        ThreadPool.QueueUserWorkItem((arg) -> {
      // 这里一般是 sql 查询数据
            cacheValue = GetProductListFromDB(); 
          // 日期设缓存时间的 2 倍，用于脏读
          CacheHelper.Add(cacheKey, cacheValue, cacheTime * 2);                 
        });
        return cacheValue;
    }
}
解释说明：
  • 缓存标记：记录缓存数据是否过期，如果过期会触发通知另外的线程在后台去更新实际 key 的缓存；
  • 缓存数据：它的过期时间比缓存标记的时间延长 1 倍，例：标记缓存时间 30 分钟，数据缓存设置为 60 分钟。这样，当缓存标记 key 过期后，实际缓存还能把旧数据返回给调用端，直到另外的线程在后台更新完成后，才会返回新缓存。
关于缓存崩溃的解决方法，这里提出了三种方案：使用锁或队列、设置过期标志更新缓存、为 key 设置不同的缓存失效时间，还有一种被称为 “二级缓存” 的解决方法。
针对业务系统，永远都是具体情况具体分析，没有最好，只有最合适。
于缓存其它问题，缓存满了和数据丢失等问题，大伙可自行学习。最后也提一下三个词 LRU、RDB、AOF，通常我们采用 LRU 策略处理溢出，Redis 的 RDB 和 AOF 持久化策略来保证一定情况下的数据安全。
参考：
https://www.cnblogs.com/zhangweizhong/p/6258797.html
https://www.cnblogs.com/jojop/p/141