高并发下缓存失效问题-缓存穿透、缓存击穿、缓存雪崩、Redis分布式锁简单实现、Redisson实现分布式锁

文章目录

缓存基本使用范式暴露的几个问题
缓存失效问题---缓存穿透
缓存失效问题---缓存击穿
- 一、单机锁
- - 正确的锁粒度
  - 不正确的锁粒度无法保证查询数据库次数是唯一
- 二、分布式锁
- - getCatalogJsonData()
  - 分布式锁演进---基本原理
  - 分布式锁(加锁)演进一：删锁失败导致死锁
  - 分布式锁(加锁)演进二：给‘锁’设置过期时间防止死锁
  - 分布式锁(加锁)演进三：必须保证过期时间和占锁动作原子性
  - 分布式锁(解锁)演进一：业务逻辑执行时间大于‘锁’的过期时间
  - 分布式锁(解锁)演进二：UUID保证删除的是自己的‘锁’
  - 分布式锁(解锁)演进三：lua脚本保证删‘锁’原子性
- 三、锁的自动续期
- 四、Redis简单实现分布式锁的完整代码
缓存失效问题---缓存雪崩
分布式锁---Redisson

缓存基本使用范式暴露的几个问题

{
	1、先查询缓存
	2、if(缓存没有命中){
		2.1、查询数据库
		2.2、查询结果放入缓存
		2.3、同时return结果
	}
	3、缓存命中直接return缓存数据
}

如下；使用缓存高效的查询‘三级分类’数据，就完全遵循上面提到的范式

    public Map<Long, List<Catalog2VO>> getCatalogJsonBaseMethod() {
        String key = ProductConstant.RedisKey.INDEX_CATEGORY_JSON;
        // 1、从缓存中获取数据
        String categoryListFromCache = redisTemplate.opsForValue().get(key);
        if (!StringUtils.hasText(categoryListFromCache)) {
            // 2.1、缓存没有命中，查询数据库
            Map<Long, List<Catalog2VO>> catalogJsonFromDB = getCatalogJsonFromDB();
            // 2.2、将查询结果放入缓存
            redisTemplate.opsForValue().set(key,JSON.toJSONString(catalogJsonFromDB));
            return catalogJsonFromDB;
        }
        // 3、缓存命中便直接return
        return JSON.parseObject(categoryListFromCache, new TypeReference<>() {});
    }

该范式在高并发、分布式下会暴露以下几个问题，这也是本章需要解决和讨论的点

高并发缓存失效之缓存穿透
高并发缓存失效之缓存击穿
高并发缓存失效之缓存雪崩
分布式架构下的分布式锁

缓存失效问题—缓存穿透

请求查询一个百分百不存在的数据

假设id=idooy这条记录在数据库中压根不存在；按照请求处理逻辑先查询缓存，但因为这本就是一条不存在的记录(假设成立)，因此缓存也不可能命中，缓存不命中接着就会查询数据库；如果没有将这一次请求查询的null写入缓存，这将导致id=idooy这条请求每次都要去数据库，直接失去了缓存的意义

风险： 利用不存在的数据发送大量请求，数据库瞬时压力增大，最终导致数据库崩溃
解决： 将null结果进行缓存，并加入短暂的过期时间；有时查询固定的值，不需要请求携带参数，这种情况本身就不会出现缓存穿透

缓存失效问题—缓存击穿

某一个Key在高并发请求期间刚好过期失效

对于一个设置了过期时间的Key，如果这个Key在将来的某个时间被高并发访问期间刚好过期失效，那么高并发的请求压力直接给到数据库
解决： 加锁；对同一个Key的高并发请求保证只有一个请求打给数据库；其他请求等待并最终从缓存中获取；下面讨论单机锁和分布式锁

一、单机锁

单机锁是指在单体应用中或同一个进程中利用锁的排他性保证高并发期间某个Key失效时只有一个请求去数据库进行查询来避免缓存击穿

