缓存击穿和穿透以及雪崩详解

「爱情、让人受尽委屈。」 2022-12-25 10:54 108阅读 0赞

# 缓存处理流程 #

在互联网中，鉴于详情页面是被用户高频访问的，所以性能必须进行尽可能的优化。一般一个系统最大的性能瓶颈，就是数据库的io操作。从数据库入手也是调优性价比最高的切入点。一般分为两个层面，一是提高数据库sql本身的性能，二是尽量避免直接查询数据库。提高数据库本身的性能首先是优化sql，包括：使用索引，减少不必要的大表关联次数，控制查询字段的行数和列数。另外当数据量巨大是可以考虑分库分表，以减轻单点压力。

今天重点要讲的是另外一个层面：尽量避免直接查询数据库。

解决办法就是：缓存

缓存可以理解是数据库的一道保护伞，任何请求只要能在缓存中命中，都不会直接访问数据库。而缓存的处理性能是数据库10-100倍。

咱们就用Redis作为缓存系统进行优化。

针对在使用redis过程中，企业喜欢问的几个问题，下面详细解释缓存穿透、缓存雪崩和缓存击穿以及相应的解决方案。

# 一 缓存穿透 #

## 1.缓存穿透的系统定义 ##

按照KEY去查询VALUE，当KEY对应的VALUE一定不存在的时候并对KEY并发请求量很大的时候，就会对后端造成很大的压力。（查询一个必然不存在的数据。比如文章表，查询一个不存在的id，每次都会访问DB，如果有人恶意破坏，很可能直接对DB造成影响。）

由于缓存不命中，每次都要查询持久层。从而失去缓存的意义。

比如我们有一张数据库表，ID都是从1开始的(正数)：

但是可能有黑客想把我的数据库搞垮，每次请求的ID都是负数。这会导致我的缓存就没用了，请求全部都找数据库去了，但数据库也没有这个值啊，所以每次都返回空出去。

这就是缓存穿透：  
请求的数据在缓存大量不命中，导致请求走数据库。  
缓存穿透如果发生了，也可能把我们的数据库搞垮，导致整个服务瘫痪！

## 2缓存穿透的解决办法 ##

### **2.1.缓存层缓存空值** ###

在第一次查询数据库，该数据不存在，同样new一个空对象返回存储到redis当中。而为了避免缓存太多空值，占用更多空间，可以每次给空值设置一个过期时间，代码如下：

