[大数据]利用Redis进行数据缓存

1. 引言

缓存有啥用？

1. 降低对数据库的请求，减轻服务器压力
2. 提高了读写效率

缓存有啥缺点？

1. 如何保证数据库与缓存的数据一致性问题？
2. 维护缓存代码
3. 搭建缓存一般是以集群的形式进行搭建，需要运维的成本

2. 将信息添加到缓存的业务流程

上图可以清晰的了解Redis在项目中所处的位置，是数据库与客户端之间的一个中间件，也是数据库的保护伞。有了Redis可以帮助数据库进行请求的阻挡，阻止请求直接打入数据库，提高响应速率，极大的提升了系统的稳定性。

3. 实现代码

下面将根据查询商铺信息来作为背景进行代码书写，具体的流程图如上所示。

3.1 代码实现（信息添加到缓存中）

public static final String SHOPCACHEPREFIX = "cache:shop:";
    
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    // JSON工具
    ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public Result queryById(Long id) {
        //从Redis查询商铺缓存
        String cacheShop = stringRedisTemplate.opsForValue().get(SHOPCACHEPREFIX + id);

        //判断缓存中数据是否存在
        if (!StringUtil.isNullOrEmpty(cacheShop)) {
            //缓存中存在则直接返回
            try {
                // 将子字符串转换为对象
                Shop shop = objectMapper.readValue(cacheShop, Shop.class);
                return Result.ok(shop);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }

        //缓存中不存在，则从数据库里进行数据查询
        Shop shop = getById(id);

        //数据库里不存在，返回404
        if (null==shop){
            return Result.fail("信息不存在");
        }
        //数据库里存在，则将信息写入Redis
        try {
            String shopJSon = objectMapper.writeValueAsString(shop);
          stringRedisTemplate.opsForValue().set(SHOPCACHEPREFIX+id,shopJSon,30,TimeUnit.MINUTES);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        //返回
        return Result.ok(shop);
    }

3.2 缓存更新策略

数据库与缓存数据一致性问题，当数据库信息修改后，缓存的信息应该如何处理？

这里其实是需要根据业务场景来进行选择

1. 高一致性：选主动更新
2. 低一致性：内存淘汰和超时剔除

3.3 实现主动更新

此时需要实现数据库与缓存一致性问题，在这个问题之中还有多个问题值得深思

1. 删除缓存还是更新缓存？
   当数据库发生变化时，我们如何处理缓存中无效的数据，是删除它还是更新它？
   更新缓存：每次更新数据库都更新缓存，无效写操作较多
   删除缓存：更新数据库时删除缓存，查询时再添加缓存
   由此可见，选择删除缓存是高效的。

2. 如何保证缓存与数据库的操作的同时成功或失败？
   单体架构：单体架构中采用事务解决
   分布式架构：利用分布式方案进行解决

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

在并发情况下，上述情况是极大可能会发生的，这样子会导致缓存与数据库数据库不一致。

先操作数据库，在操作缓存这种情况，在缓存数据TTL刚好过期时，出现一个A线程查询缓存，由于缓存中没有数据，则向数据库中查询，在这期间内有另一个B线程进行数据库更新操作和删除缓存操作，当B的操作在A的两个操作间完成时，也会导致数据库与缓存数据不一致问题。

完蛋！！！两种方案都会造成数据库与缓存一致性问题的发生，那么应该如何来进行选择呢？

虽然两者方案都会造成问题的发生，但是概率上来说还是先操作数据库，再删除缓存发生问题的概率低一些，所以可以选择先操作数据库，再删除缓存的方案。

个人见解：
如果说我们在先操作数据库，再删除缓存方案中线程B删除缓存时，我们利用java来删除缓存会有Boolean返回值，如果是false，则说明缓存已经不存在了，缓存不存在了，则会出现上图的情况，那么我们是否可以根据删除缓存的Boolean值来进行判断是否需要线程B来进行缓存的添加（因为之前是需要查询的线程来添加缓存，这里考虑线程B来添加缓存，线程B是操作数据库的缓存），如果线程B的添加也在线程A的写入缓存之前完成也会造成数据库与缓存的一致性问题发生。那么是否可以延时一段时间（例如5s，10s）再进行数据的添加，这样子虽然最终会统一数据库与缓存的一致性，但是若是在这5s，10s内又有线程C，D等等来进行缓存的访问呢？C，D线程的访问还是访问到了无效的缓存信息。
所以在数据库与缓存的一致性问题上，除非在写入正确缓存之前拒绝相关请求进行服务器来进行访问才能避免用户访问到错误信息，但是拒绝请求对用户来说是致命的，极大可能会导致用户直接放弃使用应用，所以我们只能尽可能的减少问题可能性的发生。（个人理解，有问题可以在评论区留言赐教）

  @Override
    @Transactional
    public Result updateShop(Shop shop) {
        Long id = shop.getId();
        if (null==id){
            return Result.fail("店铺id不能为空");
        }
        //更新数据库
        boolean b = updateById(shop);
        //删除缓存
        stringRedisTemplate.delete(SHOPCACHEPREFIX+shop.getId());
        return Result.ok();
    }

4. 缓存穿透

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

解决方案：

缓存空对象

缺点：

1. 空间浪费
2. 如果缓存了空对象，在空对象的有效期内，我们后台在数据库新增了和空对象相同id的数据，这样子就会造成数据库与缓存一致性问题

布隆过滤器

优点：

1. 内存占用少

缺点：

1. 实现复杂
2. 存在误判的可能（存在的数据一定会判断成功，但是不存在的数据也有可能会放行进来，有几率造成缓存穿透）

4.1 解决缓存穿透（使用空对象进行解决）

public static final String SHOPCACHEPREFIX = "cache:shop:";

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    // JSON工具
    ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public Result queryById(Long id) {
        //从Redis查询商铺缓存
        String cacheShop = stringRedisTemplate.opsForValue().get(SHOPCACHEPREFIX + id);

        //判断缓存中数据是否存在
        if (!StringUtil.isNullOrEmpty(cacheShop)) {
            //缓存中存在则直接返回
            try {
                // 将子字符串转换为对象
                Shop shop = objectMapper.readValue(cacheShop, Shop.class);
                return Result.ok(shop);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }

        // 因为上面判断了cacheShop是否为空，如果进到这个方法里面则一定是空，直接过滤，不打到数据库
        if (null != cacheShop){
            return Result.fail("信息不存在");
        }

        //缓存中不存在，则从数据库里进行数据查询
        Shop shop = getById(id);

        //数据库里不存在，返回404
        if (null==shop){
            // 缓存空对象
            stringRedisTemplate.opsForValue().set(SHOPCACHEPREFIX+id,"",2,TimeUnit.MINUTES);
            return Result.fail("信息不存在");
        }
        //数据库里存在，则将信息写入Redis
        try {
            String shopJSon = objectMapper.writeValueAsString(shop);
            stringRedisTemplate.opsForValue().set(SHOPCACHEPREFIX+id,shopJSon,30,TimeUnit.MINUTES);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        //返回
        return Result.ok(shop);
    }

上述方案终究是被动方案，我们可以采取一些主动方案，例如

1. 给id加复杂度
2. 权限
3. 热点参数的限流

5. 缓存雪崩

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

解决方案：

1. 给不同的Key的TTL添加随机值
   大量的Key同时失效，极大可能是TTL相同，我们可以随机给TTL
2. 利用Redis集群提高服务的可用性
3. 给缓存业务添加降级限流策略
4. 给业务添加多级缓存

6. 缓存击穿

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

常见的解决方案：

1. 互斥锁
2. 逻辑过期

互斥锁：

即采用锁的方式来保证只有一个线程去重建缓存数据，其余拿不到锁的线程休眠一段时间再重新重头去执行查询缓存的步骤

优点：

1. 没有额外的内存消耗（针对下面的逻辑过期方案）
2. 保证了一致性

缺点：

1. 线程需要等待，性能受到了影响
2. 可能会产生死锁

逻辑过期：

逻辑过期是在缓存数据中额外添加一个属性，这个属性就是逻辑过期的属性，为什么要使用这个来判断是否过期而不使用TTL呢？因为使用TTL的话，一旦过期，就获取不到缓存中的数据了，没有拿到锁的线程就没有旧的数据可以返回。

它与互斥锁最大的区别就是没有线程的等待了，谁先获取到锁就去重建缓存，其余线程没有获取到锁就返回旧数据，不去做休眠，轮询去获取锁。

重建缓存会新开一个线程去执行重建缓存，目的是减少抢到锁的线程的响应时间。

优点：

1. 线程无需等待，性能好

缺点：

1. 不能保证一致性
2. 缓存中有额外的内存消耗
3. 实现复杂

两个方案各有优缺点：一个保证了一致性，一个保证了可用性，选择与否主要看业务的需求是什么，侧重于可用性还是一致性。

6.1 互斥锁代码

互斥锁的锁用什么？

使用Redis命令的setnx命令。

首先实现获取锁和释放锁的代码

    /**
     * 尝试获取锁
     *
     * @param key
     * @return
     */
    private boolean tryLock(String key) {
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(flag);
    }

    /**
     * 删除锁
     *
     * @param key
     */
    private void unLock(String key) {
        stringRedisTemplate.delete(key);
    }

代码实现

public Shop queryWithMutex(Long id) throws InterruptedException {
        //从Redis查询商铺缓存
        String cacheShop = stringRedisTemplate.opsForValue().get(SHOPCACHEPREFIX + id);

        //判断缓存中数据是否存在
        if (!StringUtil.isNullOrEmpty(cacheShop)) {
            //缓存中存在则直接返回
            try {
                // 将子字符串转换为对象
                Shop shop = objectMapper.readValue(cacheShop, Shop.class);
                return shop;
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }

        // 因为上面判断了cacheShop是否为空，如果进到这个方法里面则一定是空，直接过滤，不打到数据库
        if (null != cacheShop) {
            return null;
        }

        Shop shop = new Shop();
        // 缓存击穿，获取锁
        String lockKey = "lock:shop:" + id;
        try{
            boolean b = tryLock(lockKey);
            if (!b) {
                // 获取锁失败了
                Thread.sleep(50);
                return queryWithMutex(id);
            }
            //缓存中不存在，则从数据库里进行数据查询
           shop = getById(id);

            //数据库里不存在，返回404
            if (null == shop) {
                // 缓存空对象
                stringRedisTemplate.opsForValue().set(SHOPCACHEPREFIX + id, "", 2, TimeUnit.MINUTES);
                return null;
            }
            //数据库里存在，则将信息写入Redis
            try {
                String shopJSon = objectMapper.writeValueAsString(shop);
                stringRedisTemplate.opsForValue().set(SHOPCACHEPREFIX + id, shopJSon, 30, TimeUnit.MINUTES);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }catch (Exception e){

        }finally {
            // 释放互斥锁
            unLock(lockKey);
        }

        //返回
        return shop;

    }

6.2 逻辑过期实现

逻辑过期不设置TTL

代码实现

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

由于是热点key，所以key基本都是手动导入到缓存，代码如下

  /**
     * 逻辑过期时间对象写入缓存
     * @param id
     * @param expireSeconds
     */
    public void saveShopToRedis(Long id,Long expireSeconds){
        // 查询店铺数据
        Shop shop = getById(id);
        // 封装为逻辑过期
        RedisData redisData = new RedisData();
        redisData.setData(shop);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));
        // 写入Redis
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id, JSONUtil.toJsonStr(redisData));
    }

逻辑过期代码实现

/**
     * 缓存击穿：逻辑过期解决
     * @param id
     * @return
     * @throws InterruptedException
     */
    public Shop queryWithPassLogicalExpire(Long id) throws InterruptedException {
        //1. 从Redis查询商铺缓存
        String cacheShop = stringRedisTemplate.opsForValue().get(SHOPCACHEPREFIX + id);

        //2. 判断缓存中数据是否存在
        if (StringUtil.isNullOrEmpty(cacheShop)) {
            // 3. 不存在
            return null;
        }
        // 4. 存在，判断是否过期
        RedisData redisData = JSONUtil.toBean(cacheShop, RedisData.class);
        JSONObject jsonObject = (JSONObject) redisData.getData();
        Shop shop = JSONUtil.toBean(jsonObject, Shop.class);
        LocalDateTime expireTime = redisData.getExpireTime();

        // 5. 判断是否过期
        if (expireTime.isAfter(LocalDateTime.now())){
            // 5.1 未过期
            return shop;
        }
        // 5.2 已过期
        String lockKey = "lock:shop:"+id;
        boolean flag = tryLock(lockKey);
        if (flag){
            // TODO 获取锁成功，开启独立线程，实现缓存重建，建议使用线程池去做
            CACHE_REBUILD_EXECUTOR.submit(()->{
                try {
                    // 重建缓存
                    this.saveShopToRedis(id,1800L);
                }catch (Exception e){
                    
                }finally {
                    // 释放锁
                    unLock(lockKey);
                }
             
            });

        }
        // 获取锁失败，返回过期的信息
        return shop;
    }

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

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