Redis缓存穿透，缓存击穿，缓存雪崩原因+解决方案

Bertha 。 2022-12-11 03:28 68阅读 0赞

转载：[Redis缓存穿透，缓存击穿，缓存雪崩原因+解决方案][Redis]

## 一、前言 ##

在我们日常的开发中，无不都是使用数据库来进行数据的存储，由于一般的系统任务中通常不会存在高并发的情况，所以这样看起来并没有什么问题，可是一旦涉及大数据量的需求，比如一些商品抢购的情景，或者是主页访问量瞬间较大的时候，单一使用数据库来保存数据的系统会因为面向磁盘，磁盘读/写速度比较慢的问题而存在严重的性能弊端，一瞬间成千上万的请求到来，需要系统在极短的时间内完成成千上万次的读/写操作，这个时候往往不是数据库能够承受的，极其容易造成数据库系统瘫痪，最终导致服务宕机的严重生产问题。

为了克服上述的问题，项目通常会引入NoSQL技术，这是一种基于内存的数据库，并且提供一定的持久化功能。

redis技术就是NoSQL技术中的一种，但是引入redis又有可能出现`缓存穿透`，`缓存击穿`，`缓存雪崩`等问题。本文就对这三种问题进行较深入剖析。

## 二、初认识 ##

*  **缓存穿透**：恶意用户模拟请求很多缓存中不存在的数据，由于缓存中都没有，导致这些请求短时间内直接落在了数据库上，导致数据库异常，甚至瞬间宕机。比如用一个不存在的用户id获取用户信息，不论缓存还是数据库都没有，若黑客利用此漏洞进行攻击可能压垮数据库。
 *  **缓存击穿**：key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
 *  **缓存雪崩**：缓存在同一时间内大量key过期（失效），接着无数的请求瞬间都落在了数据库中导致连接异常。

## 三、缓存穿透解决方案 ##

一个一定不存在缓存及查询不到的数据，由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。

**解决方案：**

*  方案一：采用布隆过滤器，将所有可能存在的数据哈希到一个足够大的bitmap中，一个一定不存在的数据会被 这个bitmap拦截掉，从而避免了对底层存储系统的查询压力。
 *  方案二：更为简单粗暴的方法。如果一个查询返回的数据为空（不管是数据不存在，还是系统故障），我们仍然把这个空结果进行缓存，但它的过期时间会很短，最长不超过五分钟。
 *  方案三：使用互斥锁排队。也是业界比价普遍的一种做法，即根据key获取value值为空时，锁上，从数据库中load数据后再释放锁。若其它线程获取锁失败，则等待一段时间后重试。这里要注意，分布式环境中要使用分布式锁，单机的话用普通的锁（synchronized、Lock）就够了。

互斥锁队列：
     public String getWithLock(String key, Jedis jedis, String lockKey, String uniqueId, long expireTime) {
        // 通过key获取value
        String value = redisService.get(key);
        if (StringUtil.isEmpty(value)) {
            // 分布式锁，详细可以参考https://blog.csdn.net/fanrenxiang/article/details/79803037
            //封装的tryDistributedLock包括setnx和expire两个功能，在低版本的redis中不支持
            try {
                boolean locked = redisService.tryDistributedLock(jedis, lockKey, uniqueId, expireTime);
                if (locked) {
                    value = userService.getById(key);
                    redisService.set(key, value);
                    redisService.del(lockKey);
                    return value;
                } else {
                    // 其它线程进来了没获取到锁便等待50ms后重试
                    Thread.sleep(50);
                    getWithLock(key, jedis, lockKey, uniqueId, expireTime);
                }
            } catch (Exception e) {
                log.error("getWithLock exception=" + e);
                return value;
            } finally {
                redisService.releaseDistributedLock(jedis, lockKey, uniqueId);
            }
        }
        return value;
    }
    
       
        
       
        
         1
         2
         3
         4
         5
         6
         7
         8
         9
         10
         11
         12
         13
         14
         15
         16
         17
         18
         19
         20
         21
         22
         23
         24
         25
         26
         27
         28

## 四、缓存击穿解决方案 ##

key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：缓存被“击穿”的问题。

解决方案：

*  使用互斥锁(mutex key)

业界比较常用的做法，是使用mutex。简单地来说，就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db，而是先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX或者Memcache的ADD）去set一个mutex key，当操作返回成功时，再进行load db的操作并回设缓存；否则，就重试整个get缓存的方法。

SETNX，是「SET if Not eXists」的缩写，也就是只有不存在的时候才设置，可以利用它来实现锁的效果。

redis代码：

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;      
              }
     }
    
       
        
       
        
         1
         2
         3
         4
         5
         6
         7
         8
         9
         10
         11
         12
         13
         14
         15
         16

memcache代码：

if (memcache.get(key) == null) {  
        // 3 min timeout to avoid mutex holder crash  
        if (memcache.add(key_mutex, 3 * 60 * 1000) == true) {  
            value = db.get(key);  
            memcache.set(key, value);  
            memcache.delete(key_mutex);  
        } else {  
            sleep(50);  
            retry();  
        }  
    }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13

## 五、缓存雪崩解决方案 ##

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

缓存正常从Redis中获取，示意图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70 1]  
缓存失效时的雪崩效应对底层系统的冲击非常可怕！大多数系统设计者考虑用

解决方案：

*  方案一：**加锁或者队列**的方式保证来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。
 *  方案二：还有一个简单方案就时将**缓存失效时间分散开**，比如我们可以在原有的失效时间基础上增加一个随机值，比如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;
    }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20
 *  21
 *  22

加锁排队只是为了减轻数据库的压力，并没有提高系统吞吐量。假设在高并发下，缓存重建期间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;
    }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20
 *  21
 *  22

解释说明：

*  缓存标记：记录缓存数据是否过期，如果过期会触发通知另外的线程在后台去更新实际key的缓存；
 *  缓存数据：它的过期时间比缓存标记的时间延长1倍，例：标记缓存时间30分钟，数据缓存设置为60分钟。这样，当缓存标记key过期后，实际缓存还能把旧数据返回给调用端，直到另外的线程在后台更新完成后，才会返回新缓存。

**参考链接**：  
https://blog.csdn.net/zeb\_perfect/article/details/54135506  
https://blog.csdn.net/fanrenxiang/article/details/80542580  
https://baijiahao.baidu.com/s?id=1619572269435584821&wfr=spider&for=pc  
https://blog.csdn.net/xlgen157387/article/details/79530877

[Redis]: https://blog.csdn.net/weixin_42740530/article/details/101693922?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522160126280719195188335740%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=160126280719195188335740&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~first_rank_v2~rank_v28-2-101693922.pc_first_rank_v2_rank_v28&utm_term=redis%E7%BC%93%E5%AD%98%E5%87%BB%E7%A9%BF%E5%92%8C%E9%9B%AA%E5%B4%A9&spm=1018.2118.3001.4187
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70]: /images/20221123/9879daec8b6643bdb59770fd08ea4815.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70 1]: /images/20221123/206b0fd9d8564663b1f7005ba15a91ed.png