public SkuInfo getSkuInfo(Long skuId) {    Map<String, Object> result = new HashMap<>();    //获取skuinfo信息    SkuInfo skuInfo = productFeignClient.getSkuInfo(skuId);    if (ObjectUtils.isEmpty(skuInfo)){        skuInfo = new SkuInfo();    }    redisTemplate.opsForValue().set(RedisConst.SKUKEY_PREFIX + skuId + RedisConst.SKUKEY_SUFFIX,JSON.toJSONString(skuInfo),60*60,TimeUnit.SECONDS);    return skuInfo;}

而针对在存储层更新对象，原来没有的现在有数据了，导致缓存redis里面对应的还是空值，可以设置后台更新数据库时，主动删除空值并把最新数据存进缓存。

### **2.2.bloomfilter或者压缩filter(bitmap等等)提前拦截** ###

将数据库中所有的查询条件，放布隆过滤器中。当一个查询请求来临的时候，先经过布隆过滤器进行查，如果请求存在这个条件中，那么继续执行，如果不在，直接丢弃。

**备注 ：**

比如数据库中有10000个条件，那么布隆过滤器的容量size设置的要稍微比10000大一些，比如12000.

对于误判率的设置，根据实际项目，以及硬件设施来具体定。但一定不能设置为0，并且误判率设置的越小，哈希函数跟数组长度都会更多跟更长，那么对硬件，内存中间的要求就会相应的高

private st atic BloomFilter<Inte ger> bloomFi lt er = BloomFilter.create(Funnels.integerFue l(), s ize, 000 01)

有了siz跟误判率，那么布隆过滤器会产相应的哈希函数跟数组。

综上：我们可以利用布隆过滤器，将redis缓存击穿制在一个可容的范围内。

# 二 缓存雪崩 #

## 1 缓存雪崩的系统定义 ##

是指在我们设置缓存时采用了相同的过期时间或者redis挂掉了，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。

> 我们知道Redis不可能把所有的数据都缓存起来(内存昂贵且有限)，所以Redis需要对数据设置过期时间，并采用的是惰性删除+定期删除两种策略对过期键删除。Redis对过期键的策略+持久化

## 2 缓存雪崩的解决办法 ##

对于“对缓存数据设置相同的过期时间，导致某段时间内缓存失效，请求全部走数据库。”这种情况，非常好解决：

解决方法：

1、在缓存的时候给过期时间加上一个随机值，这样就会大幅度的减少缓存在同一时间过期。

2、对于“Redis挂掉了，请求全部走数据库”这种情况，我们可以有以下的思路：  
 事发前：实现Redis的高可用(主从架构+Sentinel（哨兵） 或者Redis Cluster（集群）)，尽量避免Redis挂掉这种情况发生。  
 事发中：万一Redis真的挂了，我们可以设置本地缓存(ehcache)+限流(hystrix)，尽量避免我们的数据库被干掉(起码能保证我们的服务还是能正常工作的)  
 事发后：redis持久化，重启后自动从磁盘上加载数据，快速恢复缓存数据。

# 三 缓存击穿 #

## 1.缓存击穿的系统定义 ##

是指对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：如果这个key在大量请求同时进来之前正好失效，那么所有对这个key的数据查询都落到db，我们称为缓存击穿。***\*** 缓存击穿与缓存雪崩的区别\*\*\*\**\*

## 2.缓存击穿的解决办法 ##

### 2.1.使用分布式锁 ###

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！

使用分布式锁，采用redis的KEY过期时间实现

Redis:命令

\# set skuid:1:info “OK” NX PX 10000

EX second ：设置键的过期时间为 second 秒。

PX millisecond ：设置键的过期时间为 millisecond 毫秒。

NX ：只在键(key)不存在时，才对键(key)进行设置操作。

XX ：只在键(key)已经存在时，才对键(key)进行设置操作。

Redis SET命令用于设置给定key的值，如果key已经存在其他值，SET就会覆盖，且无视类型。

在 Redis 2.6.12 以前版本， SET 命令总是返回 OK 。

从 Redis 2.6.12 版本开始， SET 在设置操作成功完成时，才返回 OK 。

删除操作缺乏原子性。

**场景：**

1.  index1执行删除时，查询到的lock值确实和uuid相等
2.  index1执行删除前，lock刚好过期时间已到，被redis自动释放
3.  index2获取了lock
4.  index1执行删除，此时会把index2的lock删除

解决：使用LUA脚本保证删除的原子性

代码示例如下：

// 使用lua脚本删除分布式锁 
    // lua，在get到key后，根据key的具体值删除keyDefaultRedisScript<Long> luaScript = new DefaultRedisScript<();
    //luaScript.setResultType(Long.class);
    luaScript.setScriptText("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end");
    redisTemplate.execute(luaScript, Arrays.asList("sku:" + skuId + ":lock"), uuid);

redis集群状态下的问题

1.  客户端A从master获取到锁
2.  在master将锁同步到slave之前，master宕掉了。
3.  slave节点被晋级为master节点
4.  客户端B取得了同一个资源被客户端A已经获取到的另外一个锁。安全失效！
    
    > \*\*解决：\*\*使用redisson 解决分布式锁

代码示例如下：

RLock lock = redisson.getLock("anyLock");// 最常使用lock.lock();// 加锁以后10秒钟自动解锁// 无需调用unlock方法手动解锁lock.lock(10, TimeUnit.SECONDS);/// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);if (res) { try { ... } finally { lock.unlock();}

### 2.2.分布式锁+AOP实现缓存 ###

随着业务中缓存及分布式锁的加入，业务代码变的复杂起来，除了需要考虑业务逻辑本身，还要考虑缓存及分布式锁的问题，增加了程序员的工作量及开发难度。而缓存的玩法套路特别类似于事务，而声明式事务就是用了aop的思想实现的。

以 @Transactional 注解为植入点的切点，这样才能知道@Transactional注解标注的方法需要被代理。

@Transactional注解的切面逻辑类似于@Around

模拟事务，缓存可以这样实现：

1.  自定义缓存注解@RedisCache（类似于事务@Transactional）
2.  编写切面类，使用环绕通知实现缓存的逻辑封装

代码如下：

@Target({ElementType.METHOD})
      @Retention(RetentionPolicy.RUNTIME)
      @Documented
      public @interface RedisCache {
      
        /**
         * 缓存key的前缀
         * @return
         */
        String prefix() default "cache";
    }

切面类：

@Component
    @Aspect
      public class RedisCacheAspect { 
      
        @Autowired
        private RedisTemplate redisTemplate;
      
        @Autowired
        private RedissonClient redissonClient;
      
        /** * 1.返回值object * 2.参数proceedingJoinPoint * 3.抛出异常Throwable * 4.proceedingJoinPoint.proceed(args)执行业务方法 */
        @Around("@annotation(com.mark.common.cache.RedisCache)")
        public Object cacheAroundAdvice(ProceedingJoinPoint point) throws Throwable { 
      
            Object result = null;
            // 获取连接点签名
            MethodSignature signature = (MethodSignature) point.getSignature();
            // 获取连接点的GmallCache注解信息
            GmallCache gmallCache = signature.getMethod().getAnnotation(RedisCache.class);
            // 获取缓存的前缀
            String prefix = gmallCache.prefix();
      
            // 组装成key
            String key = prefix + Arrays.asList(point.getArgs()).toString();
      
            // 1. 查询缓存
            result = this.cacheHit(signature, key);
      
            if (result != null) { 
                return result;
            }
      
            // 初始化分布式锁
            RLock lock = this.redissonClient.getLock("redisCache");
            // 防止缓存穿透 加锁
            lock.lock();
      
            // 再次检查内存是否有，因为高并发下，可能在加锁这段时间内，已有其他线程放入缓存
            result = this.cacheHit(signature, key);
            if (result != null) { 
                lock.unlock();
                return result;
            }
      
            // 2. 执行查询的业务逻辑从数据库查询
            result = point.proceed(point.getArgs());
            // 并把结果放入缓存
            this.redisTemplate.opsForValue().set(key, JSONObject.toJSONString(result));
      
            // 释放锁
            lock.unlock();
      
            return result;
        }
      
        /** * 查询缓存的方法 * * @param signature * @param key * @return */
        private Object cacheHit(MethodSignature signature, String key) { 
            // 1. 查询缓存
            String cache = (String)redisTemplate.opsForValue().get(key);
            if (StringUtils.isNotBlank(cache)) { 
                // 有，则反序列化，直接返回
                Class returnType = signature.getReturnType(); // 获取方法返回类型
                // 不能使用parseArray<cache, T>，因为不知道List<T>中的泛型
                return JSONObject.parseObject(cache, returnType);
            }
            return null;
        }
    }

最后在业务层使用注解实现缓存效果：

@RedisCache(prefix = RedisConst.SKUKEY_PREFIX)
      @Override
      public SkuInfo getSkuInfo(Long skuId) { 
      
        return getSkuInfoDB(skuId);
    }
    
    
    private SkuInfo getSkuInfoDB(Long skuId) { 
            SkuInfo skuInfo = skuInfoMapper.selectById(skuId);
    
            QueryWrapper<SkuImage> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("sku_id", skuId);
            List<SkuImage> skuImages = skuImageMapper.selectList(queryWrapper);
    
            skuInfo.setSkuImageList(skuImages);
    
            return skuInfo;
        }

代码解读：此处当类似于切面，当调用getSkuInfo方法时，会根据getSkuInfoDB(skuId)，查询数据库中的数据，然后将查询到的数据存入redis缓存当中。

到此缓存穿透、击穿和雪崩问题基本得到了解决那么面试官可能又会问既然数据查询这块没问题了，那么关于在查询和更新的过程中，怎么保证缓存和数据库的数据一致呢？我们下面就简单介绍下。

# 四 缓存与数据库双写一致 #

## 1 什么是缓存与数据库双写一致问题？ ##

如果仅仅查询的话，缓存的数据和数据库的数据是没问题的。但是，当我们要更新时候呢？各种情况很可能就造成数据库和缓存的数据不一致了。

这里不一致指的是：数据库的数据跟缓存的数据不一致

## 2 数据的读取流程 ##

> 上面讲缓存穿透的时候也提到了：如果从数据库查不到数据则不写入缓存。  
> 一般我们对读操作的时候有这么一个固定的套路：>如果我们的数据在缓存里边有，那么就直接取缓存的。  
> 如果缓存里没有我们想要的数据，我们会先去查询数据库，然后将数据库查出来的数据写到缓存中。  
> 最后将数据返回给请求
> 
> 从理论上说，只要我们设置了键的过期时间，我们就能保证缓存和数据库的数据最终是一致的。因为只要缓存数据过期了，就会被删除。随后读的时候，因为缓存里没有，就可以查数据库的数据，然后将数据库查出来的数据写入到缓存中。
> 
> 除了设置过期时间，我们还需要做更多的措施来尽量避免数据库与缓存处于不一致的情况发生。

## 3 数据的更新流程 ##

一般来说，执行更新操作时，我们会有两种选择：  
先操作数据库，再操作缓存  
先操作缓存，再操作数据库首先，要明确的是，无论我们选择哪个，我们都希望这两个操作要么同时成功，要么同时失败。所以，这会演变成一个分布式事务的问题。所以，如果原子性被破坏了，可能会有以下的情况：操作数据库成功了，操作缓存失败了。  
操作缓存成功了，操作数据库失败了。如果第一步已经失败了，我们直接返回Exception出去就好了，第二步根本不会执行。  
**更新时对缓存的操作**

操作缓存也有两种方案：更新缓存  
删除缓存一般我们都是采取删除缓存缓存策略的，原因如下：

1、 高并发环境下，无论是先操作数据库还是后操作数据库而言，如果加上更新缓存，那就更加容易导致数据库与缓存数据不一致问题。(删除缓存直接和简单很多)  
2、如果每次更新了数据库，都要更新缓存【这里指的是频繁更新的场景，这会耗费一定的性能】，倒不如直接删除掉。等再次读取时，缓存里没有，那我到数据库找，在数据库找到再写到缓存里边(体现懒加载)  
基于这两点，对于缓存在更新时而言，都是建议执行删除操作！正常的情况是这样的：先操作数据库，成功；  
再删除缓存，也成功；如果原子性被破坏了：

> 第一步成功(操作数据库)，第二步失败(删除缓存)，会导致数据库里是新数据，而缓存里是旧数据。  
> \\如果第一步(操作数据库)就失败了，我们可以直接返回错误(Exception)，不会出现数据不一致。  
> 如果在高并发的场景下，出现数据库与缓存数据不一致的概率特别低，也不是没有：  
> 1、 缓存刚好失效  
> 2、线程A查询数据库，得一个旧值  
> 3、线程B将新值写入数据库  
> 4、线程B删除缓存  
> 5、线程A将查到的旧值写入缓存  
> 要达成上述情况，还是说一句概率特别低：

因为这个条件需要发生在读缓存时缓存失效，而且并发着有一个写操作。而实际上数据库的写操作会比读操作慢得多，而且还要锁表，而读操作必需在写操作前进入数据库操作，而又要晚于写操作更新缓存，所有的这些条件都具备的概率基本并不大。删除缓存失败的解决思路：

> 将需要删除的key发送到消息队列中  
> 自己消费消息，获得需要删除的key  
> 不断重试删除操作，直到成功