案例解决Redis高并发场景带来的缓存穿透、击穿、雪崩问题（超级详细！！）

假设你的网站流量量达到亿级，传统的去查询DB势必会给DB带来巨大的压力，甚至可能有宕机的风险，接下来我就分几个阶段，来讲诉各个场景可能会给DB带来巨大压力的可能，以及优化的方案。

缓存击穿：key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
缓存穿透：key对应的数据在数据源并不存在，每次针对此key的请求从缓存获取不到，请求都会到数据源，从而可能压垮数据源。比如用一个不存在的用户id获取用户信息，不论缓存还是数据库都没有，若黑客利用此漏洞进行攻击可能压垮数据库。
缓存雪崩：当缓存服务器重启或者大量缓存集中在某一个时间段失效，这样在失效的时候，也会给后端系统(比如DB)带来很大压力。

场景导入

* 最初的数据Redis的数据查询方法
* 假设是查询product 产品的信息
* 场景：当商品的数据有几亿的时候，直接请求到数据库，会造成数据库的压力过大，而产生宕机风险


    private final static String PRODUCT_NAME_KEY = "product:name:key:";

    public PatrolTask one(Long productId) {
        PatrolTask patrolTask;
        String productCacheKey = PRODUCT_NAME_KEY + productId;
        //去缓存中找到对应的产品数据，并取出
        String productStr = redisUtil.get(productCacheKey);

        if (C.isNotEmpty(productStr)) {
            patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
            return patrolTask;
        }

        PatrolTask dbData = iPatrolTaskService.getById(productId);
        if (C.isNotEmpty(dbData)) {
            //如果去数据库查找到的商品是不空的情况下，把数据再次放入到缓存中
            redisUtil.set(productCacheKey, JSON.toJSON(dbData));
        }
        return dbData;
    }

以上代码可能会导致的问题：

* 上面的会造成的问题，如果把几亿的数据量都放入的缓存中

* 但是常用的就那么几个商品，就会占用大量的内存资源，消耗内存

第二次改造

给放入的缓存加入一定的过期时间，这样不常用的数据到期就会自动删除

    public PatrolTask two(Long productId) {
        PatrolTask patrolTask;
        String productCacheKey = PRODUCT_NAME_KEY + productId;
        String productStr = redisUtil.get(productCacheKey);

        if (C.isNotEmpty(productStr)) {
            patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
            return patrolTask;
        }

        PatrolTask dbData = iPatrolTaskService.getById(productId);
        if (C.isNotEmpty(dbData)) {
            //放入缓存中时，加入过期时间，防止数量过大
            redisUtil.set(productCacheKey, JSON.toJSON(dbData), PRODUCT_TIME);
        }
        return dbData;
    }

以上代码可能会导致的问题：

* 以上代码存在问题：如果说我批量导入10000个产品，在缓存中的商品会同一时间都过期

* 而这个时间又有大量的请求打入来查询商品就会造成了缓存雪崩现象

缓存雪崩

当缓存服务器重启或者大量缓存集中在某一个时间段失效，这样在失效的时候，也会给后端系统(比如DB)带来很大压力

第三次改造

给过期时间设置随机值，不让他们同一时间失效，或者是热点缓存永不过期

    public PatrolTask third(Long productId) {
        PatrolTask patrolTask;
        String productCacheKey = PRODUCT_NAME_KEY + productId;
        String productStr = redisUtil.get(productCacheKey);

        if (C.isNotEmpty(productStr)) {
            patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
            return patrolTask;
        }

        PatrolTask dbData = iPatrolTaskService.getById(productId);
        if (C.isNotEmpty(dbData)) {
            //设置缓存时间的随机
            redisUtil.set(productCacheKey, JSON.toJSON(dbData), getProductCacheTime());
        }
        return dbData;
    }


    //设置缓存时间的随机值
    private Long getProductCacheTime() {
        return PRODUCT_TIME + new Random().nextInt(30) * 60;

    }

以上代码可能会导致的问题：

* 以上代码存在问题：如果运营人员现在把热点的商品进行了删除，但是这个时间又有大量的人来访问这个商品。
* 而在缓存中都查不到，就回去查数据库，就会给数据库造成巨大压力，这种现象就叫做缓存穿透。

缓存穿透

key对应的数据在数据源并不存在，每次针对此key的请求从缓存获取不到，请求都会到数据源，从而可能压垮数据源。比如用一个不存在的用户id获取用户信息，不论缓存还是数据库都没有，若黑客利用此漏洞进行攻击可能压垮数据库

第四次改造

解决的办法，就是如果说数据库查不到，我们就返回一个空对象，存入缓存中，防止大量数据库直接打入数据库

    public PatrolTask four(Long productId) {
        PatrolTask patrolTask;
        String productCacheKey = PRODUCT_NAME_KEY + productId;
        String productStr = redisUtil.get(productCacheKey);

        if (C.isNotEmpty(productStr)) {
            //如果查到的值为空，就直接进行返回一个空的对象
            if (PRODUCT_EMPTY.equals(productStr)) {
                return new PatrolTask();
            }
            patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
            return patrolTask;
        }

        PatrolTask dbData = iPatrolTaskService.getById(productId);
        if (C.isNotEmpty(dbData)) {

            redisUtil.set(productCacheKey, JSON.toJSON(dbData), getProductCacheTime());
        } else {
            //缓存一个空的值
            redisUtil.set(productCacheKey, PRODUCT_EMPTY, getProductCacheTime());

        }

        return dbData;
    }

以上代码可能会存在的问题：

* 以上代码存在问题：如果说之前在缓存中没有这个数据，那么之前数据就有可能没有在缓存中
* 但是突然这个数据并发了大量的并发，就有可能大量的同一时间去创建这个缓存数据，就会给系统造成压力暴增问题，就会出现缓存击穿现象

