Redis实现缓存及解决缓存产生的问题

Posted on 2022-07-22 Edited on 2023-08-16 In Redis 缓存问题 Views: Waline: Word count in article: 11k

前言

本文将以查询商铺信息业务为主线，使用Redis做缓存逐步提升查询商铺的效率。

同时提供使用Redis缓存过程中遇到的数据一致性、缓存击穿、缓存穿透、缓存雪崩等问题的解决思路。

添加商户缓存

在查询商户业务中，如果不使用缓存，而是直接像以下代码那样去查询数据库，效率是非常慢的！因为查询数据库操作的是磁盘，而操作磁盘需要大量的IO操作，从而导致查询效率非常慢！

@GetMapping("/{id}")
public Result queryShopById(@PathVariable("id") Long id) {
    //这里是直接查询数据库
    return shopService.queryById(id);
}

我们使用Postman测试一下效率

可以发现，在数据量比较少的情况下，还是需要280ms

下面我们使用redis做缓存优化一下

 /**
  * 使用redis数据类型做缓存
  * @return
  */
@Autowired
private StringRedisTemplate stringRedisTemplate;

@Override
public Result queryById(Long id) {
     // 查看redis中是否存在对应数据
     String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);

     Shop shop = new Shop();
    // 判断缓存中有没有店铺信息
     if (StrUtil.isNotBlank(shopJson)){
         // 存在 返回
         shop = JSONUtil.toBean(shopJson, Shop.class)
         return Result.ok(shop);
     }
     // 不存在 从数据库中查找
     shop =  getById(id);
     if (shop == null){
         // 查不到对应的数据 返回
         return Result.fail("店铺不存在！");
     }
     // 将对应的数据添加到redis中
     redisTemplate.opsForValue().set(CACHE_SHOP_KEY + id,JSONUtil.toJsonStr(shop));
     // 查到对应的数据 返回
     return Result.ok(shop);
 }

使用Postman测试一下效率

可以发现，查询用时为16ms，效率提升了十几倍！

下面是使用Redis做缓存的具体流程

Cache_Model

以下是Redis实现添加商户缓存的业务逻辑图

add_shop_cache

小结：

使用Redis做商铺缓存可以提升我们的查询效率，但随之也产生了一些问题：我们缓存的数据是存放在内存中的，而内存并不像JVM那样，有垃圾回收机制，如果不加以措施，我们内存很容易爆掉！同时，因为我们的数据库数据是会发生改变的，而缓存的数据来自于数据库，我们该怎么样保证数据的一致性呢？

缓存更新策略

其实，在Redis中，提供了一个缓存更新策略（其实就是缓存淘汰机制）；使用这个机制，可以很好的解决我们上一小节的问题。

下面是三种缓存更新策略

业务场景：

低一致性要求：使用内存淘汰机制。例如店铺类型的查询缓存
高一致性要求：主动更新，并以超时剔除作为兜底方案。例如店铺详情查询的缓存

现在回到我们的业务，查询商铺信息，也就是高一致性要求的场景下，我们可以使用主动更新，并以超时剔除作为兜底方案解决上面的问题。

而主动更新的话是有两种方案的：

先删除缓存、再更新数据库
先更新数据库、再删除缓存

上述两种方案看起来没什么区别，但是高并发场景下差别可是很大的，我们一般选择后者。原因如下：

场景：缓存和数据库中的字段A=10，现在线程1要将A修改成20，同时，线程2想查询这个字段

如果你选择第一种方案，在两个线程并发来访问时，线程1先来，他要更新的是A=20，他先把缓存中的字段A删了，此时线程2过来，他查询缓存时发现字段A并不存在，他就只能去数据库查了，同时他将数据库查询的数据A=10写入缓存，当他写入缓存后，线程1再执行更新动作时，实际上写入数据库的数据A=20，与缓存中的不一致，导致后面其他线程再来查询A的时候，值都是缓存中的10；

通过上述分析，下面我们修改代码实现我们的查询商铺信息业务中数据在数据库和缓存的双写一致

第一个要修改的地方，每次修改店铺时，我们需要先更新数据库，再删除缓存

@Override
public Result updateShopById(Shop shop) {
    // 根据id修改店铺
    if (shop.getId() == null) {
        return Result.fail("店铺id为空, 修改失败");
    }
    // 先更新数据库
    updateById(shop);
    // 修改完成后删除缓存
    stringRedisTemplate.delete(CACHE_SHOP_KEY + shop.getId());
    return Result.ok();
}

