redis缓存问题(数据库一致性,穿透,雪崩,击穿)

1.缓存介绍

缓存就是数据交换的缓冲区，是储存数据的临时地方，一般读写性能较高

缓存的优点

• 降低后端负载
• 提高读写效率，降低响应时间

缓存的成本

• 数据的一致性成本低
• 代码维护成本
• 运维成本

2.缓存更新策略

业务场景

• 低一致要求：使用内存淘汰机制。列如缓存店铺的类型那些基本不会发生改变的数据
• 高一致要求：主动更新，并超时删除作为兜底方案，列如店铺详情的缓存

主动更新策略

• 由缓存的调用者，在更新数据库的同时更新缓存
• 缓存与数据库整合成一个服务，由服务来维护一致性。调用者用该服务，无序关注缓存一致性问题
• 调用者只操作缓存，其他由线程异步将缓存数据持久化到数据库保证一致性

由于第二种和第三种并没有更好的第三方工具可以为我们提供服务，第一种方案的可控制性较好，因此我们采用第一种方案

3.操作缓存和数据库时仍有问题需要考虑

1. 删除缓存还是更新缓存

   更新缓存：每次的更新数据库都更新缓存，无效的写操作较多 （不推荐）

   删除缓存：在更新数据库时让缓存失效，查询时再更新缓存

2. 如何保证缓存和数据库的操作同时成功或同时失败

   单体应用：将缓存的数据库操作放在一个事务中

   分布式系统：利用TCC等分布式解决方案

3. 先操作缓存还是先操作数据库

   先删除缓存，操作数据库

   假设我们此时 缓存和数据库中的数据都为10

​

​ 当出现这种情况时，会出现异常；此刻我们线程进入先删除了缓存，由于缓存的读写是基于内存的速度很快，而数据的操作时基于磁盘IO的，速度较慢，此刻当线程2进入时候，查询缓存未命中则会继续查询数据库，此刻数据库仍未10，再讲数据库中的10写入到缓存当中，线程2执行结束，线程1开始执行，此时线程1将值更新为20，数据库和缓存不一致。

​ 先操作数据库再操作缓存

​ 这种情况也会出现异常，但出现的几率很小，我们来进行分析

​ 当查询数据库时，未命中，我们查询数据库，此时数据库为10，在我们要写入缓存的时候（纳秒级别），此时线程2抢占了cpu资源进行更新操作，将数据库中的数据更为20，删除缓存，此刻没有删除缓存无效，执行完后线程1开始运行，将线程1查询到数据库中的值10，写入到缓存当中为20

修改数据代码演示

@Override
@Transactional
public Result update(Shop shop) {
    if (shop.getId() == null){
        return Result.fail("店铺id不能为空");
    }
    //先更新数据库再删除缓存
    updateById(shop);
    stringRedisTemplate.delete(RedisConstants.CACHE_SHOP_KEY+shop.getId());

    return Result.ok();
}
1
2
3
4
5
6
7
8
9
10
11
12

4.缓存穿透

缓存穿透：是指客户端请求的数据在缓存中的数据和数据库都不存在，这样的缓存永远不会生效，这样请求都会打到数据库

常见的解决方案有两种

• 缓存空对象

  • 优点：实现简单，维护方便
  • 缺点：额外的内存消耗（可以使用较短时间的TTL来解决），可能造成短期的不一致问题

• 布隆过滤器（基于某种算法将数据的hash以二进制的形式保存）当布隆过滤器判断存在，不应当存在，判断不存在一定不存在）

  • 优点：内存占用少，没有多余的key
  • 缺点：实现复杂，可能出现误判的情况

改造方案

缓存穿透的解决方案

• 缓存null值
• 布隆过滤器
• 做好基础的格式校验
• 加强用户权限校验
• 做好热点参数的限流

5.缓存雪崩

缓存雪崩是指：在同一段大量的缓存key同时失效或者Redis服务宕机，导致大量的请求到达数据库，带来巨大压力