代码实现如下所示：

    @Override
    public Map<Long, List<Catalog2VO>> getCatalogJson() {
        String key = ProductConstant.RedisKey.INDEX_CATEGORY_JSON;
        // 1、从缓存中获取数据
        String categoryListFromCache = redisTemplate.opsForValue().get(key);
        if (!StringUtils.hasText(categoryListFromCache)) {
            // 2、缓存没有命中，查询数据库，加锁保证数据库只查询一次
            // 因为当前this实例为单例，故可以作为锁资源使用
            synchronized (this) {
                // 2.1、高并发下必然有N个请求同时等待竞争锁，所以竞争到锁的第一件事就是再查一遍缓存
                String result = redisTemplate.opsForValue().get(key);
                if (StringUtils.hasText(result)) {
                    return JSON.parseObject(result, new TypeReference<>() {});
                }
                // 2.2、缓存依旧没有命中的情况下查询数据库
                Map<Long, List<Catalog2VO>> catalogJsonFromDB = getCatalogJsonFromDB();
                // 2.3、将查询结果放入缓存
                redisTemplate.opsForValue().set(key,JSON.toJSONString(catalogJsonFromDB),2,TimeUnit.HOURS);
                return catalogJsonFromDB;
            }
        }
        // 3、缓存命中便直接return
        return JSON.parseObject(categoryListFromCache, new TypeReference<>() {});
    }

正确的锁粒度

在这里插入图片描述

不正确的锁粒度无法保证查询数据库次数是唯一

在这里插入图片描述

二、分布式锁

上面单机锁本质就是使用当前进程中的某个单例对象充当锁资源；在微服务架构分布式部署下，同一个商品服务可能部署N多个，此时每个服务进程之间相互隔离。

在这里插入图片描述
因此；本地锁，只能锁住当前进程，分布式架构下需要分布式锁

getCatalogJsonData()

    private Map<Long, List<Catalog2VO>> getCatalogJsonData() {
        String key = ProductConstant.RedisKey.INDEX_CATEGORY_JSON;
        String result = redisTemplate.opsForValue().get(key);
        if (!StringUtils.hasText(result)) {
            Map<Long, List<Catalog2VO>> catalogJsonFromDB = getCatalogJsonFromDB();
            redisTemplate.opsForValue().set(key,JSON.toJSONString(catalogJsonFromDB),2,TimeUnit.HOURS);
            return catalogJsonFromDB;
        }
        return JSON.parseObject(result, new TypeReference<>() {
        });
    }

分布式锁演进—基本原理

所有的‘商品服务’可以同时去一个地方“占坑”，如果占到就执行逻辑，否则就必须等待，直到释放锁。
“占坑”可以去Redis，也可以去数据库，可以去任何只要“商品服务”都能访问到的地方
在这里插入图片描述

分布式锁(加锁)演进一：删锁失败导致死锁

在这里插入图片描述
如上图；执行业务逻辑出现异常或者在删锁前系统宕机(kill -9)；直接导致没有执行删锁操作。那么其他请求就无法"成功占锁"，造成死锁。

接下来给"锁"设置过期时间防止死锁。即使删锁失败也会自动删除

分布式锁(加锁)演进二：给‘锁’设置过期时间防止死锁

在这里插入图片描述
所以，“占锁+设置过期时间”必须保证原子性

分布式锁(加锁)演进三：必须保证过期时间和占锁动作原子性

Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "ok",3,TimeUnit.SECONDS);

在这里插入图片描述

分布式锁(解锁)演进一：业务逻辑执行时间大于‘锁’的过期时间

业务逻辑执行时间超过‘锁’的过期时间；这也就意味着业务逻辑执行完毕以后删的就不是自己的锁。
试想如下高并发场景下, 假设‘锁’的过期时间为10s，业务的执行时间为15s;

①号请求执行到第10s,‘锁’自动过期；②号请求立马占锁成功执行业务逻辑。
在第15s①号业务逻辑执行完毕，成功删除锁。很显然此时①号删除的就不是自己的锁(自己的锁在第10s的时候已自动删除了)，而是②号的锁。
同时在15s这一时刻①号删了②号的锁；接着3号占锁成功，如此情况下‘锁永久失效’

在这里插入图片描述
该况下暴露的问题本质就是锁删除了他人的锁；那么接下来就通过唯一ID保证线程删除的是自己的锁

分布式锁(解锁)演进二：UUID保证删除的是自己的‘锁’