第二个要修改的地方，我们在添加redis缓存时，应该给我们缓存数据设置有效期

 /**
  * 使用redis数据类型做缓存
  * @return
  */
@Autowired
private StringRedisTemplate stringRedisTemplate;

@Override
public Result queryById(Long id) {
     // 查看redis中是否存在对应数据
     String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);

     Shop shop = new Shop();
    // 判断缓存中有没有店铺信息
     if (StrUtil.isNotBlank(shopJson)){
         // 存在 返回
         shop = JSONUtil.toBean(shopJson, Shop.class)
         return Result.ok(shop);
     }
     // 不存在 从数据库中查找
     shop =  getById(id);
     if (shop == null){
         // 查不到对应的数据 返回
         return Result.fail("店铺不存在！");
     }
     // 将对应的数据添加到redis中 并且设置缓存过期时间
     redisTemplate.opsForValue().set(CACHE_SHOP_KEY + id,JSONUtil.toJsonStr(shop), 30L, TimeUnit.MINUTES);
     // 查到对应的数据 返回
     return Result.ok(shop);
 }

小结：

目前，我们通过缓存更新策略，暂且解决了我们的缓存导致内存爆满以及数据一致性的要求。

但是，由于我们的缓存数据是有一定的有效期的，同时也会产生以下三个redis缓存中常见的问题：

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

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

缓存击穿：缓存击穿问题也叫热点Key问题，就是一个被高并发访问并且缓存重建业务较复杂的key突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。

下面三个小节将逐步分析这几个问题以及解决思路

缓存穿透

在我们当前查询商户信息业务下可能会发生缓存穿透的场景：

比方说现在有一个用户想要查询id=10的商户，按照我们上面的逻辑，会先去redis缓存中查询是否存在，如果redis不存在，则去数据库中查询，但是由于用户查询的是不存在的店铺，我们并不会把后面从数据库中查询到的不存在店铺的信息保存到redis中，而是直接返回店铺不存在信息给用户。

如果这个时候，这个用户不安好心，疯狂的往我们的系统中查询这个店铺的信息，那么这些请求会全部打到我们的数据库中，从而给我们数据库造成巨大的压力。

上述场景中，大量的请求穿过了我们的redis缓存，到达数据库，这样是非常危险的，这也就是所谓的缓存穿透了。

缓存穿透的解决方案有两种：

当查询到数据不存在时，我们将null数据存储到缓存当中去，从而避免大量的请求到达数据库
使用布隆过滤器

方案一设置null值解决思路：

Cache_penetration_1

如上图所示，我们只需要在查询到null数据的时候，往我们的缓存中设置null值，即可解决缓存穿透问题。

示例代码：

public Shop queryWithPassThrough(Long id) {
    // 查看redis中是否存在对应数据 shopJson的可能取值 1.shop 2.null 3.{}
    log.info("缓存穿透解决方案~ ");
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    log.info("shopJson =  {}", shopJson);
    if (StrUtil.isNotBlank(shopJson)) {
        // 存在 返回
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return shop;
    }
    // 如果redis中查到的json无效 并且不为null 说明店铺不存在
    if (shopJson != null) {
        return null;
    }
    // 不存在 从数据库中查找
    Shop shop = getById(id);
    if (shop == null) {
        // 在redis中添加一个有效期为2min的空数据 避免缓存穿透
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
        // 查不到对应的数据 返回
        return null;
    }
    // 将对应的数据添加到redis中 并且设置超时时间 做缓存剔除
    stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);
    // 查到对应的数据 返回
    return shop;
}

虽然我们使实现起来非常简单，但是如果用户多次查询到不同id的null值，那我们缓存中也会造成额外的内存消耗！

方案二布隆过滤器解决思路：

布隆过滤器其实本质就是一个二进制的数组，然后我们像布隆过滤器中存值时，会通过指定的hash算法来分别算出这个值在我们的二进制数组当中对应的下标位置，然后把这个下标位置对应的值改为1（数组初始化默认为0）；当我们要查询这个值是否存在时，我们只需要判断我们这个值对应的数组下标位置的值是否为全为1即可。

也就是说，布隆过滤器判定某个值不存在，则必定不存在；判定某个值存在，则大概率存在；