解决方案：

• 给不同的key的TTL添加随机值
• 利用Redis集群提高服务的可用性
• 给缓存业务添加降级限流策略
• 给业务添加多级缓存

6.缓存击穿

缓存击穿问题也叫热点key问题，就是一个高并发访问并且缓存重建业务比较复杂的key突然失效，会在瞬间给数据库带来巨大的压力

现象描述

互斥锁的解决方案：

• 互斥锁
• 逻辑删除

我们可以通过string的setnx 来实现互斥

当我们第一次setnx lock ’ '时候只有第一次能够成功 获得锁get lock ‘’

//上锁   
private boolean tryLock(String key){
        Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(aBoolean);
    }
//解锁
private boolean unlock(String key){
        Boolean delete = stringRedisTemplate.delete(key);
        return BooleanUtil.isTrue(delete);
    }
1
2
3
4
5
6
7
8
9
10

//互斥锁
public Shop queryWithMutex(Long id){
    String lockKey = null;
    Shop shop = null;
    try {
        String key = RedisConstants.CACHE_SHOP_KEY+id;
        //1.先尝试从redis中查取缓存
        String s = stringRedisTemplate.opsForValue().get(key);
        //2.判断是否存在
        if (StrUtil.isNotBlank(s)){
            //3.存在直接返回
            return JSONUtil.toBean(s, Shop.class);
        }
        if (s != null){
            //返回一个错误信息
            return null;
        }

        //4. 实现缓存重建
        //4.1 获取互斥锁
        lockKey = "lock:shop:"+id;
        //4.2 判断是否获取成功
        if (!tryLock(lockKey)){
            //4.3 失败 休眠重试
            Thread.sleep(50);
            return queryWithMutex(id);
        }
        shop = getById(id);
        Thread.sleep(200);
        if (shop == null){
            //5.不存在返回错误
            // 解决缓存穿透
            stringRedisTemplate.opsForValue().set(key,"",RedisConstants.CACHE_NULL_TTL,TimeUnit.MINUTES);
            return null;
        }

        //6.存在，添加到Redis当中
        stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),30, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    } finally {
        //释放互斥锁
        unlock(lockKey);
    }
    //7.返回
    return shop;
}
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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

public Shop queryWithLogicDelete(Long id, Long seconds){
        String key = RedisConstants.CACHE_SHOP_KEY+id;
        //1.先尝试从redis中查取缓存
        String s = stringRedisTemplate.opsForValue().get(key);
        //2.判断是否存在
        if (StrUtil.isBlank(s)){
            //3.不存在直接返回
            return null;
        }
        //4.命中需要先把JSON反序列化为对象
        RedisData redisData = JSONUtil.toBean(s, RedisData.class);
        JSONObject data = (JSONObject) redisData.getData();
        Shop shop1 = JSONUtil.toBean(data, Shop.class);
        LocalDateTime expireTime = redisData.getExpireTime();
        //5.判断缓存是否过期  这里实际得到的是JsonObject对象
        if (expireTime.isAfter(LocalDateTime.now())){
            //过期时间在当前时间之后 说明未过期
            //5.1 未过期，直接返回店铺信息
            return shop1;
        }
        //5.2 已过期需要进行缓存重建
        //6. 进行缓存重建
        //6.1 获取互斥锁
        String lockKey = RedisConstants.LOCK_SHOP_KEY+id;
        boolean isLock = tryLock(lockKey);
        //6.2 判断是否获取锁成功
        if (isLock){
            //6.3 成功开启独立线程 实现缓存重建
            try {
                CACHE_REBUILD_EXECUTOR.submit(()->{
                    this.saveShop2Redis(id,20L);
                });
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //释放锁
                unlock(lockKey);
            }

        }
        //6.4 返回过期的店铺信息
        return shop1;
    }

    //自定义线程池
    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

使用JMeter性能测试工具，哇真的是这样 好神奇好牛逼