在占锁的时候，值指定为uuid，每个人匹配是自己的锁才删除
在这里插入图片描述
如图；问题还是暴露了出来。get(“lock”)并且equals成立，此时锁刚好自动过期删除了，另一个线程占锁成功了，此时再执行delete删锁同样删除的不是自己的锁。
所以这个问题的本质就是删锁的过程不能保证原子性

分布式锁(解锁)演进三：lua脚本保证删‘锁’原子性

如下图；官方提供了‘解锁’的建议和保证解锁过程原子性的lua脚步

锁的值不要设置固定字符串，而是设置一个不可猜测的大随机字符串，称为token。
不是用DEL释放锁，而是发送一个脚本，仅在值匹配时才删除键

根据官方提示；解锁的核心业务代码片段

// 解锁
redisTemplate.execute(new DefaultRedisScript<>(getLuaScript(),Long.class),Arrays.asList("lock"),uuid);

private String getLuaScript(){
        return "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                "then\n" +
                "    return redis.call(\"del\",KEYS[1])\n" +
                "else\n" +
                "    return 0\n" +
                "end";
    }

在这里插入图片描述

三、锁的自动续期

业务执行时间超长；业务逻辑还未执行完毕‘锁’自动过期了，最简单的方式就是给‘锁’设置足够长的时间。
但完美的解决该问题，自己写代码实现还是很困难的，所以这个问题就抛出Redisson,它提供的分布式锁会解决上面提到的所有问题；包括锁的自动续期

四、Redis简单实现分布式锁的完整代码

加锁原子性命令；保证’设置过期时间和占锁’是原子性操作
解锁原子性命令；uuid保证删的是自己的锁；lua脚本保证了删锁的原子性
设置‘锁’的过期时间足够长，确保业务逻辑执行时间不会超过过期时间这种简单粗暴的方式来解决‘锁’过期自动续期的问题

private Map<Long, List<Catalog2VO>> getCatalogJsonWithRedisLock() {
        // 所有的请求进来先占坑，即抢占锁
        String uuid = UUID.randomUUID().toString();
        // 原子性命令；保证'设置过期时间和占锁'是原子性操作
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3,TimeUnit.SECONDS);
        if (lock) {
            // "占坑"成功，执行业务逻辑
            Map<Long, List<Catalog2VO>> result;
            try {
                result = getCatalogJsonData();
            } finally {
                // 解锁：uuid保证删的是自己的锁；lua脚本保证了删锁的原子性
                redisTemplate.execute(new DefaultRedisScript<>(getLuaScript(),Long.class),Arrays.asList("lock"),uuid);
            }
            return result;
        }else {
            // "占坑"失败,自旋
            try {
                // 防止栈溢出
                TimeUnit.MILLISECONDS.sleep(200);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return getCatalogJsonWithRedisLock();
        }
    }

    private String getLuaScript(){
        return "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                "then\n" +
                "    return redis.call(\"del\",KEYS[1])\n" +
                "else\n" +
                "    return 0\n" +
                "end";
    }

    private Map<Long, List<Catalog2VO>> getCatalogJsonData() {
        String key = ProductConstant.RedisKey.INDEX_CATEGORY_JSON;
        String result = redisTemplate.opsForValue().get(key);
        if (!StringUtils.hasText(result)) {
            Map<Long, List<Catalog2VO>> catalogJsonFromDB = getCatalogJsonFromDB();
            redisTemplate.opsForValue().set(key,JSON.toJSONString(catalogJsonFromDB),2,TimeUnit.HOURS);
            return catalogJsonFromDB;
        }
        return JSON.parseObject(result, new TypeReference<>() {
        });
    }

缓存失效问题—缓存雪崩

某一时刻大量的Key同时失效

假设缓存中大量的Key使用了相同过期时间，这直接导致在将来的某个时刻这些Key同时失效；此时再大量请求这些Key压力都来到了数据库，使数据库瞬时压力过大可能出现崩溃
解决： 再原有的失效时间上增加一个随机值，这样每个缓存的过期时间的重复率就会很低，也就很难出现Key大面积同时失效导致缓存雪崩问题

// 再原有的失效时间基础上添加随机时间片
// 这里没有增加随机时间片，因为Key的数量有限，足以保证失效时间的离散分布
redisTemplate.opsForValue().set(key,JSON.toJSONString(catalogJsonFromDB),2,TimeUnit.HOURS);