布隆过滤器解决缓存穿透的实现思路其实就是在我们的用户与redis之间增加一层过滤器，然后我们存放商铺数据的时候，需要把我们的商铺id信息存储到布隆过滤器中；当客户要查询商铺时，我们可以先通过布隆过滤器检查这个id是否存在，如果不存在直接返回null，这样就很好的解决了方案一中浪费内存的问题。但是布隆过滤器可能会存在误判的问题，以及要删除布隆过滤器中的数据也是非常困难的。

实现代码

/**
 * 缓存穿透 - 布隆过滤器解决
 * @param id
 * @return
 */
public Shop queryWithPassThrough(Long id) {
    // 查看redis中是否存在对应数据 shopJson的可能取值 1.shop 2.null 3.{}
    log.info("缓存穿透解决方案~ ");
    // 创建Redisson提供的布隆过滤器
    RBloomFilter<Object> bloomFilter = redissonConfig.redissonClient().getBloomFilter("shopCacheBloomFilter");
    // 初始化布隆过滤器
    bloomFilter.tryInit(10000,0.01);
    // 模拟向布隆过滤器添加商铺id数据 这一步通常在添加商铺时完成
    for (Long i = 0L; i < 10; i++) {
        bloomFilter.add(i);
    }
    // 检测布隆过滤器, 如果id不存在过滤器中, 直接返回null
    if (!bloomFilter.contains(id)) {
        return null;
    }
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    log.info("shopJson =  {}", shopJson);
    if (StrUtil.isNotBlank(shopJson)) {
        // 存在 返回
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return shop;
    }
    // 不存在 从数据库中查找
    Shop shop = getById(id);
    if (shop == null) {
        // 在redis中添加一个有效期为2min的空数据 避免缓存穿透
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
        // 查不到对应的数据 返回
        return null;
    }
    // 将对应的数据添加到redis中 并且设置超时时间 做缓存剔除
    stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);

    // 查到对应的数据 返回
    return shop;
}

缓存击穿

缓存击穿也叫热点key问题，就是一个被高并发访问key突然失效了，我们就需要去重建这个缓存，如果这个key的缓存重建业务比较复杂，那么就会造成大量的请求打到数据库上，给数据库造成巨大的压力。这也就是我们经常所说的热点key问题。

如上图所示，其实就是因为我们需要给key设定有效时间，而我们重构这个key的缓存的业务需要花费很长的时间，导致大量的请求都打到了我们的数据库中，给数据库造成了巨大的压力，这就是缓存击穿。

那么，针对上述产生缓存击穿的原因，目前主流的解决方案有两种：

将key的有效期设置成逻辑有效期
在重建缓存的时候，使用互斥锁，保证只有一个线程参与缓存的重建

两种方案的实现思路

互斥锁方案

在互斥锁解决缓存击穿的方案中，当我们需要进行重建缓存之前，我们需要先去获取互斥锁，只有拿到互斥锁的线程，才有资格去重建缓存。

而没有抢到互斥锁的资源，则需要休眠一段时间，之后再自旋去尝试获取互斥锁。当然，每一次自旋都需要判断需要的获取的缓存是否已经重建了，如果重建了则返回即可。
逻辑过期

在逻辑过期方案中，我们可以在value中多设置一个字段，这个字段存放的使我们当前key的过期时间，当我们命中缓存的时候，需要先去判断我们命中的缓存是否还是有效的，如果有效就返回缓存的数据；

如果当前时间在我们设置的逻辑过期时间之后，我们就需要去重建缓存，而重建缓存的之前，需要去竞争互斥锁资源，只有竞争到互斥锁资源的线程，才能去重建缓存，同时，我们可以使用并发编程中的异步方式去重建缓存，即异步开启一条线程去重建缓存。

没有竞争到互斥锁的线程则返回过期的数据（脏数据）。

两种方案的对比：

互斥锁方案：由于保证了互斥性，所以数据的一致性不会有问题，同时我们实现起来也很简单，只需要加锁即可。但是由于重建的过程中其他线程会处于阻塞，导致性能变差，并且可能会发生死锁。
逻辑过期方案：通过逻辑过期，保证了线程在读取缓存的过程中，不需要阻塞了。同时，重建缓存的过程只需要一个线程去异步进行，其他线程可以正常读取数据。但是重建缓存完成之前返回的数据都是脏数据，会有数据不一致的问题产生，并且实现起来比较难

两种方案的代码实现：

首先我们可以通过Redis的SETNX命令来实现互斥锁

这里也可以直接使用Redisson的互斥锁

