缓存穿透、缓存击穿与缓存雪崩

缓存穿透

缓存穿透：客户端查询的数据，在缓存和数据库都不存在，这样缓存永远不会命中。当用户很多的时候，缓存都没有命中，于是请求都打到了持久层数据库。这可能会给数据库造成很大的压力。

解决方案

• 缓存空对象：如果查询数据在数据库中不存在，仍把空字符串进行缓存，再设置一个较短的过期时间，不超过五分钟。（查询时，缓存命中时，要判断是否为空字符串）

  • 优点：实现简单，维护方便

  • 缺点：

    • 额外的内存消耗

    • 可能造成短期的不一致（已经缓存了空字符串，之后又插入了数据，插入数据不会删除缓存。解决：新增的时候，删除缓存）

• 布隆过滤器：在缓存前面加一层过滤器，可以过滤到一些无效的请求。

  • 优点：内存占用较少，布隆过滤器不存储数据本身。

  • 缺点：

    • 有一定的误判率，但是可以通过调整参数来降低。

    • 无法获取元素本身

    • 很难删除元素

• 增强id的复杂度，避免被猜测id规律

• 做好数据的基础格式校验

• 加强用户权限校验

• 做好热点参数的限流

布隆过滤器

        布隆过滤器（Bloom Filter）是1970年由布隆提出的。一种数据结构，它实际上是一个很长的二进制数组（初始默认值都是0）和一系列哈希函数。布隆过滤器可以用于判断一个元素是否在一个集合中。它可以判断某个数据可能存在和一定不存在。

实际使用：将所有可能存在的key加入到布隆过滤器中，当访问不存在的key时迅速返回避免缓存及 DB挂掉。

Google开源Guava中的布隆过滤器

导入依赖

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>28.0-jre</version>
</dependency>

创建了一个最多存放 最多 10000 个整数的布隆过滤器，并且我们可以容忍误判的概率为1%

import com.google.common.base.Charsets;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
import com.google.common.hash.Funnels;
 
public class GuavaBloomFilter {
    public static void main(String[] args) {
        BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), 10000, 0.01);
 
        // 往过滤器中添加元素
        bloomFilter.put("10086");
 
        // 查询
        System.out.println(bloomFilter.mightContain("123456")); // false
        System.out.println(bloomFilter.mightContain("10086")); // true
    }
}

缓存击穿（热点key问题）

介绍：当一个热点的并且重建业务较复杂的key过期（创建缓存比较复杂，如需要连表查询等，创建缓存耗时长）时，高并发的请求瞬间导致数据库压力瞬间增大。

比如热搜排行上，一个热点新闻被同时大量访问，当缓存过期的一瞬间就可能导致缓存击穿。

解决方案：逻辑过期 和 加分布式锁

逻辑过期

• 不设置过期时间，把过期时间设置在redis的value中

定义RedisData类，设置存储的时候使用此对象

@Data
public class RedisData {
    private LocalDateTime expireTime;
    private Object data;
}

实际业务使用

private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
 
public void saveShop2Redis(Long id, Long expireSeconds){
    // 查询数据
    Shop shop = getById(id);
    // 封装逻辑过期时间
    RedisData redisData = new RedisData();
    redisData.setData(shop);
    redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));
    // 写入Redis
    stringRedisTemplate.opForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(redisData));
}
 
public Shop queryWithLogicalExpire( Long id ) {
    String key = CACHE_SHOP_KEY + id;
    // 1.从redis查询商铺缓存
    String json = stringRedisTemplate.opsForValue().get(key);
    // 2.判断是否存在
    if (StrUtil.isBlank(json)) {
        // 3.存在，直接返回
        return null;
    }
    // 4.命中，需要先把json反序列化为对象
    RedisData redisData = JSONUtil.toBean(json, RedisData.class);
    Shop shop = JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);
    LocalDateTime expireTime = redisData.getExpireTime();
    // 5.判断是否过期
    if(expireTime.isAfter(LocalDateTime.now())) {
        // 5.1.未过期，直接返回店铺信息
        return shop;
    }
    // 5.2.已过期，需要缓存重建
    // 6.缓存重建
    // 6.1.获取互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    boolean isLock = tryLock(lockKey);
    // 6.2.判断是否获取锁成功
    if (isLock){
        CACHE_REBUILD_EXECUTOR.submit( ()->{
 
            try{
                //重建缓存
                this.saveShop2Redis(id,20L);
            }catch (Exception e){
                throw new RuntimeException(e);
            }finally {
                unlock(lockKey);
            }
        });
    }
    // 6.4.返回过期的商铺信息
    return shop;
}

加互斥锁（分布式锁）

// 加互斥锁(setnx并设置有效期)
private boolean tryLock(String key) {
    Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
    return BooleanUtil.isTrue(flag);
}
 
// 释放互斥锁
private void unlock(String key) {
    stringRedisTemplate.delete(key);
}

 业务代码

public Shop queryWithMutex(Long id)  {
    String key = CACHE_SHOP_KEY + id;
    // 1、从redis中查询商铺缓存
    String shopJson = stringRedisTemplate.opsForValue().get("key");
    // 2、判断是否存在
    if (StrUtil.isNotBlank(shopJson)) {
        // 存在,直接返回
        return JSONUtil.toBean(shopJson, Shop.class);
    }
    //判断命中的值是否是空值
    if (shopJson != null) {
        //返回一个错误信息
        return null;
    }
    // 4.实现缓存重构
    //4.1 获取互斥锁
    String lockKey = "lock:shop:" + id;
    Shop shop = null;
    try {
        boolean isLock = tryLock(lockKey);
        // 4.2 判断否获取成功
        if(!isLock){
            //4.3 失败，则休眠重试
            Thread.sleep(50);
            return queryWithMutex(id);
        }
        //4.4 成功，根据id查询数据库
        shop = getById(id);
        // 5.不存在，返回错误
        if(shop == null){
            //将空值写入redis
            stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
            //返回错误信息
            return null;
        }
        //6.写入redis
        stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),CACHE_NULL_TTL,TimeUnit.MINUTES);
 
    }catch (Exception e){
        throw new RuntimeException(e);
    }
    finally {
        //7.释放互斥锁
        unlock(lockKey);
    }
    return shop;
}

互斥锁方案：由于保证了互斥性，所以数据一致，且实现简单，因为仅仅只需要加一把锁而已，也没其他的事情需要操心，所以没有额外的内存消耗，缺点在于有锁就有死锁问题的发生，且只能串行执行性能肯定受到影响。

逻辑过期方案： 线程读取过程中不需要等待，性能好，有一个额外的线程持有锁去进行重构数据，但是在重构数据完成前，其他的线程只能返回之前的数据，且实现起来麻烦。

缓存雪崩

介绍：一个时间段内大量缓存key集中过期或Redis服务宕机，导致大量请求到达数据库，带来巨大压力。

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

解决方案

• 利用Redis集群提高服务的可用性：redis有可能挂掉，那多增设几台redis，这样一台挂掉之后其他的还可以继续工作，其实就是搭建的集群(异地多活)

• 给缓存业务添加降级限流策略：在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个key只允许一个线程查询数据和写缓存，其他线程等待。

• 给业务添加多级缓存（Nginx缓存、Redis缓存、JVM缓存、数据库缓存）

• 错开不同key的过期时间，防止在短时间内大量key一起过期。