缓存击穿

key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮

第五次改造

我在查询缓存的时间，加一把锁synchronized (this) ，这样只会有一个线程进来查询，利用单利模式的双重检测锁的实现来优化，并放入缓存中，不会有大量的数据进来

    public PatrolTask five(Long productId) {
        PatrolTask patrolTask;
        String productCacheKey = PRODUCT_NAME_KEY + productId;
        String productStr = redisUtil.get(productCacheKey);

        if (C.isNotEmpty(productStr)) {
            //如果查到的值为空，就直接进行返回一个空的对象
            if (PRODUCT_EMPTY.equals(productStr)) {
                return new PatrolTask();
            }
            patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
            return patrolTask;
        }

        //加锁，只保证有一个线程能够进来
        synchronized (this)  {
             // 这里再去查一次缓存，相当于第二个请求，就会查缓存，用到了单利设计模式
            if (C.isNotEmpty(productStr)) {
                if (PRODUCT_EMPTY.equals(productStr)) {
                    return new PatrolTask();
                }
                patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
                return patrolTask;
            }

            PatrolTask dbData = iPatrolTaskService.getById(productId);
            if (C.isNotEmpty(dbData)) {

                redisUtil.set(productCacheKey, JSON.toJSON(dbData), getProductCacheTime());
            } else {
               
                redisUtil.set(productCacheKey, PRODUCT_EMPTY, getProductCacheTime());

            }
            return dbData;

        }


    }

以上代码可能会存在的问题：

synchronized锁事非常重的锁，分布式环境情况下，每个系统都会去重建一次，会造成一定的性能开销，目前这个的性能开销并不大，但是（this）锁的范围非常大！！

会把所有的商品的查询都给锁住，会造成巨大的性能开销。

所以上锁的时候，一定要去考虑锁的范围，范围越小越好

第五次改造

采用分布式锁Redisson来解决以上的问题，锁的对象能加锁的到产品的对象，这样范围就会很小。

    @Autowired
    private Redisson redisson;

    private final static String product_lock = "product:lock";


    public PatrolTask six(Long productId) {
        PatrolTask patrolTask;
        String productCacheKey = PRODUCT_NAME_KEY + productId;
        String productStr = redisUtil.get(productCacheKey);

        if (C.isNotEmpty(productStr)) {
            //如果查到的值为空，就直接进行返回一个空的对象
            if (PRODUCT_EMPTY.equals(productStr)) {
                return new PatrolTask();
            }
            patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
            return patrolTask;
        }

        RLock lock = redisson.getLock(product_lock + productId);
        //加锁
        lock.lock(20, TimeUnit.MINUTES);
        try {

            if (C.isNotEmpty(productStr)) {
                if (PRODUCT_EMPTY.equals(productStr)) {
                    return new PatrolTask();
                }
                patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
                return patrolTask;
            }

            patrolTask = iPatrolTaskService.getById(productId);
            if (C.isNotEmpty(patrolTask)) {

                redisUtil.set(productCacheKey, JSON.toJSON(patrolTask), getProductCacheTime());
            } else {

                redisUtil.set(productCacheKey, PRODUCT_EMPTY, getProductCacheTime());
            }

        } finally {
            //释放锁
            lock.unlock();

        }

        return patrolTask;

    }

以上代码可能会存在的问题：

如果在我们查到数据的时间，整个网络有延时的情况，就在这个时候，有人更新人这个产品，并已存入数据库，而我们拿到的是旧的信息，并放入了缓存中，那么下次来查询的时候，查询缓存发现有数据，就会直接返回，只有等缓存过期了才会更新缓存。这就是典型的缓存与数据库不一致问题

第五次改造

常用的方案：延迟双删、加（Redisson读写锁）以下提供Redisson的读写锁的操作，下面我讲解的是读锁的代码，写的代码，同理在update更新操作的时候，你锁同一个对象，并readWriteLock.writeLock() 获取到读的锁就可以，读写锁不清楚的可自行了解下

    @Autowired
    private Redisson redisson;

    private final static String product_lock = "product:lock";


    public PatrolTask six(Long productId) {
        PatrolTask patrolTask;
        String productCacheKey = PRODUCT_NAME_KEY + productId;
        String productStr = redisUtil.get(productCacheKey);

        if (C.isNotEmpty(productStr)) {
            //如果查到的值为空，就直接进行返回一个空的对象
            if (PRODUCT_EMPTY.equals(productStr)) {
                return new PatrolTask();
            }
            patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
            return patrolTask;
        }

        RLock lock = redisson.getLock(product_lock + productId);
        //加锁
        lock.lock(20, TimeUnit.MINUTES);
        try {

            if (C.isNotEmpty(productStr)) {
                if (PRODUCT_EMPTY.equals(productStr)) {
                    return new PatrolTask();
                }
                patrolTask = JSONObject.parseObject(productStr, PatrolTask.class);
                return patrolTask;
            }

            RReadWriteLock readWriteLock = redisson.getReadWriteLock(product_lock + productId);
            RLock rLock = readWriteLock.readLock();
            try {
                patrolTask = iPatrolTaskService.getById(productId);
                if (C.isNotEmpty(patrolTask)) {

                    redisUtil.set(productCacheKey, JSON.toJSON(patrolTask), getProductCacheTime());
                } else {

                    redisUtil.set(productCacheKey, PRODUCT_EMPTY, getProductCacheTime());
                }
                
            }   finally {
                rLock.unlock();
            }
         
        } finally {
            //释放锁
            lock.unlock();

        }
        return patrolTask;

    }