/**
 * 获取互斥锁
 *
 * @param key* @return boolean
 */
private boolean tryLock(String key) {
    // 生成互斥锁 有效期大约为具体业务的十倍
    boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10L, TimeUnit.SECONDS);
    //  log.info("获取了互斥锁 {}", key);
    return BooleanUtil.isTrue(lock);
}

/**
 * 释放互斥锁
 *
 * @param key
 * @return void
 */
private void unlock(String key) {
    log.info("释放了互斥锁 {}", key);
    stringRedisTemplate.delete(key);
}

互斥锁解决缓存击穿的代码实现

/**
 * 缓存击穿 互斥锁解决方法
 *
 * @param id
 * @return com.heng.entity.Shop
 */
public Shop queryWithMutex(Long id) {
    // 查看redis中是否存在对应数据 shopJson的可能取值 1.shop 2.null 3.{}
    //log.info("缓存击穿 互斥锁解决方案~ ");
    String key = CACHE_SHOP_KEY + id;
    String shopJson = stringRedisTemplate.opsForValue().get(key);
    //log.info("shopJson =  {}", shopJson);
    if (StrUtil.isNotBlank(shopJson)) {
        // 存在 返回
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return shop;
    }
    // 如果redis中查到的json无效 并且不为null 说明店铺不存在
    if (shopJson != null) {
        return null;
    }
    // 不存在 竞争互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    boolean isLock = tryLock(lockKey);
    Shop shop = null;
    try {
        if (!isLock) {
            // 竞争不到互斥锁 休眠
            Thread.sleep(10);
            return queryWithMutex(id);
        }
        // 竞争到锁资源 判断redis是否已经重建了 避免高并发下 反复重构
        String jsonShop = stringRedisTemplate.opsForValue().get(key);
        // DoubleCheck
        if (StrUtil.isNotBlank(jsonShop)) {
            Shop cacheShop = JSONUtil.toBean(jsonShop, Shop.class);
            return cacheShop;
        }
        // 1. 从数据库中查询
        shop = getById(id);
        Thread.sleep(200);
        if (shop == null) {
            // 2. 找不到对应的数据 防止缓存穿透 在redis中构建一个空的缓存
            stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
            return null;
        }
        // 3. 重构redis缓存
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    } finally {
        // 释放互斥锁
        unlock(lockKey);
    }
    return shop;
}

逻辑过期解决缓存击穿的代码实现

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


/**
 * 缓存击穿 逻辑过期解决方案
 *
 * @param id
 * @return com.hmdp.entity.Shop
 */

public Shop queryWithLogicalExpire(Long id) {
    // 从redis缓存中去热点数据
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    // 如果没命中 返回null (不命中说明没有设置成热点数据 所以直接返回null)
    if (StrUtil.isBlank(shopJson)) {
        // 存在 返回
        return null;
    }
    // 如果命中 判断逻辑有效时间是否过期
    RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);

    // 因为RedisData类中的data可以是其他需要缓存的热点数据 所以是使用Object去接的 我们可以将它强转成JSONObject类型 以便于获取数据的实际实体类
    JSONObject jsonObject = (JSONObject) redisData.getData();
    Shop shop = JSONUtil.toBean(jsonObject, Shop.class);

    // 如果逻辑有效时间过期, 重键缓存
    LocalDateTime expireTime = redisData.getExpireTime();
    if (expireTime.isAfter(LocalDateTime.now())) {
        return shop;
    }
    // 过期 重建缓存
    // 1. 获取互斥锁
    boolean isLock = tryLock(LOCK_SHOP_KEY + id);
    if (isLock) {
        // 2. 先判断缓存是否已经被重建了
        String shopJson1 = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
        RedisData redisData1 = JSONUtil.toBean(shopJson1, RedisData.class);
        if (redisData1.getExpireTime().isAfter(LocalDateTime.now())) {
            JSONObject jsonObject1 = (JSONObject) redisData1.getData();
            shop = JSONUtil.toBean(jsonObject1, Shop.class);
            // 返回重建后的商铺缓存信息
            return shop;
        }
        // 3. 开启另外一条线程进行缓存重建
        CACHE_REBUILD_EXECUTOR.submit(() -> {
            try {
                // 重建缓存
                this.saveRedisData(id, 10L);
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                // 释放互斥锁
                unlock(LOCK_SHOP_KEY + id);
            }
        });

    }
    // 4. 返回过期的商铺信息
    return shop;
}