JMeter入门教程 - 简书 (jianshu.com)

7.封装缓存穿透和缓存击穿的工具类

大佬写的真的强

/**
 * @author XingLuHeng
 * @date 2022/4/24 10:59)
 * @description 封装缓存穿透和缓存击穿的工具类
 */
@Slf4j
@Component
public class CacheClient {

    private final StringRedisTemplate stringRedisTemplate;

    public CacheClient(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    public void set(String key, Object value, Long time, TimeUnit unit){
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value),time,unit);
    }

    public void setLogicalExpire(String key, Object value, Long time, TimeUnit timeUnit){
        //设置逻辑过期
        RedisData redisData = new RedisData();
        redisData.setData(value);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(timeUnit.toSeconds(time)));
        //写入redis

        stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(redisData));
    }

    public <R,ID> R queryWithPassThrough(String keyPrefix, ID id, Class<R> type, Function<ID,R> dbFallback,Long time, TimeUnit timeUnit){
            String key = keyPrefix+id;
            //1.先尝试从redis中查取缓存
            String json = stringRedisTemplate.opsForValue().get(key);
            //2.判断是否存在
            if (StrUtil.isNotBlank(json)){
                //3.存在直接返回
                return JSONUtil.toBean(json, type);
            }
        if (json != null){
            //返回一个错误信息
            return null;
        }
        //4.不存在，根据id查询数据库
        R r = dbFallback.apply(id);
        if (r == null){
            //5.不存在返回错误
            // 解决缓存穿透
            stringRedisTemplate.opsForValue().set(key,"",RedisConstants.CACHE_NULL_TTL,TimeUnit.MINUTES);
            return null;
        }

        //6.存在，添加到Redis当中
        this.set(key,JSONUtil.toJsonStr(r),time,timeUnit);
        //7.返回
        return r;
    }

    public <R,ID> R queryWithLogicDelete(String keyPrefix,ID id,Class<R> typeClass,Function<ID,R> dbFallback,Long time,TimeUnit timeUnit){
        String key = keyPrefix;
        //1.先尝试从redis中查取缓存
        String s = stringRedisTemplate.opsForValue().get(key);
        //2.判断是否存在
        if (StrUtil.isBlank(s)){
            //3.不存在直接返回
            return null;
        }
        //4.命中需要先把JSON反序列化为对象
        RedisData redisData = JSONUtil.toBean(s, RedisData.class);
        JSONObject data = (JSONObject) redisData.getData();
        R r = JSONUtil.toBean(data, typeClass);
        LocalDateTime expireTime = redisData.getExpireTime();
        //5.判断缓存是否过期  这里实际得到的是JsonObject对象
        if (expireTime.isAfter(LocalDateTime.now())){
            //过期时间在当前时间之后 说明未过期
            //5.1 未过期，直接返回店铺信息
            return r;
        }
        //5.2 已过期需要进行缓存重建
        //6. 进行缓存重建
        //6.1 获取互斥锁
        String lockKey = RedisConstants.LOCK_SHOP_KEY+id;
        boolean isLock = tryLock(lockKey);
        //6.2 判断是否获取锁成功
        if (isLock){
            //6.3 成功开启独立线程 实现缓存重建
            try {
                CACHE_REBUILD_EXECUTOR.submit(()->{
                    //查数据库
                    R results = dbFallback.apply(id);
                    //写入Redis
                    this.setLogicalExpire(key,results,time,timeUnit);
                });
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //释放锁
                unlock(lockKey);
            }

        }
        //6.4 返回过期的店铺信息
        return r;
    }

    //自定义线程池
    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

    private boolean tryLock(String key){
        Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(aBoolean);
    }

    private boolean unlock(String key){
        Boolean delete = stringRedisTemplate.delete(key);
        return BooleanUtil.isTrue(delete);
    }

}
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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

8.思维导图总结

缓存更新策略

缓存穿透

缓存雪崩

缓存击穿